Merge pull request #342 from kikkomep/refactor/repository-classes

refactor repository classes

lifemonitor/api/models/repositories/__init__.py

@@ -28,13 +28,13 @@
 from .github import (GithubWorkflowRepository,
                      InstallationGithubWorkflowRepository,
                      RepoCloneContextManager)
-from .local import (LocalWorkflowRepository,
-                    LocalGitWorkflowRepository,
-                    ZippedWorkflowRepository)
+from .local import (LocalGitWorkflowRepository, LocalWorkflowRepository,
+                    Base64WorkflowRepository, ZippedWorkflowRepository)
 
 __all__ = [
     "RepositoryFile", "WorkflowRepositoryConfig", "WorkflowFile", "TemplateRepositoryFile",
     "WorkflowRepository", "WorkflowRepositoryMetadata", "IssueCheckResult",
-    "LocalWorkflowRepository", "LocalGitWorkflowRepository", "ZippedWorkflowRepository",
+    "LocalWorkflowRepository", "LocalGitWorkflowRepository",
+    "Base64WorkflowRepository", "ZippedWorkflowRepository",
     "InstallationGithubWorkflowRepository", "GithubWorkflowRepository", "RepoCloneContextManager"
 ]

lifemonitor/api/models/repositories/base.py

@@ -29,16 +29,13 @@
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple, Type, Union
 
-import git
-import giturlparse
-import requests
 from rocrate.rocrate import Metadata, ROCrate
 
 import lifemonitor.api.models.issues as issues
 from lifemonitor.api.models.repositories.config import WorkflowRepositoryConfig
-from lifemonitor.exceptions import IllegalStateException, LifeMonitorException
+from lifemonitor.exceptions import IllegalStateException
 from lifemonitor.test_metadata import get_roc_suites, get_workflow_authors
-from lifemonitor.utils import to_camel_case
+from lifemonitor.utils import get_current_username, to_camel_case
 
 from .files import RepositoryFile, WorkflowFile
 
@@ -51,18 +48,23 @@
 class WorkflowRepository():
 
     def __init__(self, local_path: str,
-                 url: Optional[str] = None,
+                 remote_url: Optional[str] = None,
+                 owner: Optional[str] = None,
                  name: Optional[str] = None,
                  license: Optional[str] = None,
-                 exclude: Optional[List[str]] = None) -> None:
+                 exclude: Optional[List[str]] = None,
+                 owner_as_system_user: bool = False) -> None:
         if not local_path:
             raise ValueError("empty local_path argument")
         self._local_path = local_path
         self._metadata = None
         self.exclude = exclude if exclude is not None else DEFAULT_IGNORED_FILES
         self._config = None
-        self._url = url
+        self._remote_url = remote_url
         self._name = name
+        self._owner = owner
+        if not owner and owner_as_system_user:
+            self._owner = get_current_username()
         self._license = license
 
     @property
@@ -82,65 +84,44 @@ def metadata(self) -> Optional[WorkflowRepositoryMetadata]:
                 return None
         return self._metadata
 
-    @property
-    def _remote_parser(self) -> giturlparse.GitUrlParsed:
-        try:
-            r = git.Repo(self.local_path)
-            remote = next((x for x in r.remotes if x.name == 'origin'), r.remotes[0])
-            assert remote, "Unable to find a Git remote"
-            return giturlparse.parse(remote.url)
-        except Exception as e:
-            if logger.isEnabledFor(logging.DEBUG):
-                logger.exception(e)
-            raise LifeMonitorException(f"Not valid workflow repository: {e}")
-
     @property
     def name(self) -> str:
-        if not self._name:
-            try:
-                self._name = self._remote_parser.name
-            except Exception as e:
-                if logger.isEnabledFor(logging.DEBUG):
-                    logger.exception(e)
-                raise LifeMonitorException(f"Not valid workflow repository: {e}")
-        assert self._name, "Unable to detect repository name"
         return self._name
 
+    @name.setter
+    def name(self, value):
+        self._name = value
+
+    @property
+    def owner(self) -> str:
+        return self._owner
+
+    @owner.setter
+    def owner(self, value):
+        self._owner = value
+
     @property
     def full_name(self) -> str:
-        try:
-            parser = self._remote_parser
-            return f"{parser.owner}/{parser.name}"
-        except Exception as e:
-            if logger.isEnabledFor(logging.DEBUG):
-                logger.exception(e)
-            raise LifeMonitorException(f"Not valid workflow repository: {e}")
+        if self.owner:
+            return f"{self.owner}/{self.name}"
+        return self.name
 
     @property
-    def https_url(self) -> str:
-        if not self._url:
-            try:
-                self._url = self._remote_parser.url2https.removesuffix('.git')
-            except Exception as e:
-                if logger.isEnabledFor(logging.DEBUG):
-                    logger.exception(e)
-                raise LifeMonitorException(f"Not valid workflow repository: {e}")
-        assert self._url, "Unable to detect repository url"
-        return self._url
+    def remote_url(self) -> str:
+        return self._remote_url
+
+    @remote_url.setter
+    def remote_url(self, value):
+        self._remote_url = value
 
     @property
     def license(self) -> Optional[str]:
-        if not self._license:
-            try:
-                parser = self._remote_parser
-                if parser.host == 'github.com':
-                    l_info = requests.get(f"https://api.github.com/repos/{self.full_name}/license")
-                    self._license = l_info.json()['license']['spdx_id']
-            except Exception as e:
-                if logger.isEnabledFor(logging.DEBUG):
-                    logger.error(e)
         return self._license
 
+    @license.setter
+    def license(self, value):
+        self._license = value
+
     @abstractclassmethod
     def find_file_by_pattern(self, search: str, path: str = '.') -> RepositoryFile:
         raise NotImplementedError()

lifemonitor/api/models/repositories/github.py

@@ -30,6 +30,11 @@
 from typing import Any, Dict, List, Optional, Union
 
 import giturlparse
+import requests
+from github.ContentFile import ContentFile
+from github.Repository import Repository as GithubRepository
+from github.Requester import Requester
+
 from lifemonitor.api.models.repositories.base import (
     WorkflowRepository, WorkflowRepositoryMetadata)
 from lifemonitor.api.models.repositories.config import WorkflowRepositoryConfig
@@ -41,11 +46,7 @@
 from lifemonitor.utils import (checkout_ref, clone_repo, get_current_ref,
                                get_git_repo_revision)
 
-from github.ContentFile import ContentFile
-from github.Repository import Repository as GithubRepository
-from github.Requester import Requester
-
-from .local import ZippedWorkflowRepository
+from .local import LocalGitWorkflowRepository, ZippedWorkflowRepository
 
 DEFAULT_BASE_URL = "https://api.github.com"
 DEFAULT_TIMEOUT = 15
@@ -163,7 +164,7 @@ def __init__(self, requester: Requester,
                  headers: Dict[str, Union[str, int]],
                  attributes: Dict[str, Any], completed: bool,
                  ref: Optional[str] = None, rev: Optional[str] = None,
-                 name: Optional[str] = None, license: Optional[str] = None,
+                 exclude: Optional[List[str]] = None,
                  local_path: Optional[str] = None, auto_cleanup: bool = True) -> None:
         super().__init__(requester, headers, attributes, completed)
         self._ref = ref
@@ -173,8 +174,14 @@ def __init__(self, requester: Requester,
         self._local_repo: Optional[LocalWorkflowRepository] = None
         self._local_path = local_path
         self._config = None
-        self._name = name
-        self._license = license
+        self._license = None
+        self._exclude = exclude or []
+        # Check if the local path is a git repo:
+        # if so, we do not need to clone it again and we can disable the auto-cleanup
+        if local_path and (
+                not os.path.exists(local_path) or not LocalWorkflowRepository.is_git_repo(local_path)):
+            logger.warning("Local path %r already exists and it is a git repository. Thus, auto-cleanup is disabled.", local_path)
+            self.auto_cleanup = False
 
     def __repr__(self) -> str:
         return f"{self.__class__.__name__} bound to {self.url} (ref: {self.ref}, rev: {self.rev})"
@@ -194,13 +201,33 @@ def checkout_ref(self, ref: str, token: Optional[str] = None, branch_name: Optio
         return checkout_ref(self.local_path, ref, auth_token=token, branch_name=branch_name)
 
     @property
-    def https_url(self) -> str:
+    def remote_url(self) -> str:
         return self.html_url
 
+    @property
+    def owner(self) -> str:
+        onwer = super().owner
+        return onwer.login if onwer else None
+
+    @property
+    def license(self) -> Optional[str]:
+        if not self._license:
+            try:
+                l_info = requests.get(f"https://api.github.com/repos/{self.full_name}/license")
+                self._license = l_info.json()['license']['spdx_id']
+            except Exception as e:
+                if logger.isEnabledFor(logging.DEBUG):
+                    logger.error(e)
+        return self._license
+
+    @property
+    def exclude(self) -> List[str]:
+        return self._exclude
+
     @property
     def _remote_parser(self) -> giturlparse.GitUrlParsed:
         try:
-            return giturlparse.parse(self.https_url)
+            return giturlparse.parse(self.remote_url)
         except Exception as e:
             if logger.isEnabledFor(logging.DEBUG):
                 logger.exception(e)
@@ -333,12 +360,15 @@ def write_zip(self, target_path: str):
         return self.local_repo.write_zip(target_path=target_path)
 
     @property
-    def local_repo(self) -> LocalWorkflowRepository:
+    def local_repo(self) -> LocalGitWorkflowRepository:
         if not self._local_repo:
             local_path = self._local_path or tempfile.mkdtemp(dir=BaseConfig.BASE_TEMP_FOLDER)
-            logger.debug("Cloning %r", self)
-            clone_repo(self.clone_url, ref=self.ref, target_path=local_path)
-            self._local_repo = LocalWorkflowRepository(local_path=local_path)
+            if not os.path.exists(local_path) or not LocalWorkflowRepository.is_git_repo(local_path):
+                logger.debug("Cloning %r", self.clone_url)
+                clone_repo(self.clone_url, ref=self.ref, target_path=local_path)
+            else:
+                logger.debug("Skipping cloning of %r", self.clone_url)
+            self._local_repo = LocalGitWorkflowRepository(local_path=local_path)
         return self._local_repo
 
     @property
@@ -360,13 +390,17 @@ def cleanup(self) -> None:
 class GithubWorkflowRepository(InstallationGithubWorkflowRepository):
 
     def __init__(self, full_name_or_id: str, token: Optional[str] = None,
-                 ref: Optional[str] = None, rev: Optional[str] = None, local_path: Optional[str] = None, auto_cleanup: bool = True) -> None:
+                 exclude: Optional[List[str]] = None,
+                 ref: Optional[str] = None, rev: Optional[str] = None,
+                 local_path: Optional[str] = None, auto_cleanup: bool = True,
+                 ) -> None:
         assert isinstance(full_name_or_id, (str, int)), full_name_or_id
         url_base = "/repositories/" if isinstance(full_name_or_id, int) else "/repos/"
         url = f"{url_base}{full_name_or_id}"
         super().__init__(
             __make_requester__(token=token), headers={}, attributes={'url': url}, completed=False,
-            ref=ref, rev=rev, local_path=local_path, auto_cleanup=auto_cleanup)
+            ref=ref, rev=rev, exclude=exclude,
+            local_path=local_path, auto_cleanup=auto_cleanup)
 
     @classmethod
     def from_url(cls, url: str, token: Optional[str] = None, ref: Optional[str] = None,