diff --git a/exercise_utils/exercise_config.py b/exercise_utils/exercise_config.py new file mode 100644 index 00000000..105361b5 --- /dev/null +++ b/exercise_utils/exercise_config.py @@ -0,0 +1,62 @@ +import json +from pathlib import Path +from typing import Any + + +EXERCISE_CONFIG_FILE_NAME = ".gitmastery-exercise.json" + + +def _merge_config_fields(config: dict[str, Any], updates: dict[str, Any]) -> None: + """ + Recursively updates a JSON-like configuration as specified by the provided dictionary. + """ + for key, value in updates.items(): + if isinstance(value, dict): + current_value = config.get(key) + if not isinstance(current_value, dict): + config[key] = {} + _merge_config_fields(config[key], value) + continue + + config[key] = value + + +def update_config_fields(updates: dict[str, Any], config_path: Path) -> None: + """ + Update fields in .gitmastery-exercise.json. + + Example updates: + { + "exercise_repo": { + "pr_number": 1, + "pr_repo_full_name": "owner/repo", + }, + "teammate": "teammate-bob", + } + """ + config_path = Path(config_path) + if not config_path.exists(): + raise FileNotFoundError( + f".gitmastery-exercise.json file not found at {config_path.resolve()}" + ) + config = json.loads(config_path.read_text()) + _merge_config_fields(config, updates) + + config_path.write_text(json.dumps(config, indent=2)) + + +def add_pr_config( + config_path: Path, + pr_number: int | None = None, + pr_repo_full_name: str | None = None, +) -> None: + exercise_repo_updates: dict[str, int | str] = {} + if pr_number is not None: + exercise_repo_updates["pr_number"] = pr_number + if pr_repo_full_name is not None: + exercise_repo_updates["pr_repo_full_name"] = pr_repo_full_name + + update_config_fields( + {"exercise_repo": exercise_repo_updates}, + config_path=config_path / ".gitmastery-exercise.json", + ) diff --git a/exercise_utils/github_cli.py b/exercise_utils/github_cli.py index 1efcf279..579e9dc8 100644 --- a/exercise_utils/github_cli.py +++ b/exercise_utils/github_cli.py @@ -1,11 +1,18 @@ """Wrapper for Github CLI commands.""" # TODO: The following should be built using the builder pattern -from typing import Optional +import json +import re +from typing import Any, Optional from exercise_utils.cli import run +_PR_STATES = {"open", "closed", "merged", "all"} +_PR_MERGE_METHODS = {"merge", "squash", "rebase"} +_PR_REVIEW_ACTIONS = {"request-changes", "comment"} + + def fork_repo( repository_name: str, fork_name: str, @@ -127,3 +134,244 @@ def get_remote_url(repository_name: str, verbose: bool) -> str: remote_url = f"git@github.com:{repository_name}.git" return remote_url + + +def create_pr( + title: str, + body: str, + base: str, + head: str, + repo_name: str, + verbose: bool, + draft: bool = False, +) -> Optional[int]: + """Create a pull request.""" + command = _build_pr_command("create", repo_name=repo_name) + command = _append_value_flag(command, "--title", title) + command = _append_value_flag(command, "--body", body) + command = _append_value_flag(command, "--base", base) + command = _append_value_flag(command, "--head", head) + command = _append_bool_flag(command, draft, "--draft") + + result = run(command, verbose) + if not result.is_success(): + return None + + match = re.search(r"/pull/(\d+)", result.stdout) + if match is None: + return None + + return int(match.group(1)) + + +def _append_repo_flag(command: list[str], repo_name: str) -> list[str]: + """Append --repo flag. PR commands require explicit repository context.""" + if repo_name.strip() == "": + raise ValueError("repo_name must be provided for deterministic PR commands") + + return [*command, "--repo", repo_name] + + +def _validate_choice(value: str, allowed: set[str], field_name: str) -> str: + """Validate a string argument against a known set of values.""" + if value not in allowed: + allowed_values = ", ".join(sorted(allowed)) + raise ValueError( + f"Invalid {field_name}: {value}. Allowed values: {allowed_values}" + ) + return value + + +def _build_pr_command(subcommand: str, *args: str, repo_name: str) -> list[str]: + """Build a gh pr command and append deterministic repository context.""" + return _append_repo_flag(["gh", "pr", subcommand, *args], repo_name) + + +def _append_bool_flag(command: list[str], enabled: bool, flag: str) -> list[str]: + """Append a CLI flag when the related boolean option is enabled.""" + return [*command, flag] if enabled else command + + +def _append_value_flag(command: list[str], flag: str, value: str) -> list[str]: + """Append a value-taking CLI option in --flag=value form.""" + return [*command, f"{flag}={value}"] + + +def _parse_json_or_default(raw_output: str, default: Any) -> Any: + """Parse JSON output and return a default value on decode failure.""" + try: + return json.loads(raw_output) + except json.JSONDecodeError: + return default + + +def view_pr(pr_number: int, repo_name: str, verbose: bool) -> dict[str, Any]: + """View pull request details.""" + fields = "title,body,state,author,headRefName,baseRefName,comments,reviews" + + command = _build_pr_command( + "view", + str(pr_number), + repo_name=repo_name, + ) + command = _append_value_flag(command, "--json", fields) + + result = run( + command, + verbose, + ) + + if result.is_success(): + parsed = _parse_json_or_default(result.stdout, {}) + return parsed if isinstance(parsed, dict) else {} + return {} + + +def comment_on_pr( + pr_number: int, + comment: str, + repo_name: str, + verbose: bool, +) -> bool: + """Add a comment to a pull request.""" + command = _build_pr_command("comment", str(pr_number), repo_name=repo_name) + command = _append_value_flag(command, "--body", comment) + + result = run( + command, + verbose, + ) + return result.is_success() + + +def list_prs( + state: str, + repo_name: str, + verbose: bool, + limit: int = 30, + search: Optional[str] = None, +) -> list[dict[str, Any]]: + """ + List pull requests. + PR state filter ('open', 'closed', 'merged', 'all') + Optional search query using GitHub search syntax. + """ + validated_state = _validate_choice(state, _PR_STATES, "state") + fields = "number,title,state,author,headRefName,baseRefName" + command = _build_pr_command("list", repo_name=repo_name) + command = _append_value_flag(command, "--state", validated_state) + command = _append_value_flag(command, "--json", fields) + command = _append_value_flag(command, "--limit", str(limit)) + + if search is not None and search.strip() != "": + command = _append_value_flag(command, "--search", search) + + result = run(command, verbose) + + if result.is_success(): + parsed = _parse_json_or_default(result.stdout, []) + return parsed if isinstance(parsed, list) else [] + return [] + + +def merge_pr( + pr_number: int, + merge_method: str, + repo_name: str, + delete_branch: bool = True, + verbose: bool = False, +) -> bool: + """ + Merge a pull request. + Merge method ('merge', 'squash', 'rebase') + """ + validated_merge_method = _validate_choice( + merge_method, + _PR_MERGE_METHODS, + "merge_method", + ) + command = _build_pr_command( + "merge", + str(pr_number), + f"--{validated_merge_method}", + repo_name=repo_name, + ) + + command = _append_bool_flag(command, delete_branch, "--delete-branch") + + result = run(command, verbose) + return result.is_success() + + +def close_pr( + pr_number: int, + repo_name: str, + comment: Optional[str] = None, + delete_branch: bool = False, + verbose: bool = False, +) -> bool: + """Close a pull request without merging.""" + command = _build_pr_command( + "close", + str(pr_number), + repo_name=repo_name, + ) + command = _append_bool_flag(command, delete_branch, "--delete-branch") + + if comment: + command = _append_value_flag(command, "--comment", comment) + + result = run(command, verbose) + return result.is_success() + + +def review_pr( + pr_number: int, + comment: str, + action: str, + repo_name: str, + verbose: bool, +) -> bool: + """ + Submit a review on a pull request. + Review action ('request-changes', 'comment') + """ + validated_action = _validate_choice(action, _PR_REVIEW_ACTIONS, "action") + command = _build_pr_command("review", str(pr_number), repo_name=repo_name) + command = _append_value_flag(command, "--body", comment) + command.append(f"--{validated_action}") + + result = run(command, verbose) + return result.is_success() + + +def get_pr_numbers_by_author(username: str, repo_name: str, verbose: bool) -> list[int]: + """Return the latest opened pull request numbers created by username in the repo.""" + command = _build_pr_command("list", repo_name=repo_name) + command = _append_value_flag(command, "--author", username) + command = _append_value_flag(command, "--state", "open") + command = _append_value_flag(command, "--json", "number") + + result = run(command, verbose) + if not result.is_success(): + return [] + + import json + + try: + prs = json.loads(result.stdout) + except json.JSONDecodeError: + return [] + + pr_numbers = [pr.get("number") for pr in prs if isinstance(pr.get("number"), int)] + pr_numbers.sort() + return pr_numbers + + +def get_latest_pr_number_by_author( + username: str, repo_full_name: str, verbose: bool +) -> Optional[int]: + """Return the latest open pull request number created by username in the repo.""" + if pr_numbers := get_pr_numbers_by_author(username, repo_full_name, verbose): + return pr_numbers[-1] + raise ValueError(f"No open PRs found for user {username} in repo {repo_full_name}.") diff --git a/exercise_utils/roles.py b/exercise_utils/roles.py new file mode 100644 index 00000000..4a7690dd --- /dev/null +++ b/exercise_utils/roles.py @@ -0,0 +1,133 @@ +import re +from typing import Optional + +from exercise_utils import git, github_cli + + +class RoleMarker: + """ + Wrapper for git and GitHub operations with automatic role marker formatting. + + Usage: + bob = RoleMarker("teammate-bob") + bob.commit("Add feature", verbose=True) + # Creates: "[ROLE:teammate-bob] Add feature" + """ + + PATTERN = re.compile(r"^\[ROLE:([a-zA-Z0-9_-]+)\]\s*", re.IGNORECASE) + + def __init__(self, role: str) -> None: + """Initialize RoleMarker with a specific role.""" + self.role = role + + @staticmethod + def format(role: str, text: str) -> str: + """ + Format text with a role marker. + Example: + format('teammate-alice', 'Add feature') -> '[ROLE:teammate-alice] Add feature' + """ + return f"[ROLE:{role}] {text}" + + @staticmethod + def extract_role(text: str) -> Optional[str]: + """Extract role name from text with role marker if present.""" + match = RoleMarker.PATTERN.match(text) + return match.group(1).lower() if match else None + + @staticmethod + def has_role_marker(text: str) -> bool: + """Check if text contains a role marker.""" + return RoleMarker.PATTERN.match(text) is not None + + @staticmethod + def strip_role_marker(text: str) -> str: + """Remove role marker from text if present.""" + return RoleMarker.PATTERN.sub("", text) + + def _format_text(self, text: str) -> str: + """Format text with this instance's role marker if not already present.""" + if not self.has_role_marker(text): + return self.format(self.role, text) + return text + + # Git operations with automatic role markers + + def commit(self, message: str, verbose: bool) -> None: + """Create a commit with automatic role marker.""" + git.commit(self._format_text(message), verbose) + + def merge_with_message( + self, target_branch: str, ff: bool, message: str, verbose: bool + ) -> None: + """Merge branches with custom message and automatic role marker.""" + git.merge_with_message(target_branch, ff, self._format_text(message), verbose) + + # GitHub PR operations with automatic role markers + + def create_pr( + self, + title: str, + body: str, + base: str, + head: str, + repo_name: str, + verbose: bool, + draft: bool = False, + ) -> Optional[int]: + """Create a pull request with automatic role markers.""" + return github_cli.create_pr( + self._format_text(title), + self._format_text(body), + base, + head, + repo_name, + verbose, + draft, + ) + + def comment_on_pr( + self, + pr_number: int, + comment: str, + repo_name: str, + verbose: bool, + ) -> bool: + """Add a comment to a pull request with automatic role marker.""" + return github_cli.comment_on_pr( + pr_number, self._format_text(comment), repo_name, verbose + ) + + def review_pr( + self, + pr_number: int, + comment: str, + action: str, + repo_name: str, + verbose: bool, + ) -> bool: + """ + Submit a review on a pull request with automatic role marker. + True if review was submitted successfully, False otherwise + """ + return github_cli.review_pr( + pr_number, + self._format_text(comment), + action, + repo_name, + verbose, + ) + + def close_pr( + self, + pr_number: int, + repo_name: str, + comment: Optional[str] = None, + delete_branch: bool = False, + verbose: bool = False, + ) -> bool: + """Close a pull request without merging.""" + formatted_comment = self._format_text(comment) if comment else None + return github_cli.close_pr( + pr_number, repo_name, formatted_comment, delete_branch, verbose + )