Implement Role Marker and Wrapper Function for Github Cli

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",
+    )

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}.")