Implement infrastructure to grade exercises with pr

src/git_autograder/__init__.py

@@ -9,6 +9,9 @@
     "GitAutograderBranch",
     "GitAutograderRemote",
     "GitAutograderCommit",
+    "GitAutograderPr",
+    "GitAutograderPrComment",
+    "GitAutograderPrReview",
     "GitAutograderExercise",
 ]
 
@@ -21,6 +24,9 @@
 )
 from .exercise import GitAutograderExercise
 from .output import GitAutograderOutput
+from .pr import GitAutograderPr
+from .pr_comment import GitAutograderPrComment
+from .pr_review import GitAutograderPrReview
 from .remote import GitAutograderRemote
 from .repo.repo import GitAutograderRepo
 from .repo.repo_base import GitAutograderRepoBase

src/git_autograder/commit.py

@@ -4,6 +4,8 @@
 
 from git import Commit, Stats
 
+from git_autograder.role_marker import RoleMarker
+
 
 class GitAutograderCommit:
     def __init__(self, commit: Commit) -> None:
@@ -66,3 +68,13 @@ def file(self, file_path: str) -> Iterator[Optional[str]]:
         except Exception:
             content = None
         yield content
+
+    def is_from_user(self) -> bool:
+        message = self.commit.message
+        if isinstance(message, bytes):
+            message = message.decode("utf-8", errors="replace")
+
+        if message:
+            return not RoleMarker.has_role_marker(message)
+        # If there is no message, we can assume it's from a user.
+        return True

src/git_autograder/exercise.py

@@ -15,6 +15,7 @@
     GitAutograderWrongAnswerException,
 )
 from git_autograder.exercise_config import ExerciseConfig
+from git_autograder.helpers.pr_helper.pr_helper import PrContext
 from git_autograder.output import GitAutograderOutput
 from git_autograder.repo.null_repo import NullGitAutograderRepo
 from git_autograder.repo.repo import GitAutograderRepo
@@ -72,6 +73,7 @@ def __init__(
                 self.repo = GitAutograderRepo(
                     self.config.exercise_name,
                     Path(exercise_path) / self.config.exercise_repo.repo_name,
+                    self.get_pr_context(),
                 )
         except InvalidGitRepositoryError:
             raise GitAutograderInvalidStateException("Exercise is not a Git repository")
@@ -146,3 +148,17 @@ def to_output(
 
     def wrong_answer(self, comments: List[str]) -> GitAutograderWrongAnswerException:
         return GitAutograderWrongAnswerException(comments)
+
+    def get_pr_context(self) -> Optional[PrContext]:
+        pr_number = self.config.exercise_repo.pr_number
+        pr_repo_full_name = self.config.exercise_repo.pr_repo_full_name
+        pr_context: Optional[PrContext] = None
+        if pr_number is not None and pr_repo_full_name is not None:
+            pr_context = PrContext(
+                pr_number=pr_number,
+                pr_repo_full_name=pr_repo_full_name,
+            )
+        return pr_context
+
+    def fetch_pr(self) -> None:
+        self.repo.refresh_pr_helper()

src/git_autograder/exercise_config.py

@@ -13,6 +13,8 @@ class ExerciseRepoConfig:
         repo_title: Optional[str]
         create_fork: Optional[bool]
         init: Optional[bool]
+        pr_number: Optional[int]
+        pr_repo_full_name: Optional[str]
 
     exercise_name: str
     tags: List[str]
@@ -50,6 +52,8 @@ def read_config(path: str | Path) -> "ExerciseConfig":
                 repo_title=exercise_repo["repo_title"],
                 create_fork=exercise_repo["create_fork"],
                 init=exercise_repo["init"],
+                pr_number=exercise_repo["pr_number"] if "pr_number" in exercise_repo else None,
+                pr_repo_full_name=exercise_repo["pr_repo_full_name"] if "pr_repo_full_name" in exercise_repo else None,
             ),
             downloaded_at=raw_config["downloaded_at"],
         )

src/git_autograder/helpers/__init__.py

@@ -1,6 +1,8 @@
-__all__ = ["BranchHelper", "CommitHelper", "RemoteHelper", "FileHelper"]
+__all__ = ["BranchHelper", "CommitHelper", "RemoteHelper", "FileHelper", "PrHelper", "NullPrHelper"]
 
 from .branch_helper import BranchHelper
 from .commit_helper import CommitHelper
 from .remote_helper import RemoteHelper
 from .file_helper import FileHelper
+from .pr_helper.pr_helper import PrHelper
+from .pr_helper.null_pr_helper import NullPrHelper

src/git_autograder/helpers/pr_helper/null_pr_helper.py

@@ -0,0 +1,11 @@
+from git_autograder.helpers.pr_helper.pr_helper_base import PrHelperBase
+from git_autograder.pr import GitAutograderPr
+
+
+class NullPrHelper(PrHelperBase):
+    @property
+    def pr(self) -> GitAutograderPr:
+        raise AttributeError(
+            f"Cannot access attribute pr on NullPrHelper. "
+            "Check that your exercise repo's pr_context is properly initialized."
+        )

src/git_autograder/helpers/pr_helper/pr_helper.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+from git import Repo
+from git_autograder.helpers.pr_helper.pr_helper_base import PrHelperBase
+from git_autograder.pr import GitAutograderPr
+
+
+@dataclass(frozen=True)
+class PrContext:
+    pr_number: int
+    pr_repo_full_name: str
+
+
+class PrHelper(PrHelperBase):
+    def __init__(self, context: PrContext, repo: Repo) -> None:
+        self._pr_number = context.pr_number
+        self._pr_repo_full_name = context.pr_repo_full_name
+        self._pr = GitAutograderPr.fetch(
+            pr_number=self._pr_number,
+            pr_repo_full_name=self._pr_repo_full_name,
+            repo=repo,
+        )
+
+    @property
+    def pr(self) -> GitAutograderPr:
+        return self._pr
+

src/git_autograder/helpers/pr_helper/pr_helper_base.py

@@ -0,0 +1,9 @@
+from abc import ABC, abstractmethod
+
+from git_autograder.pr import GitAutograderPr
+
+
+class PrHelperBase(ABC):
+    @property
+    @abstractmethod
+    def pr(self) -> GitAutograderPr: ...

src/git_autograder/pr.py

@@ -0,0 +1,204 @@
+from typing import Any, Dict, List, Optional
+
+from git import Repo
+from git_autograder.commit import GitAutograderCommit
+from git_autograder.exception import GitAutograderInvalidStateException
+from git_autograder.pr_builders import (
+    build_commits,
+    build_comments,
+    build_reviews,
+)
+from .pr_gateway import fetch_pull_request_data
+from .pr_comment import GitAutograderPrComment
+from .pr_review import GitAutograderPrReview
+
+
+class GitAutograderPr:
+    @classmethod
+    def fetch(
+        cls,
+        pr_number: int,
+        pr_repo_full_name: str,
+        repo: Repo,
+    ) -> "GitAutograderPr":
+        data = fetch_pull_request_data(pr_number, pr_repo_full_name)
+        return cls._build_from_data(pr_number, pr_repo_full_name, data, repo)
+
+    @classmethod
+    def _build_from_data(
+        cls,
+        pr_number: int,
+        pr_repo_full_name: str,
+        data: Dict[str, Any],
+        repo: Repo,
+    ) -> "GitAutograderPr":
+        latest_reviews_data = (data.get("latestReviews") or {}).get("nodes") or []
+        comments_data = (data.get("comments") or {}).get("nodes") or []
+        commits = (data.get("commits") or {}).get("nodes") or []
+        merged_at = data.get("mergedAt")
+        created_at = data.get("createdAt")
+
+        built_commits = build_commits(commits, repo)
+
+        return cls(
+            number=pr_number,
+            repo_full_name=pr_repo_full_name,
+            title=str(data.get("title") or ""),
+            body=str(data.get("body") or ""),
+            state=str(data.get("state") or ""),
+            author_login=(data.get("author") or {}).get("login"),
+            base_branch=str(data.get("baseRefName") or ""),
+            head_branch=str(data.get("headRefName") or ""),
+            is_draft=bool(data.get("isDraft", False)),
+            merged_at=merged_at if isinstance(merged_at, str) else None,
+            merged_by_login=(data.get("mergedBy") or {}).get("login"),
+            created_at=created_at if isinstance(created_at, str) else None,
+            commits=built_commits,
+            reviews=build_reviews(latest_reviews_data),
+            comments=build_comments(comments_data),
+        )
+
+    def __init__(
+        self,
+        number: int,
+        repo_full_name: str,
+        title: str,
+        body: str,
+        state: str,
+        author_login: Optional[str],
+        base_branch: str,
+        head_branch: str,
+        is_draft: bool,
+        merged_at: Optional[str],
+        merged_by_login: Optional[str],
+        created_at: Optional[str],
+        commits: List[GitAutograderCommit],
+        reviews: List[GitAutograderPrReview],
+        comments: List[GitAutograderPrComment],
+    ) -> None:
+        self._number = number
+        self._repo_full_name = repo_full_name
+        self._title = title
+        self._body = body
+        self._state = state  # e.g. OPEN, CLOSED, MERGED
+        self._author_login = author_login
+        self._base_branch = base_branch  # Branch targeted by the PR, e.g. main
+        self._head_branch = head_branch  # Branch where PR changes originate
+        self._is_draft = is_draft
+        self._merged_at = merged_at
+        self._merged_by_login = merged_by_login
+        self._created_at = created_at
+        self._commits = commits
+        self._reviews = reviews
+        self._comments = comments
+        self._user_reviews = [review for review in reviews if review.is_from_user()]
+        self._user_comments = [comment for comment in comments if comment.is_from_user()]
+        self._user_commits = [commit for commit in commits if commit.is_from_user()]
+
+    def __eq__(self, value: Any) -> bool:
+        if not isinstance(value, GitAutograderPr):
+            return False
+        return (
+            value.number == self.number
+            and value.repo_full_name == self.repo_full_name
+            and value.state == self.state
+        )
+
+    @property
+    def number(self) -> int:
+        return self._number
+
+    @property
+    def repo_full_name(self) -> str:
+        return self._repo_full_name
+
+    @property
+    def title(self) -> str:
+        return self._title
+
+    @property
+    def body(self) -> str:
+        return self._body
+
+    @property
+    def state(self) -> str:
+        return self._state
+
+    @property
+    def author_login(self) -> Optional[str]:
+        return self._author_login
+
+    @property
+    def base_branch(self) -> str:
+        return self._base_branch
+
+    @property
+    def head_branch(self) -> str:
+        return self._head_branch
+
+    @property
+    def is_draft(self) -> bool:
+        return self._is_draft
+
+    @property
+    def merged_at(self) -> Optional[str]:
+        return self._merged_at
+
+    @property
+    def merged_by_login(self) -> Optional[str]:
+        return self._merged_by_login
+
+    @property
+    def created_at(self) -> Optional[str]:
+        return self._created_at
+
+    @property
+    def reviews(self) -> List[GitAutograderPrReview]:
+        return self._reviews
+
+    @property
+    def comments(self) -> List[GitAutograderPrComment]:
+        return self._comments
+
+    @property
+    def commits(self) -> List[GitAutograderCommit]:
+        return self._commits
+
+    @property
+    def user_reviews(self) -> List[GitAutograderPrReview]:
+        return self._user_reviews
+
+    @property
+    def user_comments(self) -> List[GitAutograderPrComment]:
+        return self._user_comments
+
+    @property
+    def user_commits(self) -> List[GitAutograderCommit]:
+        return self._user_commits
+
+    @property
+    def last_user_review(self) -> GitAutograderPrReview:
+        if not self._user_reviews:
+            raise GitAutograderInvalidStateException("No user reviews found for this PR.")
+        return self._user_reviews[-1]
+
+    @property
+    def last_user_comment(self) -> GitAutograderPrComment:
+        if not self._user_comments:
+            raise GitAutograderInvalidStateException("No user comments found for this PR.")
+        return self._user_comments[-1]
+
+    @property
+    def last_user_commit(self) -> GitAutograderCommit:
+        if not self._user_commits:
+            raise GitAutograderInvalidStateException("No user commits found for this PR.")
+        return self._user_commits[-1]
+
+    def is_open(self) -> bool:
+        return self._state.upper() == "OPEN"
+
+    def is_closed(self) -> bool:
+        return self._state.upper() == "CLOSED"
+
+    def is_merged(self) -> bool:
+        return self._state.upper() == "MERGED"