diff --git a/qgis_deployment_toolbelt/jobs/job_profiles_synchronizer.py b/qgis_deployment_toolbelt/jobs/job_profiles_synchronizer.py index 437549f7..746f447d 100644 --- a/qgis_deployment_toolbelt/jobs/job_profiles_synchronizer.py +++ b/qgis_deployment_toolbelt/jobs/job_profiles_synchronizer.py @@ -113,16 +113,16 @@ def run(self) -> None: if self.options.get("protocol") == "git": if self.options.get("source").startswith(("git://", "http://", "https://")): downloader = RemoteGitHandler( - remote_git_uri_or_path=self.options.get("source"), - branch=self.options.get("branch", "master"), + source_repository_url=self.options.get("source"), + branch_to_use=self.options.get("branch", "master"), ) - downloader.download(local_path=self.qdt_working_folder) + downloader.download(destination_local_path=self.qdt_working_folder) elif self.options.get("source").startswith("file://"): downloader = LocalGitHandler( - remote_git_uri_or_path=self.options.get("source"), - branch=self.options.get("branch", "master"), + source_repository_path_or_uri=self.options.get("source"), + branch_to_use=self.options.get("branch", "master"), ) - downloader.download(local_path=self.qdt_working_folder) + downloader.download(destination_local_path=self.qdt_working_folder) else: logger.error( f"Source type not implemented yet: {self.options.get('source')}" diff --git a/qgis_deployment_toolbelt/profiles/git_handler_base.py b/qgis_deployment_toolbelt/profiles/git_handler_base.py index 8c8bb23b..2159b731 100644 --- a/qgis_deployment_toolbelt/profiles/git_handler_base.py +++ b/qgis_deployment_toolbelt/profiles/git_handler_base.py @@ -27,6 +27,9 @@ from giturlparse import parse as git_parse from giturlparse import validate as git_validate +# project +from qgis_deployment_toolbelt.utils.check_path import check_folder_is_empty + # ############################################################################# # ########## Globals ############### # ################################## @@ -43,17 +46,33 @@ class GitHandlerBase: It's designed to handle thoses cases: - - the distant repository is on a local network or drive - - the distant repository is on Internet (github.com, gitlab.com, gitlab.company.org...) - - the local repository is on a local network or drive + - the distant repository (source) is on a local network or drive + - the distant repository (source) is on Internet (github.com, gitlab.com, gitlab.company.org...) + - the local repository (destination) is on a local network or drive """ - distant_git_repository_path_or_url: Path | str | None = None - distant_git_repository_type: Literal["local", "remote"] | None = None - distant_git_repository_branch: str | None = None + SOURCE_REPOSITORY_ACTIVE_BRANCH: str | None = None + SOURCE_REPOSITORY_PATH_OR_URL: Path | str | None = None + SOURCE_REPOSITORY_TYPE: Literal["local", "remote"] | None = None + + DESTINATION_PATH: Path | None = None + DESTINATION_BRANCH_TO_USE: str | None = None + + def __init__( + self, + source_repository_type: Literal["local", "remote"], + branch_to_use: str | None = None, + ) -> None: + """Object instanciation. - local_git_repository_path: Path | None = None - local_git_repository_active_branch: str | None = None + Args: + source_repository_type (Literal["local", "remote"]): type of source + repository + branch_to_use (str | None, optional): branch to clone or checkout. If None, + the source active branch will be used. Defaults to None. + """ + self.DESTINATION_BRANCH_TO_USE = branch_to_use + self.SOURCE_REPOSITORY_TYPE = source_repository_type def url_parsed(self, remote_git_url: str) -> GitUrlParsed: """Return URL parsed to extract git information. @@ -63,19 +82,98 @@ def url_parsed(self, remote_git_url: str) -> GitUrlParsed: """ return git_parse(remote_git_url) - def is_local_path_git_repository(self, local_path: Path) -> bool: - """Flag if local folder is a git repository. + def is_valid_git_repository( + self, + source_repository_path_or_url: Path | str | None = None, + force_type: Literal["local", "remote"] | None = None, + raise_error: bool = True, + ) -> bool: + """Determine if the given path or URL is a valid repository or not. + + Args: + source_repository_path_or_url (Path | str | None, optional): _description_. + Defaults to None. + force_type (Literal["local", "remote"], optional): force git repository + type to check. If None, it uses the SOURCE_REPOSITORY_TYPE attribute. + Defaults None. + raise_error (bool, optional): if True, it raises an exception. Defaults + to True. + + Raises: + NotGitRepository: if given path or URL is not a valid Git repository + + Returns: + bool: True if the given path or URL is a valid Git repository + """ + valid_source = True + + # if no local git repository passed, try to use URL defined at object level + if source_repository_path_or_url is None and isinstance( + self.SOURCE_REPOSITORY_PATH_OR_URL, (Path, str) + ): + source_repository_path_or_url: str | Path = ( + self.SOURCE_REPOSITORY_PATH_OR_URL + ) + logger.info( + f"Using source repository set at object's level: {source_repository_path_or_url}" + ) + + # use the repository type if forced or attribute + if force_type is None: + repository_type = self.SOURCE_REPOSITORY_TYPE + else: + repository_type = force_type + + # check according to the repository type + if repository_type == "remote" and not self._is_url_git_repository( + remote_git_url=source_repository_path_or_url + ): + valid_source = False + elif repository_type == "local" and not self._is_local_path_git_repository( + local_path=source_repository_path_or_url + ): + valid_source = False + + # log and/or raise + err_message = f"{source_repository_path_or_url} is not a valid repository." + if not valid_source and raise_error: + raise NotGitRepository( + f"{source_repository_path_or_url} is not a valid repository." + ) + elif not valid_source: + logger.debug(err_message) + else: + logger.debug( + f"{source_repository_path_or_url} is a valid " + f"{self.SOURCE_REPOSITORY_TYPE} repository." + ) + + return valid_source + + def _is_local_path_git_repository( + self, local_path: Path | None, raise_error: bool = True + ) -> bool: + """Check if local folder is a git repository. Args: - local_path (Path): path to check + local_path (Path, optional): path to check + raise_error (bool, optional): if enabled, log message is an error, debug + if not. Defaults to True. Returns: bool: True if there is a .git subfolder """ + # if no local git repository passed, try to use URL defined at object level + if local_path is None and isinstance(self.SOURCE_REPOSITORY_PATH_OR_URL, Path): + local_path: Path = self.SOURCE_REPOSITORY_PATH_OR_URL + try: Repo(root=f"{local_path.resolve()}") return True except NotGitRepository as err: + if not raise_error: + logger.debug(f"{local_path} is not a valid Git repository") + return False logger.error(f"{local_path} is not a valid Git repository. Trace: {err}") return False except Exception as err: @@ -85,40 +183,196 @@ def is_local_path_git_repository(self, local_path: Path) -> bool: ) return False - def is_url_git_repository(self, remote_git_url: str) -> bool: - """Flag if the remote URL is a git repository. + def _is_url_git_repository( + self, remote_git_url: str | None = None, raise_error: bool = True + ) -> bool: + """Check if the remote URL is a git repository. Args: remote_git_url (str): URL pointing to a remote git repository. + raise_error (bool, optional): if enabled, log message is an error, debug + if not. Defaults to True. Returns: bool: True if the URL is a valid git repository. """ - return git_validate(remote_git_url) + # if remote git URL not passed, try to use URL defined at object level + if remote_git_url is None and isinstance( + self.SOURCE_REPOSITORY_PATH_OR_URL, str + ): + remote_git_url = self.SOURCE_REPOSITORY_PATH_OR_URL + + try: + return git_validate(remote_git_url) + except Exception as err: + if not raise_error: + logger.debug(f"{remote_git_url} is not a valid Git repository") + return False + logger.error( + f"{remote_git_url} is not a valid Git repository. Trace: {err}" + ) + return False + + def get_active_branch_from_local_repository( + self, local_git_repository_path: Path | None = None + ) -> str: + """Retrieve git active branch from a local repository. Mainly a checker and a + wrapper around dulwich logic. + + Args: + local_git_repository_path (Path | None, optional): path to the local + repository. If not defined, the SOURCE_REPOSITORY_PATH_OR_URL object's + attribute is used if it exists. Defaults to None. + + Raises: + NotGitRepository: if the path is not a valid Git Repository + + Returns: + str: branch name + """ + # if no local git repository passed, try to use URL defined at object level + if local_git_repository_path is None and isinstance( + self.SOURCE_REPOSITORY_PATH_OR_URL, Path + ): + local_git_repository_path: Path = self.SOURCE_REPOSITORY_PATH_OR_URL + + self.is_valid_git_repository( + source_repository_path_or_url=local_git_repository_path + ) + + return porcelain.active_branch( + repo=Repo(root=f"{local_git_repository_path.resolve()}") + ).decode() + + def is_branch_existing_in_repository( + self, + branch_name: str | bytes, + repository_path_or_url: Path | str | None = None, + ) -> bool: + """Determine if the given branch name is part of the given repository. + + Args: + branch_name (str | bytes): _description_ + repository_path_or_url (Path | str, optional): URL or path pointing to a + git repository. If None, it uses the SOURCE_REPOSITORY_PATH_OR_URL + object's attribute + + Raises: + NotGitRepository: if the path is not a valid Git Repository + + Returns: + bool: True is the rbanch is part of given repository existing branches + """ + # make sure this is string + if isinstance(branch_name, bytes): + branch_name = branch_name.decode() + + # if no local git repository passed, try to use URL defined at object level + if repository_path_or_url is None and isinstance( + self.SOURCE_REPOSITORY_PATH_OR_URL, (Path, str) + ): + repository_path_or_url: Path | str = self.SOURCE_REPOSITORY_PATH_OR_URL + + # check if URL or path is pointing to a valid git repository + if not self.is_valid_git_repository( + source_repository_path_or_url=repository_path_or_url + ): + raise NotGitRepository( + f"{repository_path_or_url} is not a valid repository." + ) + + # clean branch name + refs_heads_prefix = "refs/heads/" + branch_name = branch_name.removeprefix(refs_heads_prefix) + + return branch_name in [ + branch.removeprefix(refs_heads_prefix) + for branch in self.list_remote_branches( + source_repository_path_or_url=repository_path_or_url + ) + ] + + def list_remote_branches( + self, source_repository_path_or_url: Path | str | None = None + ) -> tuple[str]: + """Retrieve git active branch from a local repository. Mainly a checker and a + wrapper around dulwich logic. + + Args: + source_repository_path_or_url (Path | str, optional): URL or path pointing + to a git repository. + + Raises: + NotGitRepository: if the path is not a valid Git Repository + + Returns: + tuple[str]: tuple of branch complete names \ + ('refs/heads/profile-for-qgis-3-34', 'refs/heads/main') + """ + # if no local git repository passed, try to use URL defined at object level + if source_repository_path_or_url is None and isinstance( + self.SOURCE_REPOSITORY_PATH_OR_URL, (Path, str) + ): + source_repository_path_or_url: Path | str = ( + self.SOURCE_REPOSITORY_PATH_OR_URL + ) + + # check if URL or path is pointing to a valid git repository + if not self.is_valid_git_repository( + source_repository_path_or_url=source_repository_path_or_url + ): + raise NotGitRepository( + f"{source_repository_path_or_url} is not a valid repository." + ) - def download(self, local_path: str | Path) -> Repo: + ls_remote_refs: dict = porcelain.ls_remote( + remote=f"{source_repository_path_or_url}" + ) + if isinstance(ls_remote_refs, dict): + source_repository_branches: list[str] = [ + ref.decode() for ref in ls_remote_refs if ref.startswith(b"refs/heads/") + ] + logger.debug( + f"{len(source_repository_branches)} branche(s) found in repository " + f"{source_repository_path_or_url}: " + f"{' ; '.join(source_repository_branches)}" + ) + return tuple(source_repository_branches) + else: + return ("",) + + def download(self, destination_local_path: Path) -> Repo: """Generic wrapper around the specific logic of this handler. Args: - local_path (str | Path): path to the local folder where to download + destination_local_path (Path): path to the local folder where to download Returns: Repo: the local repository object """ - local_git_repository = self.clone_or_pull(local_path) + if isinstance(destination_local_path, Path): + destination_local_path = destination_local_path.resolve() + + local_git_repository = self.clone_or_pull( + to_local_destination_path=destination_local_path + ) + if isinstance(local_git_repository, Repo): - self.local_git_repository_active_branch = porcelain.active_branch( + self.DESTINATION_BRANCH_TO_USE = porcelain.active_branch( local_git_repository ) + return local_git_repository - def clone_or_pull(self, local_path: str | Path) -> Repo: - """Clone or pull remote repository to local path. If this one doesn't exist, + def clone_or_pull(self, to_local_destination_path: Path, attempt: int = 1) -> Repo: + """Clone or fetch/pull remote repository to local path. If this one doesn't exist, it's created. If fetch or pull action fail, it removes the existing folder and clone the remote again. Args: - local_path (str | Path): path to the folder where to clone (or pull) + to_local_destination_path (Path): path to the folder where to clone (or pull) + attempt (int): attempt count. If attempts < 2 and it fails, the destination + path is completely removed before cloning again. Defaults to 1. Raises: err: if something fails during clone or pull operations @@ -126,51 +380,86 @@ def clone_or_pull(self, local_path: str | Path) -> Repo: Returns: Repo: the local repository object """ - # convert to path - if isinstance(local_path, str): - local_path = Path(local_path) - # clone - if local_path.exists() and not self.is_local_path_git_repository(local_path): + if ( + not to_local_destination_path.exists() + or check_folder_is_empty(to_local_destination_path) + ) and not self.is_valid_git_repository( + source_repository_path_or_url=to_local_destination_path, + raise_error=False, + force_type="local", + ): try: - return self._clone(local_path=local_path) + logger.debug("Start cloning operations...") + return self._clone(local_path=to_local_destination_path) except Exception as err: logger.error( - f"Error cloning the remote repository {self.distant_git_repository_path_or_url} " - f"(branch {self.distant_git_repository_branch}) to {local_path}. " + "Error cloning the source repository " + f"{self.SOURCE_REPOSITORY_PATH_OR_URL} " + f"(branch {self.SOURCE_REPOSITORY_ACTIVE_BRANCH}) " + f"to {to_local_destination_path}. " f"Trace: {err}." ) + if attempt < 2: + logger.error( + "Clone fail 1/2. Removing target folder and trying again..." + ) + rmtree(path=to_local_destination_path, ignore_errors=True) + return self.clone_or_pull( + to_local_destination_path=to_local_destination_path, attempt=2 + ) + logger.critical("Clone fail 2/2. Abort.") + rmtree(path=to_local_destination_path, ignore_errors=True) raise err - elif local_path.exists() and self.is_local_path_git_repository(local_path): + elif to_local_destination_path.exists() and self.is_valid_git_repository( + source_repository_path_or_url=to_local_destination_path, + raise_error=False, + force_type="local", + ): # FETCH + logger.debug("Start fetching operations...") try: - self._fetch(local_path=local_path) + self._fetch(local_path=to_local_destination_path) except GitProtocolError as error: logger.error( - f"Error fetching {self.distant_git_repository_path_or_url} " - f"repository to {local_path.resolve()}. Trace: {error}." - "Trying to remove the local folder and cloning again..." + "Error fetching source repository " + f"{self.SOURCE_REPOSITORY_PATH_OR_URL} " + f"to {to_local_destination_path.resolve()}. Trace: {error}." + ) + rmtree(path=to_local_destination_path, ignore_errors=True) + return self.clone_or_pull( + to_local_destination_path=to_local_destination_path ) - rmtree(path=local_path, ignore_errors=True) - return self.clone_or_pull(local_path=local_path) # PULL + logger.debug("Start pulling operations...") try: - return self._pull(local_path=local_path) + return self._pull(local_path=to_local_destination_path) except GitProtocolError as error: logger.error( - f"Error pulling {self.distant_git_repository_path_or_url} " - f"repository to {local_path.resolve()}. Trace: {error}." + f"Error pulling {self.SOURCE_REPOSITORY_PATH_OR_URL} " + f"repository to {to_local_destination_path.resolve()}. Trace: {error}." "Trying to remove the local folder and cloning again..." ) - rmtree(path=local_path, ignore_errors=True) - return self.clone_or_pull(local_path=local_path) - elif not local_path.exists(): + rmtree(path=to_local_destination_path, ignore_errors=True) + return self.clone_or_pull( + to_local_destination_path=to_local_destination_path + ) + elif not to_local_destination_path.exists(): logger.debug( - f"Local path does not exists: {local_path.as_uri()}. " + f"Local path does not exists: {to_local_destination_path.as_uri()}. " "Creating it and trying again..." ) - local_path.mkdir(parents=True, exist_ok=True) - return self.clone_or_pull(local_path=local_path) + to_local_destination_path.mkdir(parents=True, exist_ok=True) + return self.clone_or_pull( + to_local_destination_path=to_local_destination_path + ) + else: + logger.critical( + f"Case not handle. Context: {to_local_destination_path} " + f"(empty={check_folder_is_empty(to_local_destination_path)}) " + f"valid repo={self.is_valid_git_repository(source_repository_path_or_url=to_local_destination_path, raise_error=False)}" + ) + return None def _clone(self, local_path: Path) -> Repo: """Clone the remote repository to local path. @@ -181,23 +470,51 @@ def _clone(self, local_path: Path) -> Repo: Returns: Repo: the local repository object """ - # clone - logger.info( - f"Cloning repository {self.distant_git_repository_path_or_url} to {local_path}" - ) - local_repo = porcelain.clone( - source=self.distant_git_repository_path_or_url, - target=f"{local_path.resolve()}", - branch=self.distant_git_repository_branch, - depth=0, + # make sure folder and its parents exist + if not local_path.exists(): + local_path.mkdir(parents=True, exist_ok=True) + + # make sure branch is bytes + if isinstance(self.DESTINATION_BRANCH_TO_USE, str) and len( + self.DESTINATION_BRANCH_TO_USE + ): + branch = self.DESTINATION_BRANCH_TO_USE.encode() + elif isinstance(self.DESTINATION_BRANCH_TO_USE, bytes): + branch = self.DESTINATION_BRANCH_TO_USE + else: + branch = None + + logger.debug( + f"Cloning repository {self.SOURCE_REPOSITORY_PATH_OR_URL} ({branch=}) to {local_path}" ) - gobj = local_repo.get_object(local_repo.head()) + + if self.SOURCE_REPOSITORY_TYPE == "local": + with porcelain.open_repo_closing( + path_or_repo=self.SOURCE_REPOSITORY_PATH_OR_URL + ) as repo_obj: + repo_obj.clone( + target_path=f"{local_path.resolve()}", + branch=branch, + mkdir=False, + checkout=True, + progress=None, + ) + elif self.SOURCE_REPOSITORY_TYPE == "remote": + repo_obj = porcelain.clone( + source=self.SOURCE_REPOSITORY_PATH_OR_URL, + target=f"{local_path.resolve()}", + branch=branch, + ) + else: + raise NotImplementedError(f"{self.SOURCE_REPOSITORY_TYPE} is not supported") + + gobj = repo_obj.get_object(repo_obj.head()) logger.debug( - f"Active branch: {porcelain.active_branch(local_repo)}. " + f"Active branch: {porcelain.active_branch(repo_obj)}. " f"Latest commit cloned: {gobj.sha().hexdigest()} by {gobj.author}" f" at {gobj.commit_time}." ) - return local_repo + return repo_obj def _fetch(self, local_path: Path) -> Repo: """Fetch the remote repository from the existing local repository. @@ -208,25 +525,35 @@ def _fetch(self, local_path: Path) -> Repo: Returns: Repo: the local repository object """ + # if source repository is a local path, let's convert it into str + if isinstance(self.SOURCE_REPOSITORY_PATH_OR_URL, Path): + source_repository = f"{self.SOURCE_REPOSITORY_PATH_OR_URL.resolve()}" + else: + source_repository = str(self.SOURCE_REPOSITORY_PATH_OR_URL) + # with porcelain.open_repo_closing(str(local_path.resolve())) as local_repo: logger.info( - f"Fetching repository {self.distant_git_repository_path_or_url} to {local_path}", + f"Fetching repository {source_repository} to {local_path}", ) - local_repo = Repo(root=f"{local_path.resolve()}") + + destination_local_repository = Repo(root=f"{local_path.resolve()}") porcelain.fetch( - repo=f"{local_path.resolve()}", - remote_location=self.distant_git_repository_path_or_url, + repo=destination_local_repository, + remote_location=source_repository, force=True, prune=True, prune_tags=True, - depth=0, ) + destination_local_repository.close() + logger.debug( f"Repository {local_path.resolve()} has been fetched from " - f"remote {self.distant_git_repository_path_or_url}. " - f"Local active branch: {porcelain.active_branch(local_repo)}." + f"remote {source_repository}. " + f"Local active branch: {porcelain.active_branch(destination_local_repository)}." ) + return destination_local_repository + def _pull(self, local_path: Path) -> Repo: """Pull the remote repository from the existing local repository. @@ -236,23 +563,32 @@ def _pull(self, local_path: Path) -> Repo: Returns: Repo: the local repository object """ - local_repo = Repo(root=f"{local_path.resolve()}") - logger.info( - f"Pulling repository {self.distant_git_repository_path_or_url} to {local_path}" - ) + # if source repository is a local path, let's convert it into str + if isinstance(self.SOURCE_REPOSITORY_PATH_OR_URL, Path): + source_repository = f"{self.SOURCE_REPOSITORY_PATH_OR_URL.resolve()}" + else: + source_repository = str(self.SOURCE_REPOSITORY_PATH_OR_URL) + + logger.info(f"Pulling repository {source_repository} to {local_path}") + + destination_local_repository = Repo(root=f"{local_path.resolve()}") porcelain.pull( repo=local_path, - remote_location=self.distant_git_repository_path_or_url, + remote_location=source_repository, force=True, ) - gobj = local_repo.get_object(local_repo.head()) + gobj = destination_local_repository.get_object( + destination_local_repository.head() + ) logger.debug( f"Repository {local_path.resolve()} has been pulled. " - f"Local active branch: {porcelain.active_branch(local_repo)}. " + f"Local active branch: {porcelain.active_branch(destination_local_repository)}. " f"Latest commit cloned: {gobj.sha().hexdigest()} by {gobj.author}" f" at {gobj.commit_time}" ) - return local_repo + + destination_local_repository.close() + return destination_local_repository # ############################################################################# diff --git a/qgis_deployment_toolbelt/profiles/local_git_handler.py b/qgis_deployment_toolbelt/profiles/local_git_handler.py index 57133025..7b9293f6 100644 --- a/qgis_deployment_toolbelt/profiles/local_git_handler.py +++ b/qgis_deployment_toolbelt/profiles/local_git_handler.py @@ -17,11 +17,12 @@ import logging from pathlib import Path -# 3rd party -from dulwich.errors import NotGitRepository - # project from qgis_deployment_toolbelt.profiles.git_handler_base import GitHandlerBase +from qgis_deployment_toolbelt.utils.check_path import check_path, check_var_can_be_path + +# 3rd party + # ############################################################################# # ########## Globals ############### @@ -38,39 +39,82 @@ class LocalGitHandler(GitHandlerBase): """Handle local git repository.""" def __init__( - self, remote_git_uri_or_path: str | Path, branch: str | None = None + self, + source_repository_path_or_uri: str | Path, + source_repository_type: str = "local", + branch_to_use: str | None = None, ) -> None: """Constructor. Args: - remote_git_uri_or_path (Union[str, Path]): input URI (file://...) or path (S://) - branch (str, optional): default branch name. Defaults to None. + source_repository_path_or_uri (str | Path): path to the source repository Raises: NotGitRepository: if uri_or_path doesn't point to a valid Git repository """ - self.distant_git_repository_type = "local" + super().__init__( + source_repository_type=source_repository_type, branch_to_use=branch_to_use + ) # clean up - if remote_git_uri_or_path.startswith("file://"): - remote_git_uri_or_path = remote_git_uri_or_path[7:] + if isinstance( + source_repository_path_or_uri, str + ) and source_repository_path_or_uri.startswith("file://"): + source_repository_path_or_uri = source_repository_path_or_uri[7:] logger.debug( - f"URI cleaning: 'file://' protocol prefix removed. Result: {remote_git_uri_or_path}" + f"URI cleaning: 'file://' protocol prefix removed. Result: {source_repository_path_or_uri}" ) - if remote_git_uri_or_path.endswith(".git"): + if isinstance( + source_repository_path_or_uri, str + ) and source_repository_path_or_uri.endswith(".git"): logger.debug( - f"URI cleaning: '.git' suffix removed. Result: {remote_git_uri_or_path}" + f"URI cleaning: '.git' suffix removed. Result: {source_repository_path_or_uri}" ) - remote_git_uri_or_path = remote_git_uri_or_path[:-4] + source_repository_path_or_uri = source_repository_path_or_uri[:-4] + + if isinstance(source_repository_path_or_uri, str) and check_var_can_be_path( + input_var=source_repository_path_or_uri + ): + source_repository_path_or_uri = Path( + source_repository_path_or_uri + ).resolve() + + if isinstance(source_repository_path_or_uri, Path) and check_path( + input_path=source_repository_path_or_uri, + must_be_a_file=False, + must_be_a_folder=True, + must_exists=True, + must_be_readable=True, + ): + source_repository_path_or_uri = source_repository_path_or_uri.resolve() + + self.SOURCE_REPOSITORY_PATH_OR_URL = source_repository_path_or_uri # validation - if not self.is_local_path_git_repository(remote_git_uri_or_path): - raise NotGitRepository( - f"{remote_git_uri_or_path} is not a valid Git repository." + self.is_valid_git_repository() + + self.SOURCE_REPOSITORY_ACTIVE_BRANCH = ( + self.get_active_branch_from_local_repository() + ) + + # check if passed branch exist + if branch_to_use is None: + self.DESTINATION_BRANCH_TO_USE = self.SOURCE_REPOSITORY_ACTIVE_BRANCH + logger.info( + "No branch specified. The active branch " + f"({self.SOURCE_REPOSITORY_ACTIVE_BRANCH}) of source repository " + f"({self.SOURCE_REPOSITORY_PATH_OR_URL}) will be used." ) - - self.distant_git_repository_path_or_url = remote_git_uri_or_path - self.distant_git_repository_branch = branch + else: + if not self.is_branch_existing_in_repository(branch_name=branch_to_use): + logger.error( + f"Specified branch '{branch_to_use}' has not been found in source " + f"repository ({self.SOURCE_REPOSITORY_PATH_OR_URL}). Fallback to " + f"identified active branch: {self.SOURCE_REPOSITORY_ACTIVE_BRANCH}." + ) + self.DESTINATION_BRANCH_TO_USE = self.SOURCE_REPOSITORY_ACTIVE_BRANCH + else: + self.DESTINATION_BRANCH_TO_USE = branch_to_use # ############################################################################# diff --git a/qgis_deployment_toolbelt/profiles/remote_git_handler.py b/qgis_deployment_toolbelt/profiles/remote_git_handler.py index 77541aac..f367c533 100644 --- a/qgis_deployment_toolbelt/profiles/remote_git_handler.py +++ b/qgis_deployment_toolbelt/profiles/remote_git_handler.py @@ -16,12 +16,12 @@ # Standard library import logging -# 3rd party -from giturlparse import validate as git_validate - # project from qgis_deployment_toolbelt.profiles.git_handler_base import GitHandlerBase +# 3rd party + + # ############################################################################# # ########## Globals ############### # ################################## @@ -36,28 +36,44 @@ class RemoteGitHandler(GitHandlerBase): """Handle remote git repository.""" - def __init__(self, remote_git_uri_or_path: str, branch: str | None = None) -> None: + def __init__( + self, + source_repository_url: str, + source_repository_type: str = "remote", + branch_to_use: str | None = None, + ) -> None: """Constructor. Args: - remote_git_uri_or_path (Union[str, Path]): input URI (http://, https://, git://) - branch (str, optional): default branch name. Defaults to None. + source_repository_url (Union[str, Path]): input URI (http://, https://, git://) """ - self.distant_git_repository_type = "remote" - - # validation - if not git_validate(remote_git_uri_or_path): - raise ValueError(f"Invalid git URL: {remote_git_uri_or_path}") + super().__init__( + source_repository_type=source_repository_type, branch_to_use=branch_to_use + ) - self.distant_git_repository_path_or_url = remote_git_uri_or_path + self.SOURCE_REPOSITORY_PATH_OR_URL = source_repository_url - if isinstance(branch, (str, bytes)): - self.distant_git_repository_branch = branch + # validation + self.is_valid_git_repository() + + # check if passed branch exist + if branch_to_use is None: + self.DESTINATION_BRANCH_TO_USE = self.SOURCE_REPOSITORY_ACTIVE_BRANCH + logger.info( + "No branch specified. The default branch of source repository " + f"({self.SOURCE_REPOSITORY_PATH_OR_URL}) will be used." + ) else: - self.distant_git_repository_branch = self.url_parsed( - self.distant_git_repository_path_or_url - ).branch + if not self.is_branch_existing_in_repository(branch_name=branch_to_use): + logger.error( + f"Specified branch '{branch_to_use}' has not been found in source " + f"repository ({self.SOURCE_REPOSITORY_PATH_OR_URL}). Fallback to " + f"identified active branch: {self.SOURCE_REPOSITORY_ACTIVE_BRANCH}." + ) + self.DESTINATION_BRANCH_TO_USE = self.SOURCE_REPOSITORY_ACTIVE_BRANCH + else: + self.DESTINATION_BRANCH_TO_USE = branch_to_use # ############################################################################# diff --git a/qgis_deployment_toolbelt/utils/check_path.py b/qgis_deployment_toolbelt/utils/check_path.py index 08ccf417..42335824 100755 --- a/qgis_deployment_toolbelt/utils/check_path.py +++ b/qgis_deployment_toolbelt/utils/check_path.py @@ -28,6 +28,26 @@ # ################################## +def check_folder_is_empty(input_var: Path, raise_error: bool = True) -> bool: + """Check if a variable is valid path, exists, is a folder and is empty. + + Args: + input_var (Path): path to check + raise_error (bool, optional): if True, it raises an exception. Defaults to True. + + Returns: + bool: True if the path is pointing to an empty folder + """ + return check_path( + input_path=input_var, + must_be_a_file=False, + must_be_a_folder=True, + must_be_readable=True, + must_exists=True, + raise_error=raise_error, + ) and not any(input_var.iterdir()) + + def check_var_can_be_path( input_var: str, attempt: int = 1, raise_error: bool = True ) -> bool: diff --git a/requirements/testing.txt b/requirements/testing.txt index f6b51c45..9805fdcd 100644 --- a/requirements/testing.txt +++ b/requirements/testing.txt @@ -1,6 +1,7 @@ # Testing dependencies # -------------------- +GitPython>=3.1,<3.2 Pillow>=10.0.1,<10.2 pytest-cov>=4,<4.2 validators>=0.20,<0.23 diff --git a/tests/dev/dev_dulwich_local.py b/tests/dev/dev_dulwich_local.py index 81744790..81dc29e1 100644 --- a/tests/dev/dev_dulwich_local.py +++ b/tests/dev/dev_dulwich_local.py @@ -1,18 +1,104 @@ +import logging +import tempfile from pathlib import Path +from shutil import rmtree +from time import sleep from dulwich import porcelain +from dulwich.repo import Repo +from qgis_deployment_toolbelt.profiles.local_git_handler import LocalGitHandler + +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s||%(levelname)s||%(module)s||%(lineno)d||%(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +logger = logging.getLogger(__name__) + +# source local repository src_local_repo_path = Path("/home/jmo/Git/Geotribu/profils-qgis") -src_local_repo_path = Path("/home/jmo/Git/Geotribu/profils-qgis") -dst_local_repo_path = Path("/tmp/qdt-dev/dulwich-testdd/") -dst_local_repo_path.mkdir(parents=True, exist_ok=True) +# -- WITH DULWICH -- + +dst_local_repo_path_dulwich = Path("tests/fixtures/tmp/dulwich-test-local/") +dst_local_repo_path_dulwich.mkdir(parents=True, exist_ok=True) + +# src_repo = porcelain.open_repo(src_local_repo_path) + +# # print(src_repo.get_description(), dir(src_repo)) + +# dst_repo_dulwich = src_repo.clone( +# dst_local_repo_path_dulwich, mkdir=False, checkout=True, progress=None +# ) +# gobj = dst_repo_dulwich.get_object(dst_repo_dulwich.head()) +# print( +# f"{dst_repo_dulwich}. Active branch: {str(porcelain.active_branch(dst_repo_dulwich))}. " +# f"Latest commit cloned: {gobj.sha().hexdigest()} by {gobj.author}" +# f" at {gobj.commit_time}." +# ) +# src_repo.close() + -src_repo = porcelain.open_repo(src_local_repo_path) +# with porcelain.open_repo_closing(path_or_repo=src_local_repo_path) as repo_obj: +# repo_obj.clone( +# target_path=dst_local_repo_path_dulwich, +# mkdir=False, +# checkout=True, +# progress=None, +# branch=b"main", +# ) +# gobj = repo_obj.get_object(repo_obj.head()) +# print( +# f"{repo_obj}. Active branch: {str(porcelain.active_branch(repo_obj))}. " +# f"Latest commit cloned: {gobj.sha().hexdigest()} by {gobj.author}" +# f" at {gobj.commit_time}." +# ) -# print(src_repo.get_description(), dir(src_repo)) +# -- WITH QDT -- -dst_repo = src_repo.clone( - dst_local_repo_path, mkdir=False, checkout=True, progress=None +dst_local_repo_path_qdt = Path("tests/fixtures/tmp/dulwich-test-local-with-qdt/") + +local_git_handler = LocalGitHandler( + source_repository_path_or_uri=src_local_repo_path, + branch_to_use="ddddd", +) +assert local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL == src_local_repo_path +assert isinstance( + local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, Path +), "source path or URL should be a Path" + +assert local_git_handler.SOURCE_REPOSITORY_ACTIVE_BRANCH == "main" + +target_repo = local_git_handler.download(destination_local_path=dst_local_repo_path_qdt) +assert isinstance(target_repo, Repo) + +print("\n\nlet's try to fetch and pull") +target_repo2 = local_git_handler.download( + destination_local_path=dst_local_repo_path_qdt ) -print(dst_repo) + +print("\n\nTRY in a temporary folder") +local_git_handler = LocalGitHandler( + source_repository_path_or_uri=src_local_repo_path, + branch_to_use="test/not-here", +) + +with tempfile.TemporaryDirectory( + prefix="QDT_test_local_git_", + ignore_cleanup_errors=True, + suffix="_specified_branch_existing", +) as tmpdirname: + tempo_folder = Path(tmpdirname) + print(tempo_folder.resolve(), tempo_folder.exists(), tempo_folder.is_dir()) + target_repo = local_git_handler.download( + destination_local_path=tempo_folder.resolve() + ) + print(target_repo) + assert isinstance(target_repo, Repo) + +# -- CLEAN UP -- +print("\n3 seconds before cleaning up...") +sleep(3) +rmtree(dst_local_repo_path_dulwich, True) +rmtree(dst_local_repo_path_qdt, True) diff --git a/tests/dev/dev_dulwich_remote.py b/tests/dev/dev_dulwich_remote.py index a293e03b..cf38c466 100644 --- a/tests/dev/dev_dulwich_remote.py +++ b/tests/dev/dev_dulwich_remote.py @@ -1,4 +1,5 @@ import logging +import tempfile from pathlib import Path from dulwich import porcelain @@ -19,50 +20,63 @@ ) # git_repository_local.mkdir(parents=True, exist_ok=True) -# -- WITH DULWICH -- -print("\n\n-- WORKING WITH DULWICH --") -git_repository_local_dulwich = git_repository_local.joinpath("dulwich") -if ( - git_repository_local_dulwich.is_dir() - and git_repository_local_dulwich.joinpath(".git").is_dir() -): - print( - f"{git_repository_local_dulwich.resolve()} is a git local project ---> let's FETCH then PULL" - ) - dul_local_repo = Repo(root=f"{git_repository_local_dulwich.resolve()}") - porcelain.fetch( - repo=dul_local_repo, - remote_location=git_repository_remote_url, - force=True, - prune=True, - ) +# # -- WITH DULWICH -- +# print("\n\n-- WORKING WITH DULWICH --") +# git_repository_local_dulwich = git_repository_local.joinpath("dulwich") +# if ( +# git_repository_local_dulwich.is_dir() +# and git_repository_local_dulwich.joinpath(".git").is_dir() +# ): +# print( +# f"{git_repository_local_dulwich.resolve()} is a git local project ---> let's FETCH then PULL" +# ) +# dul_local_repo = Repo(root=f"{git_repository_local_dulwich.resolve()}") +# porcelain.fetch( +# repo=dul_local_repo, +# remote_location=git_repository_remote_url, +# force=True, +# prune=True, +# ) - porcelain.pull( - repo=dul_local_repo, - remote_location=git_repository_remote_url, - force=True, - fast_forward=True, - ) -else: - git_repository_local_dulwich.mkdir(parents=True, exist_ok=True) - print( - f"{git_repository_local_dulwich.resolve()} is not a git local project ---> let's CLONE" - ) - dul_local_repo = porcelain.clone( - source=git_repository_remote_url, - target=f"{git_repository_local_dulwich.resolve()}", - ) +# porcelain.pull( +# repo=dul_local_repo, +# remote_location=git_repository_remote_url, +# force=True, +# fast_forward=True, +# ) +# else: +# git_repository_local_dulwich.mkdir(parents=True, exist_ok=True) +# print( +# f"{git_repository_local_dulwich.resolve()} is not a git local project ---> let's CLONE" +# ) +# dul_local_repo = porcelain.clone( +# source=git_repository_remote_url, +# target=f"{git_repository_local_dulwich.resolve()}", +# ) -# get local active branch -active_branch = porcelain.active_branch(repo=dul_local_repo) -print(f"\nLocal active branch {active_branch.decode()}") +# # get local active branch +# active_branch = porcelain.active_branch(repo=dul_local_repo) +# print(f"\nLocal active branch {active_branch.decode()}") -dul_local_repo.close() -# assert dul_repo == dul_local_repo +# dul_local_repo.close() +# # assert dul_repo == dul_local_repo # -- WITH QDT -- print("\n\n-- WORKING WITH QDT --") -git_repository_local_dulwich = git_repository_local.joinpath("dulwich") -remote_git_handler = RemoteGitHandler(remote_git_uri_or_path=git_repository_remote_url) +git_repository_local_dulwich = git_repository_local.joinpath("with_qdt") +remote_git_handler = RemoteGitHandler(source_repository_url=git_repository_remote_url) +assert remote_git_handler.SOURCE_REPOSITORY_PATH_OR_URL == git_repository_remote_url +assert isinstance(remote_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, str) -remote_git_handler.clone_or_pull(local_path=git_repository_local_dulwich) +remote_git_handler.download(destination_local_path=git_repository_local_dulwich) + +# good_git_url = "https://gitlab.com/Oslandia/qgis/profils_qgis_fr_2022.git" +# remote_git_handler = RemoteGitHandler(source_repository_url=good_git_url) + +# with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdirname: +# local_dest = Path(tmpdirname) / "test_git_clone" +# # clone +# remote_git_handler.download(destination_local_path=local_dest) + +# # check pull is working +# remote_git_handler.clone_or_pull(to_local_destination_path=local_dest) diff --git a/tests/test_git_handler_local.py b/tests/test_git_handler_local.py new file mode 100644 index 00000000..60459c52 --- /dev/null +++ b/tests/test_git_handler_local.py @@ -0,0 +1,171 @@ +#! python3 # noqa E265 + +"""Usage from the repo root folder: + + .. code-block:: python + + # for whole test + python -m unittest tests.test_git_handler_local + # for specific + python -m unittest tests.test_git_handler_local.TestGitHandlerLocal.test_git_url_parsed +""" + +# ############################################################################# +# ########## Libraries ############# +# ################################## + +# Standard library +import tempfile +import unittest +from pathlib import Path +from shutil import rmtree + +# 3rd party +from dulwich.errors import NotGitRepository +from dulwich.repo import Repo +from git import Repo as GitPythonRepo + +# package +from qgis_deployment_toolbelt.profiles.local_git_handler import LocalGitHandler + +# ############################################################################# +# ########## Classes ############### +# ################################## + + +class TestGitHandlerLocal(unittest.TestCase): + """Test module.""" + + # -- Standard methods -------------------------------------------------------- + @classmethod + def setUpClass(cls): + """Executed when module is loaded before any test.""" + cls.remote_repo_url = "https://github.com/geotribu/profils-qgis.git" + cls.source_git_path_source = Path("tests/fixtures/tmp/git_handler_local_source") + cls.local_git_path_target = Path("tests/fixtures/tmp/git_handler_local_target") + cls.source_local_repository_obj = GitPythonRepo.clone_from( + url=cls.remote_repo_url, to_path=cls.source_git_path_source + ) + + @classmethod + def tearDownClass(cls) -> None: + """Executed before module is shutdown after every test.""" + cls.source_local_repository_obj.git.clear_cache() + cls.source_local_repository_obj.git = None + rmtree(path=cls.source_git_path_source, ignore_errors=True) + rmtree(path=cls.local_git_path_target, ignore_errors=True) + + # -- TESTS --------------------------------------------------------- + def test_initialization(self): + """Test module instanciation.""" + # OK - with pathlib.Path + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=self.source_git_path_source, + branch_to_use="main", + ) + + self.assertIsInstance(local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, Path) + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_TYPE, "local") + self.assertEqual( + local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, + self.source_git_path_source.resolve(), + ) + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_ACTIVE_BRANCH, "main") + + # OK - with str + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=f"{self.source_git_path_source.resolve()}", + branch_to_use="main", + ) + self.assertEqual( + local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, + self.source_git_path_source.resolve(), + ) + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_ACTIVE_BRANCH, "main") + + # OK - with str but no branch + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=f"{self.source_git_path_source.resolve()}", + ) + self.assertEqual( + local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, + self.source_git_path_source.resolve(), + ) + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_ACTIVE_BRANCH, "main") + self.assertIsInstance(local_git_handler.SOURCE_REPOSITORY_PATH_OR_URL, Path) + + # KO + with tempfile.TemporaryDirectory( + prefix="qdt_test_local_git_", ignore_cleanup_errors=True + ) as tmpdirname: + with self.assertRaises(NotGitRepository): + LocalGitHandler(source_repository_path_or_uri=tmpdirname) + + def test_clone_with_specified_branch_existing(self): + """Test clone with specified branch.""" + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=self.source_git_path_source, + branch_to_use="main", + ) + + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_TYPE, "local") + + with tempfile.TemporaryDirectory( + prefix="QDT_test_local_git_", + ignore_cleanup_errors=True, + suffix="_specified_branch_existing", + ) as tmpdirname: + target_repo = local_git_handler.download( + destination_local_path=Path(tmpdirname) + ) + self.assertIsInstance(target_repo, Repo) + + def test_clone_with_specified_branch_not_existing(self): + """Test clone with specified branch.""" + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=self.source_git_path_source, + branch_to_use="no_existing_branch", + ) + + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_TYPE, "local") + self.assertEqual(local_git_handler.DESTINATION_BRANCH_TO_USE, "main") + + local_git_handler.download( + destination_local_path=self.local_git_path_target.resolve() + ) + + with tempfile.TemporaryDirectory( + prefix="QDT_test_local_git_", + ignore_cleanup_errors=True, + suffix="_specified_branch_not_existing", + ) as tmpdirname: + target_repo = local_git_handler.download( + destination_local_path=Path(tmpdirname) + ) + self.assertIsInstance(target_repo, Repo) + + def test_clone_then_fetch_pull(self): + """Test clone with specified branch.""" + local_git_handler = LocalGitHandler( + source_repository_path_or_uri=self.source_git_path_source, + branch_to_use="main", + ) + + self.assertEqual(local_git_handler.SOURCE_REPOSITORY_TYPE, "local") + + with tempfile.TemporaryDirectory( + prefix="QDT_test_local_git_", + ignore_cleanup_errors=True, + suffix="_clone_the_pull", + ) as tmpdirname: + repo_in_temporary_folder = Path(tmpdirname) + + target_repo = local_git_handler.download( + destination_local_path=repo_in_temporary_folder + ) + self.assertIsInstance(target_repo, Repo) + self.assertTrue(repo_in_temporary_folder.joinpath(".git").is_dir()) + + target_repo = local_git_handler.download( + destination_local_path=repo_in_temporary_folder + ) diff --git a/tests/test_git_handler.py b/tests/test_git_handler_remote.py similarity index 64% rename from tests/test_git_handler.py rename to tests/test_git_handler_remote.py index 956a86ff..4d0bcdbe 100644 --- a/tests/test_git_handler.py +++ b/tests/test_git_handler_remote.py @@ -5,9 +5,9 @@ .. code-block:: python # for whole test - python -m unittest tests.test_git_handler + python -m unittest tests.test_git_handler_remote # for specific - python -m unittest tests.test_git_handler.TestGitHandler.test_git_url_parsed + python -m unittest tests.test_git_handler_remote.TestGitHandlerRemote.test_git_url_parsed """ # ############################################################################# @@ -18,10 +18,9 @@ import tempfile import unittest from pathlib import Path -from shutil import rmtree -from sys import version_info # 3rd party +from dulwich.errors import NotGitRepository from giturlparse import GitUrlParsed # package @@ -32,7 +31,7 @@ # ################################## -class TestGitHandler(unittest.TestCase): +class TestGitHandlerRemote(unittest.TestCase): """Test module.""" # -- Standard methods -------------------------------------------------------- @@ -41,25 +40,19 @@ def setUpClass(cls): """Executed when module is loaded before any test.""" cls.good_git_url = "https://gitlab.com/Oslandia/qgis/profils_qgis_fr_2022.git" - # standard methods - def setUp(self): - """Fixtures prepared before each test.""" - pass - - def tearDown(self): - """Executed after each test.""" - pass - # -- TESTS --------------------------------------------------------- def test_initialization(self): """Test remote git repo identifier""" # OK self.good_git_url = "https://gitlab.com/Oslandia/qgis/profils_qgis_fr_2022.git" - self.assertTrue(RemoteGitHandler(self.good_git_url).is_url_git_repository) + remote_git_handler = RemoteGitHandler(self.good_git_url) + + self.assertEqual(remote_git_handler.SOURCE_REPOSITORY_TYPE, "remote") + self.assertTrue(remote_git_handler.is_valid_git_repository()) # KO bad_git_url = "https://oslandia.com" - with self.assertRaises(ValueError): + with self.assertRaises(NotGitRepository): RemoteGitHandler(bad_git_url) def test_is_local_git_repo(self): @@ -68,9 +61,9 @@ def test_is_local_git_repo(self): git_handler = RemoteGitHandler(self.good_git_url) # OK - self.assertTrue(git_handler.is_local_path_git_repository(Path("."))) + self.assertTrue(git_handler._is_local_path_git_repository(Path("."))) # KO - self.assertFalse(git_handler.is_local_path_git_repository(Path("./tests"))) + self.assertFalse(git_handler._is_local_path_git_repository(Path("./tests"))) def test_git_url_parsed(self): """Test git parsed URL""" @@ -104,40 +97,17 @@ def test_git_url_parsed(self): self.assertEqual("gitlab", git_url_parsed.platform) self.assertEqual("profils_qgis_fr_2022", git_url_parsed.repo) - @unittest.skipUnless(version_info.minor >= 10, "requires python 3.10") - def test_git_clone_py310(self): - """Test git parsed URL. - - TODO: remove the decorator when python 3.8 and 3.9 are not supported anymore""" + def test_git_clone_remote_url(self): + """Test git parsed URL.""" self.good_git_url = "https://gitlab.com/Oslandia/qgis/profils_qgis_fr_2022.git" git_handler = RemoteGitHandler(self.good_git_url) with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as tmpdirname: local_dest = Path(tmpdirname) / "test_git_clone" # clone - git_handler.download(local_path=local_dest) + git_handler.download(destination_local_path=local_dest) # check if clone worked and new folder is a local git repo - self.assertTrue(git_handler.is_local_path_git_repository(local_dest)) + self.assertTrue(git_handler._is_local_path_git_repository(local_dest)) # check pull is working - git_handler.clone_or_pull(local_path=local_dest) - - def test_git_clone_py38(self): - """Test git parsed URL. - - TODO: remove this test when python 3.8 and 3.9 are not supported anymore""" - self.good_git_url = "https://gitlab.com/Oslandia/qgis/profils_qgis_fr_2022.git" - git_handler = RemoteGitHandler(self.good_git_url) - - # test folder - local_dest = Path(".") / "tests/fixtures/test_git_clone" - - # clone - git_handler.download(local_path=local_dest.resolve()) - # check if clone worked and new folder is a local git repo - self.assertTrue(git_handler.is_local_path_git_repository(local_dest)) - - # check pull is working - git_handler.clone_or_pull(local_path=local_dest) - - rmtree(local_dest, ignore_errors=True) + git_handler.clone_or_pull(to_local_destination_path=local_dest) diff --git a/tests/test_qplugin_object.py b/tests/test_qplugin_object.py index f3f77845..903d017c 100644 --- a/tests/test_qplugin_object.py +++ b/tests/test_qplugin_object.py @@ -13,6 +13,7 @@ # standard import unittest from pathlib import Path +from shutil import rmtree # project from qgis_deployment_toolbelt.plugins.plugin import QgisPlugin @@ -55,6 +56,10 @@ def setUpClass(cls): cls.sample_plugin_downloaded.is_file() is True ), "Downloading fixture plugin failed." + @classmethod + def tearDownClass(cls) -> None: + rmtree(path=cls.sample_plugin_downloaded.parent, ignore_errors=True) + def test_qplugin_load_from_profile(self): """Test plugin object loading from profile object.""" for p in self.good_profiles_files: @@ -166,6 +171,9 @@ def test_qplugin_load_from_zip(self): self.assertEqual(plugin_obj.download_url, sample_plugin_complex.get("url")) self.assertEqual(plugin_obj.uri_to_zip, sample_plugin_complex.get("url")) + # clean up + rmtree(path=local_plugin_download.parent, ignore_errors=True) + def test_qplugin_versions_comparison_semver(self): """Test plugin compare versions semver""" plugin_v1: QgisPlugin = QgisPlugin.from_dict( diff --git a/tests/test_utils_check_path.py b/tests/test_utils_check_path.py index b1a05a99..578e19a3 100644 --- a/tests/test_utils_check_path.py +++ b/tests/test_utils_check_path.py @@ -13,12 +13,14 @@ # standard library import stat +import tempfile import unittest from os import chmod, getenv from pathlib import Path # project from qgis_deployment_toolbelt.utils.check_path import ( + check_folder_is_empty, check_path, check_path_exists, check_path_is_readable, @@ -299,6 +301,28 @@ def test_check_path_meta_ko_specific(self): not_writable_file.unlink() + def test_check_folder_is_empty(self): + """Test empty folder recognition.""" + with tempfile.TemporaryDirectory( + prefix="QDT_test_check_path_", + ignore_cleanup_errors=True, + suffix="_empty_folder", + ) as tmpdirname: + self.assertTrue(check_folder_is_empty(Path(tmpdirname))) + + with tempfile.TemporaryDirectory( + prefix="QDT_test_check_path_", + ignore_cleanup_errors=True, + suffix="_not_empty_folder", + ) as tmpdirname: + Path(tmpdirname).joinpath(".gitkeep").touch() + self.assertFalse(check_folder_is_empty(Path(tmpdirname))) + + with self.assertRaises(TypeError): + check_folder_is_empty(input_var=1000) + # no exception but False + self.assertFalse(check_folder_is_empty(input_var=1000, raise_error=False)) + # ############################################################################ # ####### Stand-alone run ########