|
|
@@ -3,7 +3,10 @@ import time
|
|
|
import os
|
|
|
import fnmatch
|
|
|
import ap_git
|
|
|
+import json
|
|
|
+import jsonschema
|
|
|
from . import exceptions as ex
|
|
|
+from threading import Lock
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@@ -134,3 +137,233 @@ class APSourceMetadataFetcher:
|
|
|
@staticmethod
|
|
|
def get_singleton():
|
|
|
return APSourceMetadataFetcher.__singleton
|
|
|
+
|
|
|
+
|
|
|
+class VersionInfo:
|
|
|
+ """
|
|
|
+ Class to wrap version info properties inside a single object
|
|
|
+ """
|
|
|
+ def __init__(self,
|
|
|
+ remote: str,
|
|
|
+ commit_ref: str,
|
|
|
+ release_type: str,
|
|
|
+ version_number: str,
|
|
|
+ ap_build_artifacts_url) -> None:
|
|
|
+ self.remote = remote
|
|
|
+ self.commit_ref = commit_ref
|
|
|
+ self.release_type = release_type
|
|
|
+ self.version_number = version_number
|
|
|
+ self.ap_build_artifacts_url = ap_build_artifacts_url
|
|
|
+
|
|
|
+
|
|
|
+class RemoteInfo:
|
|
|
+ """
|
|
|
+ Class to wrap remote info properties inside a single object
|
|
|
+ """
|
|
|
+ def __init__(self,
|
|
|
+ name: str,
|
|
|
+ url: str) -> None:
|
|
|
+ self.name = name
|
|
|
+ self.url = url
|
|
|
+
|
|
|
+
|
|
|
+class VersionsFetcher:
|
|
|
+ """
|
|
|
+ Class to fetch the version-to-build metadata from remotes.json
|
|
|
+ and provide methods to view the same
|
|
|
+ """
|
|
|
+
|
|
|
+ __singleton = None
|
|
|
+
|
|
|
+ def __init__(self, remotes_json_path: str):
|
|
|
+ """
|
|
|
+ Initializes the VersionsFetcher instance
|
|
|
+ with a given remotes.json path.
|
|
|
+
|
|
|
+ Parameters:
|
|
|
+ remotes_json_path (str): Path to the remotes.json file.
|
|
|
+
|
|
|
+ Raises:
|
|
|
+ TooManyInstancesError: If an instance of this class already exists,
|
|
|
+ enforcing a singleton pattern.
|
|
|
+ """
|
|
|
+ # Enforce singleton pattern by raising an error if
|
|
|
+ # an instance already exists.
|
|
|
+ if VersionsFetcher.__singleton:
|
|
|
+ raise ex.TooManyInstancesError()
|
|
|
+
|
|
|
+ self.__remotes_json_path = remotes_json_path
|
|
|
+ self.__access_lock_versions_metadata = Lock()
|
|
|
+ self.__versions_metadata = []
|
|
|
+ VersionsFetcher.__singleton = self
|
|
|
+
|
|
|
+ def get_all_remotes_info(self) -> list[RemoteInfo]:
|
|
|
+ """
|
|
|
+ Return the list of RemoteInfo objects constructed from the
|
|
|
+ information in the remotes.json file
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ list: RemoteInfo objects for all remotes mentioned in remotes.json
|
|
|
+ """
|
|
|
+ return [
|
|
|
+ RemoteInfo(
|
|
|
+ name=remote.get('name', None),
|
|
|
+ url=remote.get('url', None)
|
|
|
+ )
|
|
|
+ for remote in self.__get_versions_metadata()
|
|
|
+ ]
|
|
|
+
|
|
|
+ def get_versions_for_vehicle(self, vehicle_name: str) -> list[VersionInfo]:
|
|
|
+ """
|
|
|
+ Return the list of dictionaries containing the info about the
|
|
|
+ versions listed to be built for a particular vehicle.
|
|
|
+
|
|
|
+ Parameters:
|
|
|
+ vehicle_name (str): the vehicle to fetch versions list for
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ list: VersionInfo objects for all versions allowed to be
|
|
|
+ built for the said vehicle.
|
|
|
+
|
|
|
+ """
|
|
|
+ if vehicle_name is None:
|
|
|
+ raise ValueError("Vehicle is a required parameter.")
|
|
|
+
|
|
|
+ versions_list = []
|
|
|
+ for remote in self.__get_versions_metadata():
|
|
|
+ for vehicle in remote['vehicles']:
|
|
|
+ if vehicle['name'] != vehicle_name:
|
|
|
+ continue
|
|
|
+
|
|
|
+ for release in vehicle['releases']:
|
|
|
+ versions_list.append(VersionInfo(
|
|
|
+ remote=remote.get('name', None),
|
|
|
+ commit_ref=release.get('commit_reference', None),
|
|
|
+ release_type=release.get('release_type', None),
|
|
|
+ version_number=release.get('version_number', None),
|
|
|
+ ap_build_artifacts_url=release.get(
|
|
|
+ 'ap_build_artifacts_url',
|
|
|
+ None
|
|
|
+ )
|
|
|
+ ))
|
|
|
+ return versions_list
|
|
|
+
|
|
|
+ def get_all_vehicles_sorted_uniq(self) -> list[str]:
|
|
|
+ """
|
|
|
+ Return a sorted list of all vehicles listed in remotes.json structure
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ list: Vehicles listed in remotes.json
|
|
|
+
|
|
|
+ """
|
|
|
+ vehicles_set = set()
|
|
|
+ for remote in self.__get_versions_metadata():
|
|
|
+ for vehicle in remote['vehicles']:
|
|
|
+ vehicles_set.add(vehicle['name'])
|
|
|
+ return sorted(list(vehicles_set))
|
|
|
+
|
|
|
+ def is_version_listed(self, vehicle: str, remote: str,
|
|
|
+ commit_ref: str) -> bool:
|
|
|
+ """
|
|
|
+ Check if a version with given properties mentioned in remotes.json
|
|
|
+
|
|
|
+ Parameters:
|
|
|
+ vehicle (str): vehicle for which version is listed
|
|
|
+ remote (str): remote under which the version is listed
|
|
|
+ commit_ref(str): commit reference for the version
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ bool: True if the said version is mentioned in remotes.json,
|
|
|
+ False otherwise
|
|
|
+
|
|
|
+ """
|
|
|
+ if vehicle is None:
|
|
|
+ raise ValueError("Vehicle is a required parameter.")
|
|
|
+
|
|
|
+ if remote is None:
|
|
|
+ raise ValueError("Remote is a required parameter.")
|
|
|
+
|
|
|
+ if commit_ref is None:
|
|
|
+ raise ValueError("Commit reference is a required parameter.")
|
|
|
+
|
|
|
+ return (remote, commit_ref) in [
|
|
|
+ (version_info.remote, version_info.commit_ref)
|
|
|
+ for version_info in
|
|
|
+ self.get_versions_for_vehicle(vehicle_name=vehicle)
|
|
|
+ ]
|
|
|
+
|
|
|
+ def get_version_info(self, vehicle: str, remote: str,
|
|
|
+ commit_ref: str) -> VersionInfo:
|
|
|
+ """
|
|
|
+ Find first version matching the given properties in remotes.json
|
|
|
+
|
|
|
+ Parameters:
|
|
|
+ vehicle (str): vehicle for which version is listed
|
|
|
+ remote (str): remote under which the version is listed
|
|
|
+ commit_ref(str): commit reference for the version
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ VersionInfo: Object for the version matching the properties,
|
|
|
+ None if not found
|
|
|
+
|
|
|
+ """
|
|
|
+ return next(
|
|
|
+ (
|
|
|
+ version
|
|
|
+ for version in self.get_versions_for_vehicle(
|
|
|
+ vehicle_name=vehicle
|
|
|
+ )
|
|
|
+ if version.remote == remote and
|
|
|
+ version.commit_ref == commit_ref
|
|
|
+ ),
|
|
|
+ None
|
|
|
+ )
|
|
|
+
|
|
|
+ def reload_remotes_json(self) -> None:
|
|
|
+ """
|
|
|
+ Read remotes.json, validate its structure against the schema
|
|
|
+ and cache it in memory
|
|
|
+ """
|
|
|
+ # load file containing vehicles listed to be built for each
|
|
|
+ # remote along with the branches/tags/commits on which the
|
|
|
+ # firmware can be built
|
|
|
+ remotes_json_schema_path = os.path.join(
|
|
|
+ os.path.dirname(__file__),
|
|
|
+ 'remotes.schema.json'
|
|
|
+ )
|
|
|
+ with open(self.__remotes_json_path, 'r') as f, \
|
|
|
+ open(remotes_json_schema_path, 'r') as s:
|
|
|
+ versions_metadata = json.loads(f.read())
|
|
|
+ schema = json.loads(s.read())
|
|
|
+ # validate schema
|
|
|
+ jsonschema.validate(instance=versions_metadata, schema=schema)
|
|
|
+ self.__set_versions_metadata(versions_metadata=versions_metadata)
|
|
|
+
|
|
|
+ def __set_versions_metadata(self, versions_metadata: list) -> None:
|
|
|
+ """
|
|
|
+ Set versions metadata property with the one passed as parameter
|
|
|
+ This requires to acquire the access lock to avoid overwriting the
|
|
|
+ object while it is being read
|
|
|
+ """
|
|
|
+ if versions_metadata is None:
|
|
|
+ raise ValueError("versions_metadata is a required parameter. "
|
|
|
+ "Cannot be None.")
|
|
|
+
|
|
|
+ with self.__access_lock_versions_metadata:
|
|
|
+ self.__versions_metadata = versions_metadata
|
|
|
+
|
|
|
+ def __get_versions_metadata(self) -> list:
|
|
|
+ """
|
|
|
+ Read versions metadata property
|
|
|
+ This requires to acquire the access lock to avoid reading the list
|
|
|
+ while it is being modified
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ list: the versions metadata list
|
|
|
+ """
|
|
|
+ with self.__access_lock_versions_metadata:
|
|
|
+ return self.__versions_metadata
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_singleton():
|
|
|
+ return VersionsFetcher.__singleton
|
Shiv Tyagi