diff --git a/run_tests.py b/run_tests.py
index 100ce16..016dca2 100755
--- a/run_tests.py
+++ b/run_tests.py
@@ -1,8 +1,12 @@
#!/usr/bin/env python3
+import sys
import unittest
if __name__ == "__main__":
loader = unittest.TestLoader()
- suite = loader.discover("tests")
+
+ test_dir = sys.argv[1] if len(sys.argv) > 1 else "tests"
+
+ suite = loader.discover(test_dir)
runner = unittest.TextTestRunner()
runner.run(suite)
\ No newline at end of file
diff --git a/src/sc/review/code_review.py b/src/sc/review/git_flow_branch_strategy.py
similarity index 58%
rename from src/sc/review/code_review.py
rename to src/sc/review/git_flow_branch_strategy.py
index fe38fac..3a0cc3c 100644
--- a/src/sc/review/code_review.py
+++ b/src/sc/review/git_flow_branch_strategy.py
@@ -11,19 +11,14 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+from pathlib import Path
-from dataclasses import dataclass
-from enum import Enum
+from git_flow_library import GitFlowLibrary
-class CRStatus(str, Enum):
- OPEN = "Open"
- CLOSED = "Closed"
- MERGED = "Merged"
-
- def __str__(self):
- return self.value
-
-@dataclass
-class CodeReview:
- url: str
- status: CRStatus
\ No newline at end of file
+class GitFlowBranchStrategy:
+ def get_target_branch(self, directory: Path, source_branch: str) -> str:
+ try:
+ base = GitFlowLibrary.get_branch_base(source_branch, directory)
+ return base if base else GitFlowLibrary.get_develop_branch(directory)
+ except RuntimeError:
+ return "develop"
diff --git a/src/sc/review/git_host_service.py b/src/sc/review/git_host_service.py
new file mode 100644
index 0000000..4b57871
--- /dev/null
+++ b/src/sc/review/git_host_service.py
@@ -0,0 +1,74 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from collections.abc import Iterable
+
+from .models import CodeReview
+from .exceptions import RemoteUrlNotFound
+from .git_flow_branch_strategy import GitFlowBranchStrategy
+from .git_instances import GitFactory, GitInstance
+from .models import RepoInfo
+from .review_config import GitHostConfig
+
+class GitHostService:
+ def __init__(
+ self,
+ git_config: GitHostConfig | None = None,
+ factory: GitFactory | None = None,
+ branch_strategy: GitFlowBranchStrategy | None = None
+ ):
+ self._git_config = git_config or GitHostConfig()
+ self._factory = factory or GitFactory()
+ self._branch_strategy = branch_strategy or GitFlowBranchStrategy()
+
+ def get_git_review_data(self, repo_info: RepoInfo) -> CodeReview | None:
+ git_instance = self._create_git_instance(repo_info.remote_url)
+ return git_instance.get_code_review(repo_info.repo_slug, repo_info.branch)
+
+ def get_create_cr_url(
+ self,
+ repo_info: RepoInfo,
+ ) -> str:
+ git_instance = self._create_git_instance(repo_info.remote_url)
+ target_branch = self._branch_strategy.get_target_branch(
+ repo_info.directory, repo_info.branch)
+ return git_instance.get_create_cr_url(repo_info.repo_slug, repo_info.branch, target_branch)
+
+ def _create_git_instance(self, remote_url: str) -> GitInstance:
+ remote_pattern = _match_remote_pattern(
+ remote_url, self._git_config.get_patterns())
+ git_data = self._git_config.get(remote_pattern)
+ return self._factory.create(
+ git_data.provider,
+ token=git_data.token,
+ base_url=git_data.url
+ )
+
+def _match_remote_pattern(remote_url: str, url_patterns: Iterable[str]) -> str:
+ """Match the remote url to a pattern in the git instance config.
+
+ Args:
+ remote_url (str): The remote url of the git repository.
+ url_patterns (Iterable[str]): An iterable of patterns to check against.
+
+ Raises:
+ RemoteUrlNotFound: Raised when the remote url matches no patterns.
+
+ Returns:
+ str: The matched pattern.
+ """
+ for pattern in url_patterns:
+ if pattern in remote_url:
+ return pattern
+ raise RemoteUrlNotFound(f"{remote_url} doesn't match any patterns! \n"
+ f"Remote patterns found: {', '.join(url_patterns)}")
diff --git a/src/sc/review/git_instances/git_factory.py b/src/sc/review/git_instances/git_factory.py
index ab7cbbd..84fb39b 100644
--- a/src/sc/review/git_instances/git_factory.py
+++ b/src/sc/review/git_instances/git_factory.py
@@ -11,6 +11,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+from functools import cache
from .instances import GithubInstance, GitlabInstance
from .git_instance import GitInstance
@@ -21,13 +22,9 @@ class GitFactory:
"gitlab": GitlabInstance
}
- @classmethod
- def types(cls) -> list[str]:
- return list(cls._registry.keys())
-
- @classmethod
- def create(cls, name: str, token: str, base_url: str | None) -> GitInstance:
+ @cache
+ def create(self, name: str, token: str, base_url: str | None) -> GitInstance:
try:
- return cls._registry[name.lower()](token=token, base_url=base_url)
+ return self._registry[name.lower()](token=token, base_url=base_url)
except KeyError:
raise ValueError(f"Provider name {name} doesn't match any VCS instance!")
diff --git a/src/sc/review/git_instances/git_instance.py b/src/sc/review/git_instances/git_instance.py
index a30551c..53d9142 100644
--- a/src/sc/review/git_instances/git_instance.py
+++ b/src/sc/review/git_instances/git_instance.py
@@ -14,7 +14,7 @@
from abc import ABC, abstractmethod
-from ..code_review import CodeReview
+from ..models import CodeReview
class GitInstance(ABC):
def __init__(self, token: str, base_url: str | None):
@@ -35,7 +35,7 @@ def validate_connection(self) -> bool:
pass
@abstractmethod
- def get_code_review(self, repo: str, source_branch: str) -> CodeReview:
+ def get_code_review(self, repo: str, source_branch: str) -> CodeReview | None:
"""Get information about a branches code review.
Args:
@@ -43,7 +43,8 @@ def get_code_review(self, repo: str, source_branch: str) -> CodeReview:
source_branch (str): The branch the code review is made from.
Returns:
- CodeReview: dataclass with information about the code review.
+ CodeReview | None: dataclass with information about the code review or None
+ if not found.
"""
pass
diff --git a/src/sc/review/git_instances/instances/github_instance.py b/src/sc/review/git_instances/instances/github_instance.py
index d909770..eddb53f 100644
--- a/src/sc/review/git_instances/instances/github_instance.py
+++ b/src/sc/review/git_instances/instances/github_instance.py
@@ -14,7 +14,7 @@
import requests
-from sc.review.code_review import CRStatus, CodeReview
+from sc.review.models import CRStatus, CodeReview
from ..git_instance import GitInstance
class GithubInstance(GitInstance):
@@ -77,6 +77,7 @@ def get_code_review(self, repo: str, source_branch: str) -> CodeReview | None:
if not prs:
return None
+
pr = prs[0]
# GitHub marks merged PRs as state="closed", merged=True
if pr.get("merged"):
diff --git a/src/sc/review/git_instances/instances/gitlab_instance.py b/src/sc/review/git_instances/instances/gitlab_instance.py
index c414a9f..4a9609f 100644
--- a/src/sc/review/git_instances/instances/gitlab_instance.py
+++ b/src/sc/review/git_instances/instances/gitlab_instance.py
@@ -15,7 +15,7 @@
import requests
import urllib.parse
-from sc.review.code_review import CodeReview, CRStatus
+from sc.review.models import CodeReview, CRStatus
from ..git_instance import GitInstance
class GitlabInstance(GitInstance):
@@ -47,7 +47,7 @@ def validate_connection(self) -> bool:
raise ConnectionError(
f"Network connection to GitLab failed for {self.base_url}") from e
- def get_code_review(self, repo: str, source_branch: str) -> CodeReview:
+ def get_code_review(self, repo: str, source_branch: str) -> CodeReview | None:
"""Get information about a code review.
Args:
@@ -97,8 +97,8 @@ def get_create_cr_url(
self,
repo: str,
source_branch: str,
- target_branch: str="develop"
- ):
+ target_branch: str = "develop"
+ ) -> str:
params = {
"merge_request[source_branch]": source_branch,
"merge_request[target_branch]": target_branch,
diff --git a/src/sc/review/main.py b/src/sc/review/main.py
deleted file mode 100644
index 9110798..0000000
--- a/src/sc/review/main.py
+++ /dev/null
@@ -1,160 +0,0 @@
-# Copyright 2025 RDK Management
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-import getpass
-import logging
-from pathlib import Path
-import sys
-
-from git_flow_library import GitFlowLibrary
-from repo_library import RepoLibrary
-from .exceptions import ReviewException
-from .review import Review
-from .review_config import ReviewConfig, TicketHostCfg, GitInstanceCfg
-from .ticketing_instances import TicketingInstanceFactory
-from .git_instances import GitFactory
-
-logger = logging.getLogger(__name__)
-
-def review():
- try:
- if root := RepoLibrary.get_repo_root_dir(Path.cwd()):
- Review(root.parent).run_repo_command()
- elif root := GitFlowLibrary.get_git_root(Path.cwd()):
- Review(root.parent).run_git_command()
- else:
- logger.error("Not in a repo project or git repository!")
- sys.exit(1)
- except (ReviewException, ConnectionError) as e:
- logger.error(e)
- sys.exit(1)
-
-def add_git_instance():
- logger.info("Enter Git provider from the list below: ")
- logger.info("github")
- logger.info("gitlab")
-
- provider = input("> ")
- print("")
-
- if provider == "github":
- url = "https://api.github.com"
- logger.info("Enter a pattern to identify Git from remote url: ")
- logger.info(
- "E.G. github.com for all github instances or "
- "github.com/org for a particular organisation")
- pattern = input("> ")
- print("")
- elif provider == "gitlab":
- logger.info(
- "Enter the URL for the gitlab instance (e.g. https://gitlab.com "
- "or https://your-instance.com): ")
- url = input("> ")
- print("")
- pattern = url.replace("https://", "").replace("http://", "")
- else:
- logger.error("Provider matches none in the list!")
- sys.exit(1)
-
- logger.info("Enter your api token: ")
- api_key = getpass.getpass("> ")
- print("")
-
- instance = GitFactory.create(provider, api_key, url)
-
- try:
- instance.validate_connection()
- except ConnectionError as e:
- logger.error(f"Failed to connect! {e}")
- sys.exit(1)
-
- logger.info("Connection validated!")
-
- git_cfg = GitInstanceCfg(url=url, token=api_key, provider=provider)
- ReviewConfig().write_git_data(pattern, git_cfg)
-
- logger.info("Git Provider Added!")
-
-def add_ticketing_instance():
- logger.info("Enter the ticketing provider from the list below: ")
- logger.info("jira")
- logger.info("redmine")
- provider = input("> ")
- print("")
-
- if provider not in ("jira", "redmine"):
- logger.error(f"Provider {provider} not supported!")
- sys.exit(1)
-
- logger.info("Enter the branch prefix (e.g ABC for feature/ABC-123_ticket): ")
- branch_prefix = input("> ")
- print("")
-
- username = None
- if provider == "jira":
- project_prefix = f"{branch_prefix}-"
-
- logger.info("Auth type:")
- logger.info("token")
- logger.info("basic")
- auth_type = input("> ")
- print("")
-
- if auth_type not in ("token", "basic"):
- logger.error(f"Auth type {auth_type} not supported!")
- sys.exit(1)
-
- if auth_type == "basic":
- logger.info("Username:")
- username = input("> ")
- print("")
-
- else:
- project_prefix = None
- auth_type = "token"
-
- logger.info("Enter the base URL: ")
- base_url = input("> ")
- print("")
-
- logger.info("API token or password: ")
- api_token = getpass.getpass("> ")
- print("")
-
- try:
- TicketingInstanceFactory.create(
- provider=provider,
- url=base_url,
- token=api_token,
- auth_type=auth_type,
- username=username
- )
- except ConnectionError as e:
- logger.error(f"Failed to connect! {e}")
- sys.exit(1)
-
- logger.info("Connection successful!")
-
- ticket_cfg = TicketHostCfg(
- url=base_url,
- provider=provider,
- api_key=api_token,
- username=username,
- auth_type=auth_type,
- project_prefix=project_prefix
- )
-
- ReviewConfig().write_ticketing_data(branch_prefix, ticket_cfg)
-
- logger.info("Added ticketing instance!")
diff --git a/src/sc/review/models.py b/src/sc/review/models.py
new file mode 100644
index 0000000..423a521
--- /dev/null
+++ b/src/sc/review/models.py
@@ -0,0 +1,150 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum
+from pathlib import Path
+from urllib import parse
+
+@dataclass
+class Ticket:
+ url: str
+ author: str | None = None
+ assignee: str | None = None
+ comments: str | None = None
+ id: str | None = None
+ status: str | None = None
+ target_version: str | None = None
+ title: str | None = None
+
+class CRStatus(str, Enum):
+ OPEN = "Open"
+ CLOSED = "Closed"
+ MERGED = "Merged"
+
+ def __str__(self):
+ return self.value
+
+@dataclass
+class CodeReview:
+ url: str
+ status: CRStatus
+
+@dataclass
+class RepoInfo:
+ branch: str
+ directory: str | Path
+ remote_url: str
+ commit_sha: str
+ commit_author: str
+ commit_date: datetime
+ commit_message: str
+
+ @property
+ def repo_slug(self) -> str:
+ """Return the repository slug (e.g. "org/repo") from a remote url."""
+ if self.remote_url.startswith("git@"):
+ slug = self.remote_url.split(":", 1)[1]
+ else:
+ slug = parse.urlparse(self.remote_url).path.lstrip("/")
+
+ slug = slug.strip("/")
+
+ if slug.endswith(".git"):
+ slug = slug[:-4]
+
+ return slug
+
+@dataclass
+class CommentData:
+ branch: str
+ directory: str | Path
+ remote_url: str
+ ticket_url: str
+ review_status: str
+ review_url: str | None
+ create_cr_url: str | None
+ commit_sha: str
+ commit_author: str
+ commit_date: datetime
+ commit_message: str
+
+ def to_terminal(self) -> str:
+ """Generate the information for one repo to be displayed in the terminal.
+
+ Returns:
+ str: Information from one repo to be displayed in the terminal.
+ """
+ def c(code, text):
+ return f"\033[{code}m{text}\033[0m"
+
+ header = [
+ f"Branch: [{self.branch}]",
+ f"Directory: [{self.directory}]",
+ f"Git: [{self.remote_url}]",
+ ]
+
+ ticket_link = f"Ticket: [{c('34', self.ticket_url)}]"
+ if self.review_url:
+ review_status = f"Review Status: [{c('32', self.review_status)}]"
+ review_link = f"Review URL: [{c('32', self.review_url)}]"
+ else:
+ review_status = f"Review Status: [{c('31', self.review_status)}]"
+ review_link = f"Create Review URL: [{c('33', self.create_cr_url)}]"
+
+ review = [ticket_link, review_status, review_link]
+
+ commit = (
+ f"Last Commit: [{self.commit_sha}]",
+ f"Author: [{self.commit_author}]",
+ f"Date: [{self.commit_date}]",
+ "",
+ f"{self.commit_message}"
+ )
+
+ return "\n".join([*header, "", *review, "", *commit])
+
+ def to_ticket(self) -> str:
+ """Generate the information for one repo formatted for a ticket comment.
+
+ Returns:
+ str: A formatted ticket comment.
+ """
+ header = [
+ f"Branch: [{self.branch}]",
+ f"Directory: [{self.directory}]",
+ f"Git: [{self.remote_url}]",
+ ]
+
+ ticket_link = f"Ticket: [{self.ticket_url}]"
+ if self.review_url:
+ review_status = f"Review Status: [{self.review_status}]"
+ review_link = f"Review URL: [{self.review_url}]"
+ else:
+ review_status = f"Review Status: [{self.review_status}]"
+ review_link = f"Create Review URL: [{self.create_cr_url}]"
+
+ review = [ticket_link, review_status, review_link]
+
+ commit = (
+ "",
+ f"Last Commit: [{self.commit_sha}]",
+ f"Author: [{self.commit_author}]",
+ f"Date: [{self.commit_date}]",
+ "",
+ f"{self.commit_message}",
+ ""
+ )
+
+ return "\n".join([*header, "", *review, "", *commit])
diff --git a/src/sc/review/prompter.py b/src/sc/review/prompter.py
new file mode 100644
index 0000000..98f1195
--- /dev/null
+++ b/src/sc/review/prompter.py
@@ -0,0 +1,45 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import logging
+
+from .review_config import TicketHostModel
+
+logger = logging.getLogger(__name__)
+
+class Prompter:
+ def yn(self, msg: str) -> bool:
+ return input(f"{msg} (y/n): ").strip().lower() == 'y'
+
+ def ticket_selection(
+ self, ticket_conf: dict[str, TicketHostModel])-> tuple[str, str]:
+ """Prompt the user to select a ticketing instance and enter a ticket number.
+
+ Returns:
+ tuple[str, str]: The selected ticketing instance identifier and ticket
+ number.
+ """
+ logger.info("Please enter the prefix of the ticket instance:")
+ logger.info("PREFIX --- INSTANCE URL --- DESCRIPTION")
+ for id, conf in ticket_conf.items():
+ logger.info(f"{id} --- {conf.url} --- {conf.description or ''}")
+
+ input_id = input("> ")
+ while input_id not in ticket_conf.keys():
+ logger.info(f"Prefix {input_id} not found in instances.")
+ input_id = input("> ")
+
+ logger.info("Please enter your ticket number:")
+ input_num = input("> ")
+
+ return input_id, input_num
diff --git a/src/sc/review/repo_source/__init__.py b/src/sc/review/repo_source/__init__.py
new file mode 100644
index 0000000..7f3869c
--- /dev/null
+++ b/src/sc/review/repo_source/__init__.py
@@ -0,0 +1,3 @@
+from .repo_source import RepoSource
+from .manifest_repo_source import ManifestRepoSource
+from .single_repo_source import SingleRepoSource
diff --git a/src/sc/review/repo_source/manifest_repo_source.py b/src/sc/review/repo_source/manifest_repo_source.py
new file mode 100644
index 0000000..5e9315c
--- /dev/null
+++ b/src/sc/review/repo_source/manifest_repo_source.py
@@ -0,0 +1,60 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from pathlib import Path
+
+from git import Repo
+from sc_manifest_parser import ScManifest
+
+from ..models import RepoInfo
+from .repo_source import RepoSource
+
+class ManifestRepoSource(RepoSource):
+ """Returns information about repos in a Repo workspace from its manifest."""
+ def __init__(self, top_dir: Path):
+ self._top_dir = top_dir
+ self.manifest_dir = top_dir / ".repo" / "manifests"
+
+ @property
+ def active_branch(self) -> str:
+ return Repo(self.manifest_dir).active_branch.name
+
+ def get_repos(self) -> list[RepoInfo]:
+ """Get info of all repos in a manifest that are tracked to a branch."""
+ repos = self._get_project_repos()
+ repos.append(self._get_repo_info(Repo(self.manifest_dir)))
+
+ return repos
+
+ def _get_project_repos(self) -> list[RepoInfo]:
+ """Get tracked project repos defined by the manifest."""
+ manifest = ScManifest.from_repo_root(self._top_dir / ".repo")
+ repos = []
+ for proj in manifest.projects:
+ proj_repo = Repo(self._top_dir / proj.path)
+ if self._should_include_project_repo(proj_repo):
+ repos.append(self._get_repo_info(proj_repo))
+
+ return repos
+
+ def _should_include_project_repo(self, proj_repo: Repo) -> bool:
+ if proj_repo.head.is_detached:
+ return False
+
+ if not proj_repo.active_branch.tracking_branch():
+ return False
+
+ return True
+
+
+
diff --git a/src/sc/review/repo_source/repo_source.py b/src/sc/review/repo_source/repo_source.py
new file mode 100644
index 0000000..03487a6
--- /dev/null
+++ b/src/sc/review/repo_source/repo_source.py
@@ -0,0 +1,42 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from abc import ABC, abstractmethod
+
+from git import Repo
+
+from ..models import RepoInfo
+
+class RepoSource(ABC):
+ """An interface for a source that provides a list of git repositories."""
+ @property
+ @abstractmethod
+ def active_branch(self) -> str:
+ pass
+
+ @abstractmethod
+ def get_repos(self) -> list[RepoInfo]:
+ pass
+
+ def _get_repo_info(self, repo: Repo) -> RepoInfo:
+ commit = repo.head.commit
+
+ return RepoInfo(
+ branch=repo.active_branch.name,
+ directory=repo.working_dir,
+ remote_url=repo.remotes[0].url,
+ commit_sha=commit.hexsha[:10],
+ commit_author=f"{commit.author.name} <{commit.author.email}>",
+ commit_date=commit.committed_datetime,
+ commit_message=commit.message.strip()
+ )
diff --git a/src/sc/review/ticket.py b/src/sc/review/repo_source/single_repo_source.py
similarity index 54%
rename from src/sc/review/ticket.py
rename to src/sc/review/repo_source/single_repo_source.py
index 36c3e47..ebc010b 100644
--- a/src/sc/review/ticket.py
+++ b/src/sc/review/repo_source/single_repo_source.py
@@ -11,15 +11,21 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from dataclasses import dataclass
+from pathlib import Path
-@dataclass
-class Ticket:
- url: str
- author: str | None = None
- assignee: str | None = None
- comments: str | None = None
- id: str | None = None
- status: str | None = None
- target_version: str | None = None
- title: str | None = None
+from git import Repo
+
+from ..models import RepoInfo
+from .repo_source import RepoSource
+
+class SingleRepoSource(RepoSource):
+ """Returns information about a singular git repo."""
+ def __init__(self, top_dir: Path):
+ self._top_dir = top_dir
+
+ @property
+ def active_branch(self) -> str:
+ return Repo(self._top_dir).active_branch.name
+
+ def get_repos(self) -> list[RepoInfo]:
+ return [self._get_repo_info(Repo(self._top_dir))]
diff --git a/src/sc/review/review.py b/src/sc/review/review.py
index 35f6779..ebc4514 100644
--- a/src/sc/review/review.py
+++ b/src/sc/review/review.py
@@ -12,352 +12,155 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from collections.abc import Iterable
-from dataclasses import dataclass
-from datetime import datetime
+import getpass
import logging
from pathlib import Path
-import re
-from urllib import parse
-
-from git import Repo
+import sys
+from .exceptions import ReviewException
from git_flow_library import GitFlowLibrary
-from sc_manifest_parser import ScManifest
-from .exceptions import RemoteUrlNotFound, TicketIdentifierNotFound
-from .review_config import ReviewConfig, TicketHostCfg
-from .ticketing_instances import TicketingInstance, TicketingInstanceFactory
-from .git_instances import GitFactory, GitInstance
+from repo_library import RepoLibrary
+from .repo_source import ManifestRepoSource, SingleRepoSource
+from .ticket_updater import TicketUpdater
+from .review_config import GitHostConfig, GitHostModel, TicketHostConfig, TicketHostModel
+from .ticketing_instances import TicketingInstanceFactory
+from .git_instances import GitFactory
logger = logging.getLogger(__name__)
-@dataclass
-class CommentData:
- branch: str
- directory: str | Path
- remote_url: str
- review_status: str
- review_url: str | None
- create_pr_url: str
- commit_sha: str
- commit_author: str
- commit_date: datetime
- commit_message: str
-
-
-class Review:
- def __init__(self, top_dir: Path | str):
- self.top_dir = Path(top_dir)
-
- self._config = ReviewConfig()
-
- def run_git_command(self):
- repo = Repo(self.top_dir)
- try:
- identifier, ticket_num = self._match_branch(repo.active_branch.name)
- except TicketIdentifierNotFound as e:
- logger.warning(e)
- identifier, ticket_num = self._prompt_ticket_selection()
-
- ticketing_cfg = self._config.get_ticket_host_data(identifier)
- ticketing_instance = self._create_ticketing_instance(ticketing_cfg)
-
- ticket_id = f"{ticketing_cfg.project_prefix or ''}{ticket_num}"
- ticket = ticketing_instance.read_ticket(ticket_id)
-
- git_instance = self._create_git_instance(repo.remote().url)
- comment_data = self._create_comment_data(repo, git_instance)
-
- logger.info(f"Ticket URL: [{ticket.url if ticket else 'None'}]")
- logger.info("Ticket info: \n")
- print(self._generate_terminal_comment(comment_data))
- print()
-
- if self._prompt_yn("Update ticket?"):
- ticket_comment = self._generate_ticket_comment(comment_data)
- ticketing_instance.add_comment_to_ticket(ticket_id, ticket_comment)
-
- def run_repo_command(self):
- logger.info("Show check ins across all repos. Note branch must be PUSHED.\n")
- manifest_repo = Repo(self.top_dir / '.repo' / 'manifests')
-
- try:
- identifier, ticket_num = self._match_branch(manifest_repo.active_branch.name)
- except TicketIdentifierNotFound as e:
- logger.warning(e)
- identifier, ticket_num = self._prompt_ticket_selection()
-
- ticketing_cfg = self._config.get_ticket_host_data(identifier)
- ticketing_instance = self._create_ticketing_instance(ticketing_cfg)
-
- ticket_id = f"{ticketing_cfg.project_prefix or ''}{ticket_num}"
- ticket = ticketing_instance.read_ticket(ticket_id)
-
- logger.info(f"Ticket URL: [{ticket.url if ticket else ''}]")
- logger.info("Ticket info: \n")
-
- manifest = ScManifest.from_repo_root(self.top_dir / '.repo')
- comments = []
- for project in manifest.projects:
- if project.lock_status:
- continue
-
- proj_repo = Repo(self.top_dir / project.path)
- # Don't generate for projects that haven't got an upstream
- if not proj_repo.active_branch.tracking_branch():
- continue
-
- proj_git = self._create_git_instance(proj_repo.remotes[project.remote].url)
- comment_data = self._create_comment_data(
- proj_repo, proj_git)
- comments.append(comment_data)
-
- manifest_git = self._create_git_instance(manifest_repo.remote().url)
- comment_data = self._create_comment_data(
- manifest_repo, manifest_git)
- comments.append(comment_data)
-
- print(self._generate_combined_terminal_comment(comments))
- print()
-
- if self._prompt_yn("Update tickets?"):
- ticket_comment = self._generate_combined_ticket_comment(comments)
- ticketing_instance.add_comment_to_ticket(ticket_id, ticket_comment)
-
- def _match_branch(self, branch_name: str) -> tuple[str, str]:
- """Match the branch to an identifier in the config.
-
- Args:
- branch_name (str): The current branch name.
-
- Raises:
- TicketIdentifierNotFound: Raised when the branch doesn't match any
- identifiers in the ticket host config.
-
- Returns:
- tuple[str, str]: (matched_identifier, ticket_number)
- """
- host_identifiers = self._config.get_ticket_host_identifiers()
- for identifier in host_identifiers:
- # Matches the identifier, followed by - or _, followed by a number
- if m := re.search(fr'{identifier}[-_]?(\d+)', branch_name):
- ticket_num = m.group(1)
- return identifier, ticket_num
- raise TicketIdentifierNotFound(
- f"Branch {branch_name} doesn't match any ticketing instances! "
- f"Found instances {', '.join(host_identifiers)}")
-
- def _create_git_instance(self, remote_url: str) -> GitInstance:
- git_url_patterns = self._config.get_git_patterns()
- try:
- remote_pattern = self._match_remote_url(
- remote_url, git_url_patterns)
- except RemoteUrlNotFound as e:
- raise RemoteUrlNotFound(
- str(e) + f"\nRemotes patterns found: {', '.join(git_url_patterns)}"
- )
- git_data = self._config.get_git_data(remote_pattern)
- return GitFactory.create(
- git_data.provider,
- token=git_data.token,
- base_url=git_data.url
+def update_ticket():
+ """Add commit/PR information to your ticket."""
+ if root := RepoLibrary.get_repo_root_dir(Path.cwd()):
+ repo_source = ManifestRepoSource(root.parent)
+ elif root := GitFlowLibrary.get_git_root(Path.cwd()):
+ repo_source = SingleRepoSource(root.parent)
+ else:
+ logger.error("Not in a repo project or git repository!")
+ sys.exit(1)
+
+ try:
+ TicketUpdater(repo_source).run()
+ except (ReviewException, ConnectionError) as e:
+ logger.error(e)
+ sys.exit(1)
+
+def add_git_instance():
+ """Add a VCS instance for sc review."""
+ logger.info("Enter Git provider from the list below: ")
+ logger.info("github")
+ logger.info("gitlab")
+
+ provider = input("> ")
+ print("")
+
+ if provider == "github":
+ url = "https://api.github.com"
+ logger.info("Enter a pattern to identify Git from remote url: ")
+ logger.info(
+ "E.G. github.com for all github instances or "
+ "github.com/org for a particular organisation")
+ pattern = input("> ")
+ print("")
+ elif provider == "gitlab":
+ logger.info(
+ "Enter the URL for the gitlab instance (e.g. https://gitlab.com "
+ "or https://your-instance.com): ")
+ url = input("> ")
+ print("")
+ pattern = url.replace("https://", "").replace("http://", "")
+ else:
+ logger.error("Provider matches none in the list!")
+ sys.exit(1)
+
+ logger.info("Enter your api token: ")
+ api_key = getpass.getpass("> ")
+ print("")
+
+ instance = GitFactory().create(provider, api_key, url)
+
+ try:
+ instance.validate_connection()
+ except ConnectionError as e:
+ logger.error(f"Failed to connect! {e}")
+ sys.exit(1)
+
+ logger.info("Connection validated!")
+
+ git_cfg = GitHostModel(url=url, token=api_key, provider=provider)
+ GitHostConfig().write(pattern, git_cfg)
+
+ logger.info("Git Provider Added!")
+
+def add_ticketing_instance():
+ """Add a ticketing instance for sc review."""
+ logger.info("Enter the ticketing provider from the list below: ")
+ logger.info("jira")
+ logger.info("redmine")
+ provider = input("> ")
+ print("")
+
+ if provider not in ("jira", "redmine"):
+ logger.error(f"Provider {provider} not supported!")
+ sys.exit(1)
+
+ logger.info("Enter the branch prefix (e.g ABC for feature/ABC-123_ticket): ")
+ branch_prefix = input("> ")
+ print("")
+
+ username = None
+ if provider == "jira":
+ project_prefix = f"{branch_prefix}-"
+
+ logger.info("Auth type:")
+ logger.info("token")
+ logger.info("basic")
+ auth_type = input("> ")
+ print("")
+
+ if auth_type not in ("token", "basic"):
+ logger.error(f"Auth type {auth_type} not supported!")
+ sys.exit(1)
+
+ if auth_type == "basic":
+ logger.info("Username:")
+ username = input("> ")
+ print("")
+
+ else:
+ project_prefix = None
+ auth_type = "token"
+
+ logger.info("Enter the base URL: ")
+ base_url = input("> ")
+ print("")
+
+ logger.info("API token or password: ")
+ api_token = getpass.getpass("> ")
+ print("")
+
+ try:
+ TicketingInstanceFactory.create(
+ provider=provider,
+ url=base_url,
+ token=api_token,
+ auth_type=auth_type,
+ username=username
)
+ except ConnectionError as e:
+ logger.error(f"Failed to connect! {e}")
+ sys.exit(1)
- def _match_remote_url(
- self,
- remote_url: str,
- git_patterns: Iterable[str]
- ) -> str:
- """Match the remote url to a pattern in the git instance config.
-
- Args:
- remote_url (str): The remote url of the git repository.
- git_patterns (Iterable[str]): An iterable of patterns to check against.
-
- Raises:
- RemoteUrlNotFound: Raised when the remote url matches no patterns.
-
- Returns:
- str: The matched pattern.
- """
- for pattern in git_patterns:
- if pattern in remote_url:
- return pattern
- raise RemoteUrlNotFound(f"{remote_url} doesn't match any patterns!")
-
- def _get_repo_slug(self, remote_url: str) -> str:
- """Return the repository slug (e.g. "org/repo") from a remote url."""
- if remote_url.startswith("git@"):
- slug = remote_url.split(":", 1)[1]
- else:
- slug = parse.urlparse(remote_url).path.lstrip("/")
-
- if slug.endswith(".git"):
- slug = slug[:-4]
-
- return slug
-
- def _get_target_branch(self, directory: Path, source_branch: str) -> str:
- if GitFlowLibrary.is_gitflow_enabled(directory):
- base = GitFlowLibrary.get_branch_base(source_branch, directory)
- return base if base else GitFlowLibrary.get_develop_branch(directory)
- else:
- return "develop"
-
- def _prompt_yn(self, msg: str) -> bool:
- return input(f"{msg} (y/n): ").strip().lower() == 'y'
+ logger.info("Connection successful!")
- def _create_comment_data(self, repo: Repo, git_instance: GitInstance) -> CommentData:
- branch_name = repo.active_branch.name
- repo_slug = self._get_repo_slug(repo.remotes[0].url)
- cr = git_instance.get_code_review(repo_slug, branch_name)
+ ticket_cfg = TicketHostModel(
+ url=base_url,
+ provider=provider,
+ api_key=api_token,
+ username=username,
+ auth_type=auth_type,
+ project_prefix=project_prefix
+ )
- target_branch = self._get_target_branch(repo.working_dir, branch_name)
- create_pr_url = git_instance.get_create_cr_url(
- repo_slug, branch_name, target_branch)
-
- commit = repo.head.commit
-
- review_status = str(cr.status) if cr else "Not Created"
- review_url = cr.url if cr else None
-
- return CommentData(
- branch=branch_name,
- directory=repo.working_dir,
- remote_url=repo.remotes[0].url,
- review_status=review_status,
- review_url=review_url,
- create_pr_url=create_pr_url,
- commit_sha=commit.hexsha[:10],
- commit_author=f"{commit.author.name} <{commit.author.email}>",
- commit_date=commit.committed_datetime,
- commit_message=commit.message.strip()
- )
-
- def _create_ticketing_instance(self, cfg: TicketHostCfg) -> TicketingInstance:
- """Create a ticketing instance.
-
- Args:
- cfg (TicketHostCfg): Config describing a ticketing instance.
-
- Raises:
- ConnectionError: Failed to connect to ticketing instance.
-
- Returns:
- TicketingInstance: A ticketing instance class.
- """
- inst = TicketingInstanceFactory.create(
- provider=cfg.provider,
- url=cfg.url,
- token=cfg.api_key,
- auth_type=cfg.auth_type,
- username=cfg.username,
- cert=cfg.cert
- )
- return inst
-
- def _prompt_ticket_selection(self) -> tuple[str, str]:
- """Prompt the user to select a ticketing instance and enter a ticket number.
-
- Raises:
- TicketIdentifierNotFound: If the instance identifier doesn't match any
- in the config.
-
- Returns:
- tuple[str, str]: The selected ticketing instance identifier and ticket
- number.
- """
- ticket_conf = self._config.get_ticketing_config()
- logger.info("Please enter the prefix of the ticket instance:")
- logger.info("PREFIX --- INSTANCE URL --- DESCRIPTION")
- for id, conf in ticket_conf.items():
- logger.info(f"{id} --- {conf.url} --- {conf.description or ''}")
-
- input_id = input("> ")
- while input_id not in ticket_conf.keys():
- logger.info(f"Prefix {input_id} not found in instances.")
- input_id = input("> ")
-
- logger.info("Please enter your ticket number:")
- input_num = input("> ")
-
- return input_id, input_num
-
- def _generate_combined_terminal_comment(self, comments: list[CommentData]) -> str:
- return "\n\n".join(self._generate_terminal_comment(c) for c in comments)
-
- def _generate_combined_ticket_comment(self, comments: list[CommentData]) -> str:
- return "\n\n".join(self._generate_ticket_comment(c) for c in comments)
-
- def _generate_terminal_comment(self, data: CommentData) -> str:
- """Generate the information for one repo to be displayed in the terminal.
-
- Args:
- data (CommentData): The data collated from one repo.
-
- Returns:
- str: Information from one repo to be displayed in the terminal.
- """
- def c(code, text):
- return f"\033[{code}m{text}\033[0m"
-
- header = [
- f"Branch: [{data.branch}]",
- f"Directory: [{data.directory}]",
- f"Git: [{data.remote_url}]",
- ]
-
- if data.review_url:
- review_status = f"Review Status: [{c('32', data.review_status)}]"
- review_link = f"Review URL: [{c('32', data.review_url)}]"
- else:
- review_status = f"Review Status: [{c('31', data.review_status)}]"
- review_link = f"Create Review URL: [{c('33', data.create_pr_url)}]"
-
- review = [review_status, review_link]
-
- commit = (
- f"Last Commit: [{data.commit_sha}]",
- f"Author: [{data.commit_author}]",
- f"Date: [{data.commit_date}]",
- "",
- f"{data.commit_message}"
- )
-
- return "\n".join([*header, "", *review, "", *commit])
-
- def _generate_ticket_comment(self, data: CommentData) -> str:
- """Generate the information for one repo formatted for a ticket comment.
-
- Args:
- data (CommentData): The data collated for one repo.
-
- Returns:
- str: A formatted ticket comment.
- """
- header = [
- f"Branch: [{data.branch}]",
- f"Directory: [{data.directory}]",
- f"Git: [{data.remote_url}]",
- ]
-
- if data.review_url:
- review_status = f"Review Status: [{data.review_status}]"
- review_link = f"Review URL: [{data.review_url}]"
- else:
- review_status = f"Review Status: [{data.review_status}]"
- review_link = f"Create Review URL: [{data.create_pr_url}]"
-
- review = [review_status, review_link]
-
- commit = (
- "",
- f"Last Commit: [{data.commit_sha}]",
- f"Author: [{data.commit_author}]",
- f"Date: [{data.commit_date}]",
- "",
- f"{data.commit_message}",
- ""
- )
+ TicketHostConfig().write(branch_prefix, ticket_cfg)
- return "\n".join([*header, "", *review, "", *commit])
+ logger.info("Added ticketing instance!")
diff --git a/src/sc/review/review_config.py b/src/sc/review/review_config.py
index 1ca3826..5e841a3 100644
--- a/src/sc/review/review_config.py
+++ b/src/sc/review/review_config.py
@@ -19,7 +19,7 @@
from .exceptions import ConfigError
from sc.config_manager import ConfigManager
-class TicketHostCfg(BaseModel):
+class TicketHostModel(BaseModel):
model_config = ConfigDict(extra='forbid')
url: str
@@ -31,56 +31,59 @@ class TicketHostCfg(BaseModel):
description: str | None = None
cert: str | None = None
-class GitInstanceCfg(BaseModel):
- model_config = ConfigDict(extra='forbid')
-
- url: str | None = None
- token: str
- provider: str
-
-class ReviewConfig:
+class TicketHostConfig:
def __init__(self):
self._ticket_config = ConfigManager('ticketing_instances')
- self._git_config = ConfigManager('git_instances')
- def get_ticketing_config(self) -> dict[str, TicketHostCfg]:
+ def get_config(self) -> dict[str, TicketHostModel]:
"""Return all ticketing instance configs keyed by identifier."""
- return {k: TicketHostCfg(**v) for k,v in self._ticket_config.get_config().items()}
+ return {k: TicketHostModel(**v) for k,v in self._ticket_config.get_config().items()}
- def get_ticket_host_identifiers(self) -> set[str]:
+ def get_identifiers(self) -> set[str]:
"""Return all configured ticketing instance identifiers."""
- return self._ticket_config.get_config().keys()
+ return set(self._ticket_config.get_config().keys())
- def get_ticket_host_data(self, identifier: str) -> TicketHostCfg:
+ def get(self, identifier: str) -> TicketHostModel:
"""Return the ticketing config for a specific identifier."""
data = self._ticket_config.get_config().get(identifier)
if not data:
raise ConfigError(
f"Ticket instance config doesn't contain entry for {identifier}")
try:
- return TicketHostCfg(**data)
+ return TicketHostModel(**data)
except ValidationError as e:
raise ConfigError(f"Invalid config for ticketing instance {identifier}: {e}")
- def write_ticketing_data(self, branch_prefix: str, ticket_data: TicketHostCfg):
+ def write(self, branch_prefix: str, ticket_data: TicketHostModel):
"""Persist ticketing config for a branch prefix."""
self._ticket_config.update_config(
{branch_prefix: ticket_data.model_dump(exclude_none=True)})
- def get_git_patterns(self) -> set[str]:
+class GitHostModel(BaseModel):
+ model_config = ConfigDict(extra='forbid')
+
+ url: str | None = None
+ token: str
+ provider: str
+
+class GitHostConfig:
+ def __init__(self):
+ self._git_config = ConfigManager('git_instances')
+
+ def get_patterns(self) -> set[str]:
"""Return all configured git URL patterns."""
return self._git_config.get_config().keys()
- def get_git_data(self, url_pattern: str) -> GitInstanceCfg:
+ def get(self, url_pattern: str) -> GitHostModel:
"""Return the git config for a specific URL pattern."""
data = self._git_config.get_config().get(url_pattern)
if not data:
raise ConfigError(f"Git config doesn't contain entry for {url_pattern}")
try:
- return GitInstanceCfg(**data)
+ return GitHostModel(**data)
except ValidationError as e:
raise ConfigError(f"Invalid config for git instance {url_pattern}: {e}")
- def write_git_data(self, pattern: str, git_config: GitInstanceCfg):
+ def write(self, pattern: str, git_config: GitHostModel):
"""Persist the config for a specific git host."""
self._git_config.update_config({pattern: git_config.model_dump(exclude_none=True)})
diff --git a/src/sc/review/ticket_service.py b/src/sc/review/ticket_service.py
new file mode 100644
index 0000000..8a6964c
--- /dev/null
+++ b/src/sc/review/ticket_service.py
@@ -0,0 +1,79 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import re
+
+from .exceptions import TicketIdentifierNotFound
+from .models import Ticket
+from .prompter import Prompter
+from .review_config import TicketHostConfig
+from .ticketing_instances import TicketingInstance, TicketingInstanceFactory
+
+class TicketService:
+ def __init__(
+ self,
+ config: TicketHostConfig | None = None,
+ factory: TicketingInstanceFactory | None = None,
+ prompter: Prompter | None = None
+ ):
+ self._config = config or TicketHostConfig()
+ self._factory = factory or TicketingInstanceFactory()
+ self._prompter = prompter or Prompter()
+
+ def resolve(
+ self, identifier: str, ticket_num: str) -> tuple[TicketingInstance, Ticket]:
+ """Match an instance identifier and ticket num to a ticket instance and ticket."""
+ cfg = self._config.get(identifier)
+ instance = self._factory.create(
+ provider=cfg.provider,
+ url=cfg.url,
+ token=cfg.api_key,
+ auth_type=cfg.auth_type,
+ username=cfg.username,
+ cert=cfg.cert
+ )
+ ticket_id = f"{cfg.project_prefix or ''}{ticket_num}"
+ ticket = instance.read_ticket(ticket_id)
+
+ return instance, ticket
+
+ def update(self, instance: TicketingInstance, ticket: Ticket, comment: str):
+ """Update a ticket on an instance with a comment."""
+ instance.add_comment_to_ticket(ticket.id, comment)
+
+ def match_branch(self, branch_name: str) -> tuple[str, str]:
+ """Match the branch to an identifier in the config.
+
+ Args:
+ branch_name (str): The current branch name.
+
+ Raises:
+ TicketIdentifierNotFound: Raised when the branch doesn't match any
+ identifiers in the ticket host config.
+
+ Returns:
+ tuple[str, str]: (matched_identifier, ticket_number)
+ """
+ host_identifiers = self._config.get_identifiers()
+ for identifier in host_identifiers:
+ # Matches the identifier, followed by - or _, followed by a number
+ if m := re.search(fr'(?:^|/){re.escape(identifier)}[-_]?(\d+)', branch_name):
+ ticket_num = m.group(1)
+ return identifier, ticket_num
+ raise TicketIdentifierNotFound(
+ f"Branch {branch_name} doesn't match any ticketing instances! "
+ f"Found instances {', '.join(host_identifiers)}")
+
+ def prompt_ticket(self) -> tuple[str, str]:
+ """Returns identifier and ticket num by user choice."""
+ return self._prompter.ticket_selection(self._config.get_config())
diff --git a/src/sc/review/ticket_updater.py b/src/sc/review/ticket_updater.py
new file mode 100644
index 0000000..0499d72
--- /dev/null
+++ b/src/sc/review/ticket_updater.py
@@ -0,0 +1,95 @@
+# Copyright 2025 RDK Management
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from .exceptions import TicketIdentifierNotFound
+from .git_host_service import GitHostService
+from .models import CodeReview, CommentData, RepoInfo, Ticket
+from .prompter import Prompter
+from .repo_source import RepoSource
+from .ticket_service import TicketService
+
+logger = logging.getLogger(__name__)
+
+class TicketUpdater:
+ def __init__(
+ self,
+ repo_source: RepoSource,
+ ticket_service: TicketService | None = None,
+ git_service: GitHostService | None = None,
+ prompter: Prompter | None = None
+ ):
+ self.repo_source = repo_source
+ self._ticket_service = ticket_service or TicketService()
+ self._git_service = git_service or GitHostService()
+ self._prompter = prompter or Prompter()
+
+ def run(self):
+ try:
+ identifier, ticket_num = self._ticket_service.match_branch(
+ self.repo_source.active_branch)
+ except TicketIdentifierNotFound as e:
+ logger.warning(e)
+ identifier, ticket_num = self._ticket_service.prompt_ticket()
+
+ ticket_instance, ticket = self._ticket_service.resolve(identifier, ticket_num)
+
+ comments = []
+ for repo_info in self.repo_source.get_repos():
+ create_cr_url = None
+ cr = self._git_service.get_git_review_data(repo_info)
+ if not cr:
+ create_cr_url = self._git_service.get_create_cr_url(repo_info)
+
+ comments.append(self._create_comment_data(repo_info, ticket, cr, create_cr_url))
+
+ logger.info(f"Ticket URL: [{ticket.url if ticket else 'None'}]")
+ logger.info("Ticket info: \n")
+ print(self._generate_combined_terminal_comment(comments))
+ print()
+
+ if self._prompter.yn("Update ticket?"):
+ ticket_comment = self._generate_combined_ticket_comment(comments)
+ self._ticket_service.update(ticket_instance, ticket, ticket_comment)
+
+ def _create_comment_data(
+ self,
+ repo_info: RepoInfo,
+ ticket: Ticket,
+ cr: CodeReview | None,
+ create_cr_url: str | None) -> CommentData:
+ review_status = str(cr.status) if cr else "Not Created"
+ review_url = cr.url if cr else None
+
+ return CommentData(
+ branch=repo_info.branch,
+ directory=repo_info.directory,
+ remote_url=repo_info.remote_url,
+ ticket_url=ticket.url,
+ review_status=review_status,
+ review_url=review_url,
+ create_cr_url=create_cr_url,
+ commit_sha=repo_info.commit_sha,
+ commit_author=repo_info.commit_author,
+ commit_date=repo_info.commit_date,
+ commit_message=repo_info.commit_message
+ )
+
+ def _generate_combined_terminal_comment(self, comments: list[CommentData]) -> str:
+ return f"\n{'-'*100}\n".join(c.to_terminal() for c in comments)
+
+ def _generate_combined_ticket_comment(self, comments: list[CommentData]) -> str:
+ return f"\n{'-'*100}\n".join(c.to_ticket() for c in comments)
+
diff --git a/src/sc/review/ticketing_instances/instances/jira_instance.py b/src/sc/review/ticketing_instances/instances/jira_instance.py
index cd00fd1..4695a22 100644
--- a/src/sc/review/ticketing_instances/instances/jira_instance.py
+++ b/src/sc/review/ticketing_instances/instances/jira_instance.py
@@ -19,7 +19,7 @@
from jira.exceptions import JIRAError
from sc.review.exceptions import PermissionsError, TicketNotFound
-from sc.review.ticket import Ticket
+from sc.review.models import Ticket
from .. import TicketingInstance
class JiraInstance(TicketingInstance):
diff --git a/src/sc/review/ticketing_instances/instances/redmine_instance.py b/src/sc/review/ticketing_instances/instances/redmine_instance.py
index 6f1f689..78bd9e9 100644
--- a/src/sc/review/ticketing_instances/instances/redmine_instance.py
+++ b/src/sc/review/ticketing_instances/instances/redmine_instance.py
@@ -19,7 +19,7 @@
from requests.exceptions import RequestException, SSLError
from sc.review.exceptions import PermissionsError, TicketingInstanceUnreachable, TicketNotFound
-from sc.review.ticket import Ticket
+from sc.review.models import Ticket
from .. import TicketingInstance
class RedmineInstance(TicketingInstance):
diff --git a/src/sc/review/ticketing_instances/ticketing_instance.py b/src/sc/review/ticketing_instances/ticketing_instance.py
index 0dd5e12..d0df9a9 100644
--- a/src/sc/review/ticketing_instances/ticketing_instance.py
+++ b/src/sc/review/ticketing_instances/ticketing_instance.py
@@ -14,7 +14,7 @@
from abc import ABC, abstractmethod
-from ..ticket import Ticket
+from ..models import Ticket
class TicketingInstance(ABC):
"""
diff --git a/src/sc/review_cli.py b/src/sc/review_cli.py
index 3035581..58725f5 100644
--- a/src/sc/review_cli.py
+++ b/src/sc/review_cli.py
@@ -14,23 +14,23 @@
import click
-from .review import main
+from .review import review
@click.group()
def cli():
pass
-@cli.command()
-def review():
+@cli.command(name="review")
+def update_ticket():
"""Add commit/PR information to your ticket."""
- main.review()
+ review.update_ticket()
@cli.command()
def add_git_instance():
"""Add a VCS instance for sc review."""
- main.add_git_instance()
+ review.add_git_instance()
@cli.command()
def add_ticketing_instance():
"""Add a ticketing instance for sc review."""
- main.add_ticketing_instance()
\ No newline at end of file
+ review.add_ticketing_instance()
\ No newline at end of file
diff --git a/tests/review/test_git_host_service.py b/tests/review/test_git_host_service.py
new file mode 100644
index 0000000..6a769b4
--- /dev/null
+++ b/tests/review/test_git_host_service.py
@@ -0,0 +1,103 @@
+import unittest
+from unittest.mock import MagicMock
+
+from sc.review.git_host_service import GitHostService, _match_remote_pattern
+from sc.review.exceptions import RemoteUrlNotFound
+
+
+class TestGitHostService(unittest.TestCase):
+
+ def setUp(self):
+ self.mock_config = MagicMock()
+ self.mock_factory = MagicMock()
+ self.mock_strategy = MagicMock()
+
+ self.service = GitHostService(
+ git_config=self.mock_config,
+ factory=self.mock_factory,
+ branch_strategy=self.mock_strategy
+ )
+
+ def test_get_git_review_data(self):
+ repo_info = MagicMock(
+ remote_url="https://gitlab.com/org/repo",
+ repo_slug="org/repo",
+ branch="feature-1"
+ )
+
+ self.mock_config.get_patterns.return_value = ["gitlab.com"]
+ self.mock_config.get.return_value = MagicMock(
+ provider="gitlab",
+ token="token",
+ url="base"
+ )
+
+ mock_instance = MagicMock()
+ self.mock_factory.create.return_value = mock_instance
+ mock_instance.get_code_review.return_value = "review"
+
+ result = self.service.get_git_review_data(repo_info)
+
+ self.assertEqual(result, "review")
+ mock_instance.get_code_review.assert_called_once_with("org/repo", "feature-1")
+
+ def test_get_create_cr_url(self):
+ repo_info = MagicMock(
+ remote_url="https://github.com/org/repo",
+ repo_slug="org/repo",
+ branch="feature-1",
+ directory="/repo"
+ )
+
+ self.mock_config.get_patterns.return_value = ["github.com"]
+ self.mock_config.get.return_value = MagicMock(
+ provider="github",
+ token="token",
+ url="base"
+ )
+
+ mock_instance = MagicMock()
+ self.mock_factory.create.return_value = mock_instance
+
+ self.mock_strategy.get_target_branch.return_value = "main"
+ mock_instance.get_create_cr_url.return_value = "url"
+
+ result = self.service.get_create_cr_url(repo_info)
+
+ self.assertEqual(result, "url")
+ mock_instance.get_create_cr_url.assert_called_once_with(
+ "org/repo", "feature-1", "main"
+ )
+
+ def test_create_git_instance(self):
+ self.mock_config.get_patterns.return_value = ["github.com"]
+ git_data = MagicMock(provider="github", token="t", url="u")
+ self.mock_config.get.return_value = git_data
+
+ instance = MagicMock()
+ self.mock_factory.create.return_value = instance
+
+ result = self.service._create_git_instance("https://github.com/org/repo")
+
+ self.assertEqual(result, instance)
+ self.mock_factory.create.assert_called_once_with(
+ "github", token="t", base_url="u"
+ )
+
+ def test_match_remote_pattern_success(self):
+ result = _match_remote_pattern(
+ "https://github.com/org/repo",
+ ["gitlab.com", "github.com"]
+ )
+ self.assertEqual(result, "github.com")
+
+ def test_match_remote_pattern_failure(self):
+ with self.assertRaises(RemoteUrlNotFound):
+ _match_remote_pattern(
+ "https://bitbucket.org/org/repo",
+ ["gitlab.com", "github.com"]
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/tests/review/test_manifest_repo_source.py b/tests/review/test_manifest_repo_source.py
new file mode 100644
index 0000000..9ca518b
--- /dev/null
+++ b/tests/review/test_manifest_repo_source.py
@@ -0,0 +1,84 @@
+import unittest
+from unittest.mock import MagicMock, patch
+from pathlib import Path
+
+from sc.review.repo_source.manifest_repo_source import ManifestRepoSource
+
+
+class TestManifestRepoSource(unittest.TestCase):
+ def setUp(self):
+ self.top_dir = Path("/fake/top")
+ self.source = ManifestRepoSource(self.top_dir)
+
+ @patch("sc.review.repo_source.manifest_repo_source.Repo")
+ def test_active_branch(self, mock_repo):
+ mock_repo.return_value.active_branch.name = "main"
+
+ branch = self.source.active_branch
+
+ self.assertEqual(branch, "main")
+ mock_repo.assert_called_with(self.source.manifest_dir)
+
+ @patch("sc.review.repo_source.manifest_repo_source.Repo")
+ def test_should_include_project_repo_true(self, mock_repo):
+ repo = MagicMock()
+ repo.head.is_detached = False
+ repo.active_branch.tracking_branch.return_value = "origin/main"
+
+ result = self.source._should_include_project_repo(repo)
+
+ self.assertTrue(result)
+
+ def test_should_include_project_repo_detached(self):
+ repo = MagicMock()
+ repo.head.is_detached = True
+
+ result = self.source._should_include_project_repo(repo)
+
+ self.assertFalse(result)
+
+ def test_should_include_project_repo_no_tracking(self):
+ repo = MagicMock()
+ repo.head.is_detached = False
+ repo.active_branch.tracking_branch.return_value = None
+
+ result = self.source._should_include_project_repo(repo)
+
+ self.assertFalse(result)
+
+ @patch("sc.review.repo_source.manifest_repo_source.ScManifest")
+ @patch("sc.review.repo_source.manifest_repo_source.Repo")
+ def test_get_project_repos_filters_correctly(self, mock_repo, mock_manifest):
+ # Mock manifest projects
+ proj1 = MagicMock(path="proj1")
+ proj2 = MagicMock(path="proj2")
+ mock_manifest.from_repo_root.return_value.projects = [proj1, proj2]
+
+ repo1 = MagicMock()
+ repo1.head.is_detached = False
+ repo1.active_branch.tracking_branch.return_value = "origin/main"
+
+ repo2 = MagicMock()
+ repo2.head.is_detached = True # excluded
+
+ mock_repo.side_effect = [repo1, repo2]
+
+ self.source._get_repo_info = MagicMock(side_effect=["info1"])
+
+ repos = self.source._get_project_repos()
+
+ self.assertEqual(repos, ["info1"])
+ self.source._get_repo_info.assert_called_once_with(repo1)
+
+ @patch("sc.review.repo_source.manifest_repo_source.Repo")
+ def test_get_repos_includes_manifest_repo(self, mock_repo):
+ mock_repo_instance = MagicMock()
+ mock_repo.return_value = mock_repo_instance
+
+ self.source._get_project_repos = MagicMock(return_value=["proj_info"])
+ self.source._get_repo_info = MagicMock(return_value="manifest_info")
+
+ repos = self.source.get_repos()
+
+ self.assertEqual(repos, ["proj_info", "manifest_info"])
+ self.source._get_repo_info.assert_called_with(mock_repo_instance)
diff --git a/tests/review/test_repo_info.py b/tests/review/test_repo_info.py
new file mode 100644
index 0000000..70be939
--- /dev/null
+++ b/tests/review/test_repo_info.py
@@ -0,0 +1,55 @@
+import unittest
+from datetime import datetime
+from pathlib import Path
+
+from sc.review.models import RepoInfo
+
+class TestRepoInfo(unittest.TestCase):
+ def setUp(self):
+ self.base_kwargs = dict(
+ branch="main",
+ directory=Path("."),
+ commit_sha="abc123",
+ commit_author="author",
+ commit_date=datetime.now(),
+ commit_message="msg",
+ )
+
+ def test_https_url(self):
+ repo = RepoInfo(
+ remote_url="https://github.com/org/repo.git",
+ **self.base_kwargs
+ )
+ self.assertEqual(repo.repo_slug, "org/repo")
+
+ def test_https_url_no_git_suffix(self):
+ repo = RepoInfo(
+ remote_url="https://github.com/org/repo",
+ **self.base_kwargs
+ )
+ self.assertEqual(repo.repo_slug, "org/repo")
+
+ def test_ssh_url(self):
+ repo = RepoInfo(
+ remote_url="git@github.com:org/repo.git",
+ **self.base_kwargs
+ )
+ self.assertEqual(repo.repo_slug, "org/repo")
+
+ def test_ssh_url_no_git_suffix(self):
+ repo = RepoInfo(
+ remote_url="git@github.com:org/repo",
+ **self.base_kwargs
+ )
+ self.assertEqual(repo.repo_slug, "org/repo")
+
+ def test_trailing_slash(self):
+ repo = RepoInfo(
+ remote_url="https://github.com/org/repo/",
+ **self.base_kwargs
+ )
+ self.assertEqual(repo.repo_slug, "org/repo")
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file
diff --git a/tests/review/test_ticket_service.py b/tests/review/test_ticket_service.py
new file mode 100644
index 0000000..0a90240
--- /dev/null
+++ b/tests/review/test_ticket_service.py
@@ -0,0 +1,73 @@
+import unittest
+from unittest.mock import MagicMock
+
+from sc.review.ticket_service import TicketService
+from sc.review.exceptions import TicketIdentifierNotFound
+
+class TestTicketService(unittest.TestCase):
+ def setUp(self):
+ self.config = MagicMock()
+ self.factory = MagicMock()
+ self.prompter = MagicMock()
+
+ self.service = TicketService(
+ config=self.config,
+ factory=self.factory,
+ prompter=self.prompter
+ )
+
+ def test_resolve(self):
+ cfg = MagicMock(
+ provider="prov",
+ url="url",
+ api_key="key",
+ auth_type="config",
+ username="user",
+ cert="cert",
+ project_prefix="ABC-"
+ )
+ self.config.get.return_value = cfg
+
+ mock_instance = MagicMock()
+ self.factory.create.return_value = mock_instance
+
+ mock_ticket = MagicMock(id="ABC-123")
+ mock_instance.read_ticket.return_value = mock_ticket
+
+ instance, ticket = self.service.resolve("jira", 123)
+
+ self.factory.create.assert_called_once()
+ instance.read_ticket.assert_called_once_with("ABC-123")
+ self.assertEqual(instance, mock_instance)
+ self.assertEqual(ticket, mock_ticket)
+
+ def test_update(self):
+ mock_instance = MagicMock()
+ mock_ticket = MagicMock(id="ABC-123")
+
+ self.service.update(mock_instance, mock_ticket, "comment")
+
+ mock_instance.add_comment_to_ticket.assert_called_once_with("ABC-123", "comment")
+
+ def test_match_branch_success(self):
+ self.config.get_identifiers.return_value = ["ABC"]
+
+ identifier, ticket_num = self.service.match_branch("feature/ABC-123")
+
+ self.assertEqual(identifier, "ABC")
+ self.assertEqual(ticket_num, "123")
+
+ def test_match_branch_failure(self):
+ self.config.get_identifiers.return_value = ["ABC"]
+
+ with self.assertRaises(TicketIdentifierNotFound):
+ self.service.match_branch("feature/no-match")
+
+ def test_prompt_ticket(self):
+ self.config.get_config.return_value = {"cfg": "value"}
+ self.prompter.ticket_selection.return_value = ("ABC", "123")
+
+ result = self.service.prompt_ticket()
+
+ self.prompter.ticket_selection.assert_called_once_with({"cfg": "value"})
+ self.assertEqual(result, ("ABC", "123"))
diff --git a/tests/review/test_ticket_updater.py b/tests/review/test_ticket_updater.py
new file mode 100644
index 0000000..7d0acc5
--- /dev/null
+++ b/tests/review/test_ticket_updater.py
@@ -0,0 +1,117 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from sc.review.ticket_updater import TicketUpdater
+from sc.review.exceptions import TicketIdentifierNotFound
+from sc.review.models import CodeReview, RepoInfo
+
+
+class TestReview(unittest.TestCase):
+
+ def setUp(self):
+ self.repo_source = MagicMock()
+ self.ticket_service = MagicMock()
+ self.git_service = MagicMock()
+ self.prompter = MagicMock()
+
+ self.review = TicketUpdater(
+ repo_source=self.repo_source,
+ ticket_service=self.ticket_service,
+ git_service=self.git_service,
+ prompter=self.prompter
+ )
+
+ self.repo_info = RepoInfo(
+ branch="feature/test",
+ directory="dir",
+ remote_url="url",
+ commit_sha="sha",
+ commit_author="author",
+ commit_date="date",
+ commit_message="msg"
+ )
+
+ self.repo_source.get_repos.return_value = [self.repo_info]
+ self.repo_source.active_branch = "feature/test"
+
+ self.ticket = MagicMock(id=1, url="http://ticket")
+ self.ticket_service.resolve.return_value = ("instance", self.ticket)
+
+ @patch("builtins.print")
+ def test_run_happy_path(self, mock_print):
+ cr = CodeReview(status="OPEN", url="http://cr")
+ self.git_service.get_git_review_data.return_value = cr
+ self.ticket_service.match_branch.return_value = ("ABC", "123")
+ self.prompter.yn.return_value = True
+
+ self.review.run()
+
+ self.ticket_service.update.assert_called_once()
+ self.git_service.get_git_review_data.assert_called_once()
+ self.ticket_service.resolve.assert_called_once_with("ABC", "123")
+
+ @patch("builtins.print")
+ def test_run_ticket_not_found_fallback(self, mock_print):
+ self.ticket_service.match_branch.side_effect = TicketIdentifierNotFound("err")
+ self.ticket_service.prompt_ticket.return_value = ("ABC", "123")
+ self.git_service.get_git_review_data.return_value = CodeReview(status=None, url=None)
+ self.prompter.yn.return_value = False
+
+ self.review.run()
+
+ self.ticket_service.prompt_ticket.assert_called_once()
+ self.ticket_service.update.assert_not_called()
+
+ @patch("builtins.print")
+ def test_run_git_failure_creates_url(self, mock_print):
+ self.ticket_service.match_branch.return_value = ("ABC", "123")
+ self.git_service.get_git_review_data.return_value = None
+ self.git_service.get_create_cr_url.return_value = "http://create"
+ self.prompter.yn.return_value = False
+
+ self.review.run()
+
+ self.git_service.get_create_cr_url.assert_called_once()
+
+ def test_create_comment_data(self):
+ cr = CodeReview(status="OPEN", url="http://cr")
+ ticket = MagicMock(url="http://ticket")
+
+ result = self.review._create_comment_data(self.repo_info, ticket, cr, None)
+
+ self.assertEqual(result.ticket_url, "http://ticket")
+ self.assertEqual(result.review_status, "OPEN")
+ self.assertEqual(result.review_url, "http://cr")
+
+ def test_create_comment_data_no_cr(self):
+ ticket = MagicMock(url="http://ticket")
+ result = self.review._create_comment_data(self.repo_info, ticket, None, "http://create")
+
+ self.assertEqual(result.ticket_url, "http://ticket")
+ self.assertEqual(result.review_status, "Not Created")
+ self.assertIsNone(result.review_url)
+ self.assertEqual(result.create_cr_url, "http://create")
+
+ def test_generate_combined_terminal_comment(self):
+ c1 = MagicMock()
+ c1.to_terminal.return_value = "A"
+ c2 = MagicMock()
+ c2.to_terminal.return_value = "B"
+
+ result = self.review._generate_combined_terminal_comment([c1, c2])
+
+ self.assertEqual(result, f"A\n{'-'*100}\nB")
+
+ def test_generate_combined_ticket_comment(self):
+ c1 = MagicMock()
+ c1.to_ticket.return_value = "A"
+ c2 = MagicMock()
+ c2.to_ticket.return_value = "B"
+
+ result = self.review._generate_combined_ticket_comment([c1, c2])
+
+ self.assertEqual(result, f"A\n{'-'*100}\nB")
+
+
+if __name__ == "__main__":
+ unittest.main()
\ No newline at end of file