From dff3aa7ead89f509c28005eb1f86bbc6572e8aa5 Mon Sep 17 00:00:00 2001 From: Jerry Zhang Date: Wed, 20 May 2026 16:58:57 -0700 Subject: [PATCH] github: Rework GraphQL batching and error handling to salvage partial work Restructure the GitHub GraphQL layer around a flat, resizable query so that partial results are never thrown away and failed work is re-transacted precisely. GraphqlQuery now holds a heterogeneous list of aliased fields (SingleQuery) with stable per-prefix alias indices, so it can split() in half or subset() to arbitrary fields without result aliases colliding. GithubQuery subclasses it with the repo-specific field builders and result parsers. endpoint.graphql no longer raises on GraphQL field errors. It returns a GraphqlResponse carrying partial data plus per-field errors, and only raises for a request error (200 with no data) or a non-retryable HTTP status. Retry backoff is now driven by GitHub's headers, honoring Retry-After and waiting out an exhausted rate-limit budget, and retries secondary-limit (403) as well as transient 5xx. Also recover a PR's id via a head-ref lookup when a create comes back "already exists", so a later update never targets an empty id, and log the remaining rate-limit budget at the end of a verbose run. --- revup/forge_utils.py | 6 + revup/github/endpoint.py | 147 +++--- revup/github/github.py | 707 +++++++++++++------------- revup/github/graphql.py | 288 +++++++++++ revup/types.py | 2 +- tests/test_github.py | 1043 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 1780 insertions(+), 413 deletions(-) create mode 100644 revup/github/graphql.py create mode 100644 tests/test_github.py diff --git a/revup/forge_utils.py b/revup/forge_utils.py index 077acaa..0d5ae5a 100644 --- a/revup/forge_utils.py +++ b/revup/forge_utils.py @@ -125,6 +125,12 @@ async def forge_connection( try: yield forge finally: + if endpoint.last_ratelimit_remaining is not None: + logging.debug( + "GitHub rate limit: {} points remaining, resets at {}".format( + endpoint.last_ratelimit_remaining, endpoint.last_ratelimit_reset + ) + ) await forge.close() else: raise RevupUsageException( diff --git a/revup/github/endpoint.py b/revup/github/endpoint.py index 2cbb9e5..f060047 100644 --- a/revup/github/endpoint.py +++ b/revup/github/endpoint.py @@ -7,10 +7,37 @@ from aiohttp import ClientSession, ContentTypeError +from revup.github.graphql import GraphqlResponse from revup.types import RevupForgeException, RevupRequestException +# HTTP statuses worth retrying: gateway/timeout (5xx) and secondary-rate-limit (403). TRANSIENT_STATUSES = frozenset({500, 502, 503, 504}) -RETRYABLE_GRAPHQL_ERRORS = frozenset({"RESOURCE_LIMITS_EXCEEDED"}) +SECONDARY_LIMIT_STATUS = 403 +# Cap how long we'll auto-sleep waiting for a rate limit to reset. A longer wait +# (primary budget exhausted) is surfaced to the user instead of hanging silently. +MAX_BACKOFF_SECONDS = 60.0 + + +def _backoff_delay(headers: Any, attempt: int, base_delay: float) -> float: + """Seconds to wait before retrying, driven by GitHub's rate-limit headers. + + Precedence: an explicit Retry-After, then waiting out an exhausted budget + (remaining == 0) until its reset, else plain exponential backoff. + """ + retry_after = headers.get("retry-after") + if retry_after is not None: + try: + return float(retry_after) + except ValueError: + pass + remaining = headers.get("x-ratelimit-remaining") + reset = headers.get("x-ratelimit-reset") + if remaining == "0" and reset is not None: + try: + return max(0.0, int(reset) - time.time()) + except ValueError: + pass + return base_delay * (2**attempt) class GitHubEndpoint: @@ -51,26 +78,20 @@ def __init__( self.oauth_token = oauth_token self.proxy = proxy self.graphql_endpoint = f"https://api.{github_url}/graphql" + # Rate-limit budget from the most recent response, for end-of-run reporting. + self.last_ratelimit_remaining: Optional[str] = None + self.last_ratelimit_reset: Optional[str] = None async def close(self) -> None: if self.session: await self.session.close() - async def _should_retry( - self, attempt: int, max_retries: int, base_delay: float, message: str - ) -> bool: - """Sleep with exponential backoff if retries remain. Returns True to retry.""" - if attempt >= max_retries - 1: - return False - delay = base_delay * (2**attempt) - logging.warning( - "{}, retrying in {}s (attempt {}/{})".format(message, delay, attempt + 1, max_retries) - ) - await asyncio.sleep(delay) - return True - - async def _graphql_once(self, query: str, **kwargs: Any) -> Any: - """Execute a single GraphQL request. Raises on any error.""" + async def _post(self, query: str, kwargs: Any) -> Tuple[int, Any, Any]: + """POST a GraphQL request. Returns (status, headers, body). + + No policy: never raises for HTTP status or GraphQL errors. body is the + parsed JSON, or None if the response wasn't JSON. + """ if self.session is None: self.session = ClientSession() @@ -92,56 +113,64 @@ async def _graphql_once(self, query: str, **kwargs: Any) -> Any: logging.debug( "Response status: {} took {}".format(resp.status, time.time() - start_time) ) - ratelimit_reset = resp.headers.get("x-ratelimit-reset") - if ratelimit_reset is not None: - reset_timestamp = datetime.datetime.fromtimestamp(int(ratelimit_reset)).isoformat() - else: - reset_timestamp = "Unknown" + reset = resp.headers.get("x-ratelimit-reset") + reset_str = ( + datetime.datetime.fromtimestamp(int(reset)).isoformat() + if reset is not None + else "Unknown" + ) + self.last_ratelimit_remaining = resp.headers.get("x-ratelimit-remaining") + self.last_ratelimit_reset = reset_str logging.debug( "Ratelimit: {} remaining, resets at {}".format( - resp.headers.get("x-ratelimit-remaining"), - reset_timestamp, + self.last_ratelimit_remaining, reset_str ) ) - - if resp.status != 200: - try: - r = await resp.json() - except (ValueError, ContentTypeError) as exc: - logging.warning("Response body:\n{}".format(await resp.text())) - raise RevupRequestException(resp.status, {}) from exc - raise RevupRequestException(resp.status, r) - try: - r = await resp.json() + body = await resp.json() + logging.debug("Response JSON:\n{}".format(json.dumps(body, indent=1))) except (ValueError, ContentTypeError): logging.warning("Response body:\n{}".format(await resp.text())) - raise - else: - pretty_json = json.dumps(r, indent=1) - logging.debug("Response JSON:\n{}".format(pretty_json)) - - if "errors" in r: - raise RevupForgeException(r["errors"]) - - return r + body = None + return resp.status, resp.headers, body async def graphql( - self, query: str, *, max_retries: int = 3, base_delay: float = 1.0, **kwargs: Any - ) -> Any: + self, + query: str, + *, + max_retries: int = 3, + base_delay: float = 1.0, + **kwargs: Any, + ) -> GraphqlResponse: + """Execute a GraphQL request, retrying transient failures with backoff. + + Returns a GraphqlResponse carrying partial data and per-field errors — a + 200 with field errors is NOT raised, so callers can salvage what resolved + and re-transact the rest. Raises RevupForgeException for a request error + (200 with no data) and RevupRequestException for a non-retryable HTTP error. + """ for attempt in range(max_retries): - try: - return await self._graphql_once(query, **kwargs) - except RevupRequestException as e: - if e.status not in TRANSIENT_STATUSES: - raise - msg = "GitHub returned {}".format(e.status) - if not await self._should_retry(attempt, max_retries, base_delay, msg): - raise - except RevupForgeException as e: - retryable = set(e.types) & RETRYABLE_GRAPHQL_ERRORS - if not retryable: - raise - msg = "GitHub GraphQL error ({})".format(", ".join(retryable)) - if not await self._should_retry(attempt, max_retries, base_delay, msg): - raise + status, headers, body = await self._post(query, kwargs) + + if status == 200: + if body is None: + raise RevupRequestException(status, {}) + # A request error (bad query, unresolvable variable) returns no data + # and can't be salvaged or retried; fail loudly. + if body.get("data") is None and "errors" in body: + raise RevupForgeException(body["errors"]) + return GraphqlResponse.parse(body) + + retryable = status in TRANSIENT_STATUSES or status == SECONDARY_LIMIT_STATUS + if not retryable or attempt >= max_retries - 1: + raise RevupRequestException(status, body if body is not None else {}) + + delay = min(_backoff_delay(headers, attempt, base_delay), MAX_BACKOFF_SECONDS) + logging.warning( + "GitHub returned {}, retrying in {}s (attempt {}/{})".format( + status, delay, attempt + 1, max_retries + ) + ) + await asyncio.sleep(delay) + + raise RevupRequestException(status, body if body is not None else {}) diff --git a/revup/github/github.py b/revup/github/github.py index 1ce147d..4fce78d 100644 --- a/revup/github/github.py +++ b/revup/github/github.py @@ -1,5 +1,5 @@ import logging -from typing import Any, Dict, Iterable, List, Optional, Set, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple from revup.forge import ( MAX_COMMENTS_TO_QUERY, @@ -10,151 +10,16 @@ PrUpdate, ) from revup.github.endpoint import GitHubEndpoint +from revup.github.graphql import ( + ErrorClass, + GraphqlError, + GraphqlOperation, + GraphqlQuery, + GraphqlResponse, +) from revup.types import RevupForgeException - -def _get_args_dict(args: List[Any], prefix: str) -> Dict[str, Any]: - return {f"{prefix}{n}": arg for n, arg in enumerate(args)} - - -def _get_args_declaration(args: Dict[str, Any], typ: str) -> List[str]: - return [f"${var}: {typ}" for var in args] - - -def _get_result_args(num: int, prefix: str) -> List[str]: - return [f"{prefix}{n}" for n in range(num)] - - -def _zip_and_flatten(l1: Iterable[str], l2: Iterable[str]) -> List[str]: - ret: List[str] = [] - iter1 = iter(l1) - iter2 = iter(l2) - while True: - try: - ret.append(next(iter1)) - ret.append(next(iter2)) - except StopIteration: - break - return ret - - -class Github(Forge): - def __init__( - self, - endpoint: GitHubEndpoint, - repo_info: ForgeRepoInfo, - fork_info: ForgeRepoInfo, - ): - self.endpoint = endpoint - self.repo_info = repo_info - self.fork_info = fork_info - - @property - def repo_owner(self) -> str: - return self.fork_info.owner - - @property - def repo_name(self) -> str: - return self.repo_info.name - - @property - def is_fork(self) -> bool: - return self.fork_info.owner != self.repo_info.owner - - async def close(self) -> None: - await self.endpoint.close() - - async def query_everything( - self, - head_refs: List[str], - user_ids: List[str], - labels: List[str], - teams: List[Tuple[str, str]], - ) -> Tuple[ - str, - List[Optional[PrInfo]], - Dict[str, str], - Dict[str, str], - Dict[str, str], - Dict[str, str], - Dict[str, Optional[Set[str]]], - ]: - head_refs_args = _get_args_dict(head_refs, "pr") - user_id_args = _get_args_dict(user_ids, "user") - label_args = _get_args_dict(labels, "label") - team_org_args = _get_args_dict([t[0] for t in teams], "team_org") - team_slug_args = _get_args_dict([t[1] for t in teams], "team_slug") - - prs_out = _get_result_args(len(head_refs), "pr_out") - user_id_out = _get_result_args(len(user_ids), "user_out") - label_out = _get_result_args(len(labels), "label_out") - team_out = _get_result_args(len(teams), "team_out") - - arg_str = ", ".join( - _get_args_declaration(head_refs_args, "String!") - + _get_args_declaration(user_id_args, "String!") - + _get_args_declaration(label_args, "String!") - + _get_args_declaration(team_org_args, "String!") - + _get_args_declaration(team_slug_args, "String!") - ) - - # NOTE: There are possible limitations here because we depend on PRs being - # returned in order of OPEN prs, followed by MERGED prs in the order that - # they merged. github doesn't offer these options and it is excessively - # expensive to always fetch multiple prs and order them on this side. For now - # we hope that the most relevant PR will have the most recent update time. - request_str = "".join( - len(head_refs) - * [ - "{}: pullRequests (headRefName: ${}, states: [OPEN, MERGED], first: 1, " - "orderBy: {{direction: DESC, field:UPDATED_AT}}) {{" - "...PrResult" - "}}," - ] - ) - request_str = request_str.format(*_zip_and_flatten(prs_out, head_refs_args.keys())) - - user_str = "".join( - len(user_ids) * ["{}: assignableUsers (query: ${}, first: 25) {{...UserResult}},"] - ) - user_str = user_str.format(*_zip_and_flatten(user_id_out, user_id_args.keys())) - - label_str = "".join(len(labels) * ["{}: label (name: ${}) {{...LabelResult}},"]) - label_str = label_str.format(*_zip_and_flatten(label_out, label_args.keys())) - - team_str = "" - for i in range(len(teams)): - team_str += ( - f"{team_out[i]}: organization(login: ${list(team_org_args.keys())[i]}) " - f"{{team(slug: ${list(team_slug_args.keys())[i]}) " - f"{{id, members(first: 100) {{nodes {{login}}, totalCount}}}}}}," - ) - - multi_query_str = f""" - query GetPrResults($owner: String!, $name: String!, {arg_str}) {{ - repository(name: $name, owner: $owner) {{ - id - {request_str}{user_str}{label_str} - }} - {team_str} - }}""" - if user_str: - multi_query_str += """ - fragment UserResult on UserConnection { - nodes { - login - id - } - totalCount - }""" - if label_str: - multi_query_str += """ - fragment LabelResult on Label { - id - name - }""" - if request_str: - multi_query_str += f""" +PR_FRAGMENT = f""" fragment PrResult on PullRequestConnection {{ nodes {{ id @@ -257,20 +122,96 @@ async def query_everything( totalCount }}""" - pr_result = await self.endpoint.graphql( - multi_query_str, - owner=self.repo_info.owner, - name=self.repo_info.name, - **head_refs_args, - **user_id_args, - **label_args, - **team_org_args, - **team_slug_args, - ) +USER_FRAGMENT = """ + fragment UserResult on UserConnection { + nodes { + login + id + } + totalCount + }""" +LABEL_FRAGMENT = """ + fragment LabelResult on Label { + id + name + }""" + + +# How many times to resend a subquery that makes no progress (lone too-big field +# or a repeated timeout) before treating its errors as fatal. +_MAX_STALLED_RETRIES = 2 + + +def _merge_data(into: Dict[str, Any], src: Any) -> None: + """Merge GraphQL `data` dict `src` into `into`, combining nested repository fields.""" + if not src: + return + for key, val in src.items(): + if key == "repository" and isinstance(into.get("repository"), dict) and val: + into["repository"].update(val) + else: + into[key] = val + + +class GithubQuery(GraphqlQuery): + """A GraphqlQuery with GitHub-specific field builders and result parsers.""" + + def add_pr_queries(self, head_refs: List[str]) -> None: + for ref in head_refs: + self.add( + prefix="pr", + scope="repo", + field_template=( + "{}: pullRequests (headRefName: {}, states: [OPEN, MERGED], first: 1, " + "orderBy: {{direction: DESC, field:UPDATED_AT}}) {{...PrResult}}," + ), + var_types=["String!"], + values=[ref], + fragment=PR_FRAGMENT, + ) + + def add_user_queries(self, user_ids: List[str]) -> None: + for uid in user_ids: + self.add( + prefix="user", + scope="repo", + field_template="{}: assignableUsers (query: {}, first: 25) {{...UserResult}},", + var_types=["String!"], + values=[uid], + fragment=USER_FRAGMENT, + ) + + def add_label_queries(self, labels: List[str]) -> None: + for label in labels: + self.add( + prefix="label", + scope="repo", + field_template="{}: label (name: {}) {{...LabelResult}},", + var_types=["String!"], + values=[label], + fragment=LABEL_FRAGMENT, + ) + + def add_team_queries(self, teams: List[Tuple[str, str]]) -> None: + for org, slug in teams: + self.add( + prefix="team", + scope="top", + field_template=( + "{}: organization(login: {}) " + "{{team(slug: {}) " + "{{id, members(first: 100) {{nodes {{login}}, totalCount}}}}}}," + ), + var_types=["String!", "String!"], + values=[org, slug], + ) + + def parse_prs(self, result: Any, head_refs: List[str]) -> List[Optional[PrInfo]]: + raw = self.extract(result, "pr") prs: List[Optional[PrInfo]] = [] for i, branch_name in enumerate(head_refs): - this_node = pr_result["data"]["repository"][prs_out[i]] + this_node = raw[i] if len(this_node["nodes"]) == 1: this_node = this_node["nodes"][0] pr_labels: Set[str] = set() @@ -297,7 +238,6 @@ async def query_everything( reviewers.add(requested["login"]) reviewer_ids.add(requested["id"]) for revs in this_node["latestReviews"]["nodes"]: - # Ignore self reviews and bot reviews (without a login) if not revs["viewerDidAuthor"] and "login" in revs["author"]: reviewers.add(revs["author"]["login"]) reviewer_ids.add(revs["author"]["id"]) @@ -305,10 +245,6 @@ async def query_everything( assignees.add(user["login"]) assignee_ids.add(user["id"]) - # The plain headRef and baseRef fields return the latest commit id associated with - # that branch name which may be newer than the PR itself if it was merged. We want - # the ids of the commits actually last associated with the PR, which we query from - # the commit list. This can also mean they are None if the PR has 0 commits. headRefOid = ( this_node["headCommit"]["nodes"][0]["commit"]["oid"] if this_node["headCommit"]["nodes"] @@ -367,11 +303,16 @@ async def query_everything( ) else: prs.append(None) + return prs + def parse_users( + self, result: Any, user_ids: List[str] + ) -> Tuple[Dict[str, str], Dict[str, str]]: + raw = self.extract(result, "user") names_to_ids: Dict[str, str] = {} names_to_logins: Dict[str, str] = {} for i, user_id in enumerate(user_ids): - this_node = pr_result["data"]["repository"][user_id_out[i]] + this_node = raw[i] if len(this_node["nodes"]) == 0: logging.warning("No matching user found for {}".format(user_id)) else: @@ -398,34 +339,210 @@ async def query_everything( user_id, shortest_name ) ) + return names_to_ids, names_to_logins + def parse_labels(self, result: Any, labels: List[str]) -> Dict[str, str]: + raw = self.extract(result, "label") labels_to_ids: Dict[str, str] = {} for i, label in enumerate(labels): - this_node = pr_result["data"]["repository"][label_out[i]] + this_node = raw[i] if this_node is not None: labels_to_ids[label] = this_node["id"] else: logging.warning("Couldn't find an existing label named {}".format(label)) + return labels_to_ids + def parse_teams( + self, result: Any, teams: List[Tuple[str, str]] + ) -> Tuple[Dict[str, str], Dict[str, Optional[Set[str]]]]: + raw = self.extract(result, "team") teams_to_ids: Dict[str, str] = {} teams_to_members: Dict[str, Optional[Set[str]]] = {} for i, (org, slug) in enumerate(teams): - team_node = pr_result["data"][team_out[i]] + team_node = raw[i] if team_node is not None and team_node["team"] is not None: team_ref = f"{org}/{slug}" teams_to_ids[team_ref] = team_node["team"]["id"] members_node = team_node["team"]["members"] member_logins = {m["login"] for m in members_node["nodes"]} if members_node["totalCount"] > len(members_node["nodes"]): - # Team has more members than we fetched; we can't check membership reliably. teams_to_members[team_ref] = None else: teams_to_members[team_ref] = member_logins else: logging.warning("Couldn't find a team matching {}/{}".format(org, slug)) + return teams_to_ids, teams_to_members + + +class Github(Forge): + def __init__( + self, + endpoint: GitHubEndpoint, + repo_info: ForgeRepoInfo, + fork_info: ForgeRepoInfo, + ): + self.endpoint = endpoint + self.repo_info = repo_info + self.fork_info = fork_info + + @property + def repo_owner(self) -> str: + return self.fork_info.owner + + @property + def repo_name(self) -> str: + return self.repo_info.name + + @property + def is_fork(self) -> bool: + return self.fork_info.owner != self.repo_info.owner + + async def close(self) -> None: + await self.endpoint.close() + + def _make_query_everything( + self, + head_refs: List[str], + user_ids: List[str], + labels: List[str], + teams: List[Tuple[str, str]], + ) -> GithubQuery: + q = GithubQuery(name="GetEverything") + q.add_fixed_var("owner", "String!", self.repo_info.owner) + q.add_fixed_var("name", "String!", self.repo_info.name) + q.fixed_repo_fields = "id\n" + + q.add_pr_queries(head_refs) + q.add_user_queries(user_ids) + q.add_label_queries(labels) + q.add_team_queries(teams) + + return q + + async def _run_once(self, q: GraphqlQuery) -> GraphqlResponse: + query_str, variables = q.build() + return await self.endpoint.graphql(query_str, **variables) + + async def _execute(self, q: GraphqlQuery) -> Dict[str, Any]: + """Run a query/mutation, salvaging partial results and re-transacting the rest. + + Returns the merged `data` for every field that ultimately succeeded. Fields + that GitHub couldn't compute (resource limits) are re-run — split smaller if a + field still fails alone. Fields whose effect already exists are treated as + done. Any other (fatal) error is raised. + + Only never-computed fields are ever re-sent, so mutations with real side + effects (created PRs, posted comments) are never re-executed. + """ + merged: Dict[str, Any] = {} + fatal: List[Any] = [] + # Queue of (subquery, attempts_without_progress). Start with the whole thing. + pending: List[Tuple[GraphqlQuery, int]] = [(q, 0)] + while pending: + sub, stalls = pending.pop() + if sub.total_items() == 0: + continue + resp = await self._run_once(sub) + _merge_data(merged, resp.data) + + # A field with a non-null result completed (for a mutation, its side + # effect applied); a null field did not. This is the source of truth for + # what to resubmit, so a completed mutation is never re-sent. + unfulfilled = sub.unfulfilled_aliases({"data": merged}) + errors_by_alias: Dict[str, GraphqlError] = {} + timed_out = False + for err in resp.errors: + if err.error_class is ErrorClass.INFORMATIONAL: + continue + if err.alias: + errors_by_alias[err.alias] = err + elif err.is_timeout: + timed_out = True # a whole-request timeout names no field + else: + fatal.append(err.raw) # request-level error with nothing to salvage + + resubmit: Set[str] = set() + too_big = False # a resource limit means the request must shrink, not just retry + for alias in unfulfilled: + field_err = errors_by_alias.get(alias) + if field_err is None: + # No per-field error: resubmit only if the whole request timed out + # before reaching it, else it's a legitimate null. + if timed_out: + resubmit.add(alias) + elif field_err.error_class is ErrorClass.ALREADY_DONE: + continue # effect already exists; nothing to redo + elif field_err.error_class is ErrorClass.RETRYABLE: + resubmit.add(alias) + too_big = True + elif field_err.is_timeout: + resubmit.add(alias) + else: + fatal.append(field_err.raw) + + if not resubmit: + continue + retry = sub.subset(resubmit) + made_progress = retry.total_items() < sub.total_items() + if too_big and not made_progress and retry.total_items() > 1: + # No progress and too big: halve to shrink the request. + left, right = retry.split() + logging.warning( + "Request too large, splitting {} fields into {} + {}".format( + retry.total_items(), left.total_items(), right.total_items() + ) + ) + pending.extend([(left, 0), (right, 0)]) + elif made_progress: + # Some fields completed; resubmit the remainder fresh. + pending.append((retry, 0)) + elif stalls < _MAX_STALLED_RETRIES: + # No progress (a lone too-big field, or a repeated timeout): retry a + # bounded number of times before giving up. + pending.append((retry, stalls + 1)) + else: + # Give up. Surface a fatal error for each field that never + # completed — synthesizing one for timeouts, which name no field — + # so a stalled request never silently drops work. + for alias in resubmit: + stalled = errors_by_alias.get(alias) + fatal.append( + stalled.raw + if stalled is not None + else {"message": "Request repeatedly failed to complete: {}".format(alias)} + ) + + if fatal: + raise RevupForgeException(fatal) + return {"data": merged} + + async def query_everything( + self, + head_refs: List[str], + user_ids: List[str], + labels: List[str], + teams: List[Tuple[str, str]], + ) -> Tuple[ + str, + List[Optional[PrInfo]], + Dict[str, str], + Dict[str, str], + Dict[str, str], + Dict[str, str], + Dict[str, Optional[Set[str]]], + ]: + q = self._make_query_everything(head_refs, user_ids, labels, teams) + + result = await self._execute(q) + + repo_id = result["data"]["repository"]["id"] + prs = q.parse_prs(result, head_refs) + names_to_ids, names_to_logins = q.parse_users(result, user_ids) + labels_to_ids = q.parse_labels(result, labels) + teams_to_ids, teams_to_members = q.parse_teams(result, teams) return ( - pr_result["data"]["repository"]["id"], + repo_id, prs, names_to_ids, names_to_logins, @@ -437,11 +554,7 @@ async def query_everything( async def create_pull_requests(self, repo_id: str, prs: List[PrInfo]) -> None: inputs = [] for pr in prs: - headRef = ( - pr.headRef - if self.fork_info.owner == self.repo_info.owner - else f"{self.fork_info.owner}:{pr.headRef}" - ) + headRef = pr.headRef if not self.is_fork else f"{self.fork_info.owner}:{pr.headRef}" inputs.append( { "baseRefName": pr.baseRef, @@ -453,39 +566,56 @@ async def create_pull_requests(self, repo_id: str, prs: List[PrInfo]) -> None: "draft": pr.is_draft, } ) - inputs_args = _get_args_dict(inputs, "pr") - prs_out = _get_result_args(len(inputs), "pr_out") - - arg_str = ", ".join(_get_args_declaration(inputs_args, "CreatePullRequestInput!")) - request_str = "".join( - len(inputs) - * [ - """ - {}: createPullRequest(input: ${}) {{ + q = GraphqlQuery(operation=GraphqlOperation.MUTATION) + for inp in inputs: + q.add( + prefix="pr", + scope="mutation", + field_template=""" + {}: createPullRequest(input: {}) {{ pullRequest {{ id url }} - }},""" - ] - ) - request_str = request_str.format(*_zip_and_flatten(prs_out, inputs_args.keys())) - - mutation_str = f""" - mutation ({arg_str}) {{ - {request_str} - }}""" + }},""", + var_types=["CreatePullRequestInput!"], + values=[inp], + ) - # Creating a pull request can fail if the branch is already merged. - pr_results = await self.endpoint.graphql(mutation_str, require_success=False, **inputs_args) + pr_results = await self._execute(q) + raw = q.extract(pr_results, "pr") for i, pr in enumerate(prs): - result = pr_results["data"][prs_out[i]]["pullRequest"] - if result is not None: - pr.id = result["id"] - pr.url = result["url"] + result_node = raw[i]["pullRequest"] if raw[i] is not None else None + if result_node is not None: + pr.id = result_node["id"] + pr.url = result_node["url"] + + # A create that came back "already exists" (from a prior partial run) leaves + # id/url unset. The PR does exist, so look it up by head ref — otherwise a + # downstream update would target an empty id and fail with NOT_FOUND. + missing = [pr for pr in prs if not pr.id] + if missing: + await self._populate_existing_pr_ids(missing) + + async def _populate_existing_pr_ids(self, prs: List[PrInfo]) -> None: + q = GithubQuery(name="FindExisting") + q.add_fixed_var("owner", "String!", self.repo_info.owner) + q.add_fixed_var("name", "String!", self.repo_info.name) + q.fixed_repo_fields = "id\n" + q.add_pr_queries([pr.headRef for pr in prs]) + + result = await self._execute(q) + for pr, node in zip(prs, q.extract(result, "pr")): + nodes = node["nodes"] if node else [] + if nodes: + pr.id = nodes[0]["id"] + pr.url = nodes[0]["url"] async def update_pull_requests(self, prs: List[PrUpdate]) -> None: + await self._execute(self._build_update_mutation(prs)) + + def _build_update_mutation(self, prs: List[PrUpdate]) -> GraphqlQuery: inputs = [] labels = [] reviewers = [] @@ -515,7 +645,6 @@ async def update_pull_requests(self, prs: List[PrUpdate]) -> None: "labelableId": pr.id, } ) - if pr.reviewer_ids or pr.reviewer_team_ids: reviewers.append( { @@ -534,7 +663,6 @@ async def update_pull_requests(self, prs: List[PrUpdate]) -> None: "assignableId": pr.id, } ) - if pr.is_draft is not None: if pr.is_draft: convert_to_draft.append( @@ -550,7 +678,6 @@ async def update_pull_requests(self, prs: List[PrUpdate]) -> None: "pullRequestId": pr.id, } ) - for c in pr.comments: if c.id: edit_comments.append( @@ -569,165 +696,39 @@ async def update_pull_requests(self, prs: List[PrUpdate]) -> None: } ) - inputs_args = _get_args_dict(inputs, "pr") - prs_out = _get_result_args(len(inputs), "pr_out") - - labels_args = _get_args_dict(labels, "label") - labels_out = _get_result_args(len(labels), "label_out") - - reviewers_args = _get_args_dict(reviewers, "rev") - reviewers_out = _get_result_args(len(reviewers), "rev_out") - - assignees_args = _get_args_dict(assignees, "asn") - assignees_out = _get_result_args(len(assignees), "asn_out") - - to_draft_args = _get_args_dict(convert_to_draft, "to_d") - to_draft_out = _get_result_args(len(convert_to_draft), "to_d_out") - - from_draft_args = _get_args_dict(convert_from_draft, "from_d") - from_draft_out = _get_result_args(len(convert_from_draft), "from_d_out") - - comments_args = _get_args_dict(comments, "com") - comments_out = _get_result_args(len(comments), "com_out") - - edit_comments_args = _get_args_dict(edit_comments, "edit_com") - edit_comments_out = _get_result_args(len(edit_comments), "edit_com_out") - - arg_str = ", ".join( - _get_args_declaration(inputs_args, "UpdatePullRequestInput!") - + _get_args_declaration(labels_args, "AddLabelsToLabelableInput!") - + _get_args_declaration(reviewers_args, "RequestReviewsInput!") - + _get_args_declaration(assignees_args, "AddAssigneesToAssignableInput!") - + _get_args_declaration(to_draft_args, "ConvertPullRequestToDraftInput!") - + _get_args_declaration(from_draft_args, "MarkPullRequestReadyForReviewInput!") - + _get_args_declaration(comments_args, "AddCommentInput!") - + _get_args_declaration(edit_comments_args, "UpdateIssueCommentInput!") - ) - - update_str = "".join( - len(inputs) - * [ - """ - {}: updatePullRequest(input: ${}) {{ - clientMutationId - }},""" - ] - ) - update_str = update_str.format(*_zip_and_flatten(prs_out, inputs_args.keys())) - - request_reviewers_str = "".join( - len(reviewers_args) - * [ - """ - {}: requestReviews(input: ${}) {{ - clientMutationId - }},""" - ] - ) - request_reviewers_str = request_reviewers_str.format( - *_zip_and_flatten(reviewers_out, reviewers_args.keys()) - ) - assignees_str = "".join( - len(assignees_args) - * [ - """ - {}: addAssigneesToAssignable(input: ${}) {{ + q = GraphqlQuery(operation=GraphqlOperation.MUTATION) + + def add_all(prefix: str, mutation: str, var_type: str, items: List[Any]) -> None: + for inp in items: + q.add( + prefix=prefix, + scope="mutation", + field_template=""" + {}: """ + + mutation + + """(input: {}) {{ clientMutationId - }},""" - ] - ) - assignees_str = assignees_str.format( - *_zip_and_flatten(assignees_out, assignees_args.keys()) - ) - - add_labels_str = "".join( - len(labels_args) - * [ - """ - {}: addLabelsToLabelable(input: ${}) {{ - clientMutationId - }},""" - ] - ) - add_labels_str = add_labels_str.format(*_zip_and_flatten(labels_out, labels_args.keys())) - - to_draft_str = "".join( - len(convert_to_draft) - * [ - """ - {}: convertPullRequestToDraft(input: ${}) {{ - clientMutationId - }},""" - ] - ) - to_draft_str = to_draft_str.format(*_zip_and_flatten(to_draft_out, to_draft_args.keys())) - - from_draft_str = "".join( - len(convert_from_draft) - * [ - """ - {}: markPullRequestReadyForReview(input: ${}) {{ - clientMutationId - }},""" - ] - ) - from_draft_str = from_draft_str.format( - *_zip_and_flatten(from_draft_out, from_draft_args.keys()) - ) - - add_comments_str = "".join( - len(comments_args) - * [ - """ - {}: addComment(input: ${}) {{ - clientMutationId - }},""" - ] - ) - add_comments_str = add_comments_str.format( - *_zip_and_flatten(comments_out, comments_args.keys()) - ) + }},""", + var_types=[var_type], + values=[inp], + ) - edit_comments_str = "".join( - len(edit_comments_args) - * [ - """ - {}: updateIssueComment(input: ${}) {{ - clientMutationId - }},""" - ] + add_all("com", "addComment", "AddCommentInput!", comments) + add_all("pr", "updatePullRequest", "UpdatePullRequestInput!", inputs) + add_all("rev", "requestReviews", "RequestReviewsInput!", reviewers) + add_all("asn", "addAssigneesToAssignable", "AddAssigneesToAssignableInput!", assignees) + add_all("label", "addLabelsToLabelable", "AddLabelsToLabelableInput!", labels) + add_all( + "to_d", "convertPullRequestToDraft", "ConvertPullRequestToDraftInput!", convert_to_draft ) - edit_comments_str = edit_comments_str.format( - *_zip_and_flatten(edit_comments_out, edit_comments_args.keys()) + add_all( + "from_d", + "markPullRequestReadyForReview", + "MarkPullRequestReadyForReviewInput!", + convert_from_draft, ) - - # Add comment mutations first to ensure comments are at the top of the PR - mutation_str = f""" - mutation ({arg_str}) {{ - {add_comments_str}{update_str}{request_reviewers_str}{assignees_str}{add_labels_str}\ -{to_draft_str}{from_draft_str}{edit_comments_str} - }}""" - - try: - await self.endpoint.graphql( - mutation_str, - **comments_args, - **inputs_args, - **reviewers_args, - **assignees_args, - **labels_args, - **to_draft_args, - **from_draft_args, - **edit_comments_args, - ) - except RevupForgeException as e: - if "timeout" in e.message: - logging.warning( - "Github update request timed out! Most likely this is a false alarm and changes" - " actually succeeded. You may want to rerun this command to verify." - ) - else: - raise e + add_all("edit_com", "updateIssueComment", "UpdateIssueCommentInput!", edit_comments) + return q async def query_pr_by_number(self, owner: str, name: str, number: int) -> Tuple[str, str]: result = await self.endpoint.graphql( @@ -744,5 +745,5 @@ async def query_pr_by_number(self, owner: str, name: str, number: int) -> Tuple[ name=name, number=number, ) - pr = result["data"]["repository"]["pullRequest"] + pr = result.data["repository"]["pullRequest"] return pr["headRefName"], pr["baseRefName"] diff --git a/revup/github/graphql.py b/revup/github/graphql.py new file mode 100644 index 0000000..5047e48 --- /dev/null +++ b/revup/github/graphql.py @@ -0,0 +1,288 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Optional, Set, Tuple + + +class GraphqlOperation(str, Enum): + """The top-level GraphQL operation type. str-valued so it renders as itself.""" + + QUERY = "query" + MUTATION = "mutation" + + +class ErrorClass(Enum): + """How a per-field GraphQL error should be handled on retry.""" + + # Field wasn't computed because the request was too expensive. Re-transacting + # just this field (alone or split smaller) can succeed. + RETRYABLE = "retryable" + # The intended effect already exists (a prior partial attempt applied it). + ALREADY_DONE = "already_done" + # An informational notice that accompanies an otherwise-successful response. + INFORMATIONAL = "informational" + # Won't succeed on retry (not found, no permission, malformed). + FATAL = "fatal" + + +# GraphQL error `type` values, mapped to how we treat the affected field. +# Anything unmapped is treated as FATAL (fail loudly rather than silently retry). +_ERROR_CLASS_BY_TYPE: Dict[str, ErrorClass] = { + "RESOURCE_LIMITS_EXCEEDED": ErrorClass.RETRYABLE, + # A mutation whose effect already exists (a PR/comment/etc from a partial retry). + "UNPROCESSABLE": ErrorClass.ALREADY_DONE, + # A warning that a field/id is deprecated; the response still succeeded. + "DEPRECATION": ErrorClass.INFORMATIONAL, +} + + +@dataclass +class GraphqlError: + """One entry from a GraphQL response `errors` array.""" + + message: str + type: str # "" if GitHub gave no type (usually a request/validation error) + path: List[Any] # alias-first path + raw: Dict[str, Any] # the original error object, for surfacing fatal errors + + @property + def error_class(self) -> ErrorClass: + return _ERROR_CLASS_BY_TYPE.get(self.type, ErrorClass.FATAL) + + @property + def is_timeout(self) -> bool: + # GitHub reports execution timeouts as a 200 body error ("We couldn't + # respond to your request in time") with no distinct type, so match message. + return "timeout" in self.message.lower() or "in time" in self.message.lower() + + @property + def alias(self) -> Optional[str]: + """The field alias this error is anchored to, if any. + + GitHub's path is alias-first for aliased fields (repo-scoped fields are not + prefixed with `repository`), so the alias is the first `*_out*` path element. + """ + for part in self.path: + if isinstance(part, str) and "_out" in part: + return part + return None + + +@dataclass +class GraphqlResponse: + """A parsed GraphQL response: partial data plus any per-field errors. + + Unlike an exception, this represents a response that may be partially usable — + `data` holds whatever resolved, `errors` explains what didn't. + """ + + data: Any + errors: List[GraphqlError] = field(default_factory=list) + + @classmethod + def parse(cls, raw: Any) -> GraphqlResponse: + errors = [ + GraphqlError( + message=e.get("message", ""), + type=e.get("type", ""), + path=e.get("path", []), + raw=e, + ) + for e in raw.get("errors", []) + ] + return cls(data=raw.get("data"), errors=errors) + + +@dataclass +class SingleQuery: + """One aliased field in a GraphQL operation, owned and managed by GraphqlQuery.""" + + prefix: str + scope: str # "repo" | "top" | "mutation" + field_template: str + var_types: List[str] + values: List[Any] + fragment: str + index: int + + @property + def alias(self) -> str: + return f"{self.prefix}_out{self.index}" + + def var_name(self, var_idx: int) -> str: + if len(self.var_types) == 1: + return f"{self.prefix}{self.index}" + return f"{self.prefix}{self.index}_{var_idx}" + + def render_field(self) -> str: + var_names = [f"${self.var_name(j)}" for j in range(len(self.var_types))] + return self.field_template.format(self.alias, *var_names) + + def render_declarations(self) -> List[str]: + return [f"${self.var_name(j)}: {vtype}" for j, vtype in enumerate(self.var_types)] + + def render_variables(self) -> Dict[str, Any]: + return {self.var_name(j): val for j, val in enumerate(self.values)} + + def extract(self, result: Any) -> Any: + if self.scope == "repo": + return result["data"]["repository"][self.alias] + return result["data"][self.alias] + + +class GraphqlQuery: + """Builds a GraphQL query/mutation from a flat list of aliased fields. + + Fields of different types (prefixes) coexist in one list, so split() can + divide the list in half regardless of type. GraphqlQuery is the only owner + of the field objects; callers add fields by value and read results back by + prefix, never touching the field objects directly. + """ + + def __init__(self, operation: GraphqlOperation = GraphqlOperation.QUERY, name: str = ""): + self.operation = operation + self.name = name + self.fixed_vars: List[Tuple[str, str, Any]] = [] + self.fixed_repo_fields: str = "" + self.queries: List[SingleQuery] = [] + # Per-prefix counter so each type's aliases are 0, 1, 2, ... and unique. + self._prefix_counts: Dict[str, int] = {} + + def add_fixed_var(self, name: str, gql_type: str, value: Any) -> None: + self.fixed_vars.append((name, gql_type, value)) + + def add( + self, + *, + prefix: str, + scope: str, + field_template: str, + var_types: List[str], + values: List[Any], + fragment: str = "", + ) -> None: + """Add a field. Its alias index is fixed here so results stay addressable + by prefix even after the query is split and its results merged.""" + assert len(values) == len(var_types) + idx = self._prefix_counts.get(prefix, 0) + self._prefix_counts[prefix] = idx + 1 + self.queries.append( + SingleQuery( + prefix=prefix, + scope=scope, + field_template=field_template, + var_types=list(var_types), + values=list(values), + fragment=fragment, + index=idx, + ) + ) + + def extract(self, result: Any, prefix: str) -> List[Any]: + """Return the result node for every field with the given prefix, in add order.""" + return [q.extract(result) for q in self.queries if q.prefix == prefix] + + def unfulfilled_aliases(self, result: Any) -> Set[str]: + """Aliases whose result is null or absent in `result`. + + For a mutation this means the field did not apply (a completed mutation + returns its payload); for a query it means the field did not resolve. + """ + missing: Set[str] = set() + for q in self.queries: + try: + node = q.extract(result) + except (KeyError, TypeError): + node = None + if node is None: + missing.add(q.alias) + return missing + + def total_items(self) -> int: + return len(self.queries) + + def build(self) -> Tuple[str, Dict[str, Any]]: + all_decls: List[str] = [] + variables: Dict[str, Any] = {} + + for name, gql_type, value in self.fixed_vars: + all_decls.append(f"${name}: {gql_type}") + variables[name] = value + + for q in self.queries: + all_decls.extend(q.render_declarations()) + variables.update(q.render_variables()) + + decl_str = ", ".join(all_decls) + name_str = f" {self.name}" if self.name else "" + + repo_fields = self.fixed_repo_fields + top_fields = "" + mutation_fields = "" + for q in self.queries: + rendered = q.render_field() + if q.scope == "repo": + repo_fields += rendered + elif q.scope == "top": + top_fields += rendered + else: + mutation_fields += rendered + + if self.operation == GraphqlOperation.QUERY: + body = "" + if repo_fields: + body += f""" + repository(name: $name, owner: $owner) {{ + {repo_fields} + }}""" + body += top_fields + query_str = f""" + {self.operation.value}{name_str} ({decl_str}) {{{body} + }}""" + else: + query_str = f""" + {self.operation.value}{name_str} ({decl_str}) {{ + {mutation_fields} + }}""" + + fragments = "" + seen: set = set() + for q in self.queries: + if q.fragment and q.fragment not in seen: + fragments += q.fragment + seen.add(q.fragment) + query_str += fragments + + return query_str, variables + + def _empty_clone(self) -> GraphqlQuery: + # type(self) so a subclass survives a split as the same type. + clone = type(self)(operation=self.operation, name=self.name) + clone.fixed_vars = list(self.fixed_vars) + clone.fixed_repo_fields = self.fixed_repo_fields + return clone + + def _with_fields(self, fields: List[SingleQuery]) -> GraphqlQuery: + clone = self._empty_clone() + clone.queries = fields + return clone + + def split(self) -> Tuple[GraphqlQuery, GraphqlQuery]: + """Split the flat field list in half. + + Each half may be heterogeneous in field type. Aliases are baked into each + field, so results from the two halves never collide when merged. + """ + # Round up so an odd count keeps the extra item on the left (and a lone + # item lands left with an empty right). + mid = (len(self.queries) + 1) // 2 + return self._with_fields(self.queries[:mid]), self._with_fields(self.queries[mid:]) + + def subset(self, aliases: Set[str]) -> GraphqlQuery: + """A query containing only the fields whose alias is in `aliases`. + + Fields keep their original alias index, so results merge back without + collision. Used to re-transact only the fields that failed retryably. + """ + return self._with_fields([q for q in self.queries if q.alias in aliases]) diff --git a/revup/types.py b/revup/types.py index 10c81d0..da1a651 100644 --- a/revup/types.py +++ b/revup/types.py @@ -70,7 +70,7 @@ class RevupShellException(Exception): class RevupForgeException(Exception): - def __init__(self, error_json: Dict): + def __init__(self, error_json: List[Dict]): super().__init__() self.error_json = error_json messages = [] diff --git a/tests/test_github.py b/tests/test_github.py new file mode 100644 index 0000000..4b46e44 --- /dev/null +++ b/tests/test_github.py @@ -0,0 +1,1043 @@ +"""Unit tests for the Github class with a mocked GraphQL endpoint.""" + +import asyncio +import time +from typing import Any, Dict, List +from unittest.mock import AsyncMock, patch + +import pytest + +from revup.forge import ForgeRepoInfo, PrComment, PrInfo, PrUpdate +from revup.github.endpoint import GitHubEndpoint, _backoff_delay +from revup.github.github import _MAX_STALLED_RETRIES, Github, _merge_data +from revup.github.graphql import GraphqlResponse +from revup.types import RevupForgeException, RevupRequestException + + +def gql(data, errors=None): + """Build the GraphqlResponse that endpoint.graphql now returns.""" + return GraphqlResponse.parse({"data": data, "errors": errors or []}) + + +def gql_raw(raw): + """Wrap a raw {data, errors} dict as the GraphqlResponse endpoint.graphql returns.""" + return GraphqlResponse.parse(raw) + + +def make_github(endpoint: GitHubEndpoint, fork_owner: str = "owner") -> Github: + return Github( + endpoint=endpoint, + repo_info=ForgeRepoInfo(owner="owner", name="repo"), + fork_info=ForgeRepoInfo(owner=fork_owner, name="repo"), + ) + + +def make_pr_node( + pr_id: str = "PR_1", + url: str = "https://github.com/owner/repo/pull/1", + state: str = "OPEN", + base_ref: str = "main", + head_oid: str = "abc123", + base_oid: str = "def456", + is_draft: bool = False, + reviewers: List[Dict] = None, + team_reviewers: List[Dict] = None, + latest_reviews: List[Dict] = None, + assignees: List[Dict] = None, + labels: List[Dict] = None, + comments: List[Dict] = None, + timeline_items: List[Dict] = None, +) -> Dict[str, Any]: + review_requests = [] + for r in reviewers or []: + review_requests.append({"requestedReviewer": r}) + for t in team_reviewers or []: + review_requests.append({"requestedReviewer": t}) + + return { + "nodes": [ + { + "id": pr_id, + "state": state, + "url": url, + "baseRefName": base_ref, + "body": "body", + "title": "title", + "isDraft": is_draft, + "baseCommit": {"nodes": [{"commit": {"parents": {"nodes": [{"oid": base_oid}]}}}]}, + "headCommit": {"nodes": [{"commit": {"oid": head_oid}}]}, + "reviewRequests": {"nodes": review_requests}, + "timelineItems": {"nodes": timeline_items or []}, + "latestReviews": {"nodes": latest_reviews or []}, + "assignees": {"nodes": assignees or []}, + "labels": {"nodes": labels or []}, + "comments": {"nodes": comments or []}, + } + ], + "totalCount": 1, + } + + +def make_user_node(login: str = "alice", node_id: str = "U_1") -> Dict[str, Any]: + return {"nodes": [{"login": login, "id": node_id}], "totalCount": 1} + + +def make_label_node(name: str = "bug", node_id: str = "L_1") -> Dict[str, Any]: + return {"id": node_id, "name": name} + + +def make_team_node(team_id: str = "T_1", members: List[str] = None) -> Dict[str, Any]: + if members is None: + members = ["alice"] + return { + "team": { + "id": team_id, + "members": { + "nodes": [{"login": m} for m in members], + "totalCount": len(members), + }, + } + } + + +class TestQueryEverything: + def test_parses_reviewers_and_teams(self): + """Verifies user reviewers, team reviewers, and latestReviews are all extracted.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node( + reviewers=[{"login": "alice", "id": "U_1"}], + team_reviewers=[ + { + "slug": "backend", + "id": "T_1", + "organization": {"login": "acme"}, + } + ], + latest_reviews=[ + { + "author": {"login": "bob", "id": "U_2"}, + "viewerDidAuthor": False, + }, + { + "author": {"login": "me", "id": "U_3"}, + "viewerDidAuthor": True, + }, + ], + ), + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["feat"], user_ids=[], labels=[], teams=[]) + ) + + pr = prs[0] + assert pr.reviewers == {"alice", "bob"} + assert pr.reviewer_ids == {"U_1", "U_2"} + assert pr.reviewer_teams == {"acme/backend"} + assert pr.reviewer_team_ids == {"T_1"} + + def test_parses_removed_reviewers_from_timeline(self): + """Removed review requests show up only if the user isn't currently a reviewer.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node( + reviewers=[{"login": "alice", "id": "U_1"}], + timeline_items=[ + {"requestedReviewer": {"login": "alice", "id": "U_1"}}, + {"requestedReviewer": {"login": "carol", "id": "U_3"}}, + {"assignee": {"login": "dave", "id": "U_4"}}, + ], + ), + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["feat"], user_ids=[], labels=[], teams=[]) + ) + + pr = prs[0] + # alice is still a reviewer so not in removed set + assert "alice" not in pr.removed_reviewers + assert pr.removed_reviewers == {"carol"} + assert pr.removed_reviewer_ids == {"U_3"} + assert pr.removed_assignees == {"dave"} + assert pr.removed_assignee_ids == {"U_4"} + + def test_parses_labels_assignees_comments(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node( + assignees=[{"login": "alice", "id": "U_1"}], + labels=[{"name": "urgent", "id": "L_1"}], + comments=[ + {"body": "hello", "id": "C_1"}, + {"body": "world", "id": "C_2"}, + ], + ), + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["feat"], user_ids=[], labels=[], teams=[]) + ) + + pr = prs[0] + assert pr.assignees == {"alice"} + assert pr.labels == {"urgent"} + assert pr.label_ids == {"L_1"} + assert len(pr.comments) == 2 + assert pr.comments[0] == PrComment("hello", "C_1") + + def test_draft_state_preserved(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node(is_draft=True, state="OPEN"), + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["feat"], user_ids=[], labels=[], teams=[]) + ) + + assert prs[0].is_draft is True + assert prs[0].state == "OPEN" + + def test_user_prefix_match_picks_shortest(self): + """When multiple users match, picks the shortest login that starts with the query.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "user_out0": { + "nodes": [ + {"login": "alice-long", "id": "U_long"}, + {"login": "alice", "id": "U_exact"}, + {"login": "alice-longer", "id": "U_longer"}, + ], + "totalCount": 3, + }, + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, _, names_to_ids, names_to_logins, _, _, _ = asyncio.run( + gh.query_everything(head_refs=[], user_ids=["alice"], labels=[], teams=[]) + ) + + assert names_to_ids["alice"] == "U_exact" + assert names_to_logins["alice"] == "alice" + + def test_user_no_prefix_match_falls_back_to_first(self): + """If no login starts with the query, falls back to first result.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "user_out0": { + "nodes": [ + {"login": "bob", "id": "U_bob"}, + {"login": "carol", "id": "U_carol"}, + ], + "totalCount": 2, + }, + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, _, names_to_ids, names_to_logins, _, _, _ = asyncio.run( + gh.query_everything(head_refs=[], user_ids=["al"], labels=[], teams=[]) + ) + + # Falls back to first node's id, but login NOT added to names_to_logins + assert names_to_ids["al"] == "U_bob" + assert "al" not in names_to_logins + + def test_team_with_too_many_members_returns_none(self): + """When totalCount > returned nodes, members set is None (can't enumerate all).""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": {"id": "R_1"}, + "team_out0": { + "team": { + "id": "T_1", + "members": { + "nodes": [{"login": "alice"}], + "totalCount": 150, + }, + } + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, _, _, _, _, teams_to_ids, teams_to_members = asyncio.run( + gh.query_everything(head_refs=[], user_ids=[], labels=[], teams=[("org", "big")]) + ) + + assert teams_to_ids == {"org/big": "T_1"} + assert teams_to_members["org/big"] is None + + def test_resource_limit_salvages_partial_and_retries_rest(self): + """First response resolves pr_out0 but nulls pr_out1 with a resource-limit + error. The retry must request ONLY pr_out1 (never re-request pr_out0).""" + endpoint = AsyncMock(spec=GitHubEndpoint) + requested_aliases = [] + + async def mock_graphql(query, **kwargs): + aliases = sorted(f"pr_out{k[2:]}" for k in kwargs if k.startswith("pr")) + requested_aliases.append(aliases) + if aliases == ["pr_out0", "pr_out1"]: + # Partial: pr_out0 resolved, pr_out1 hit the resource limit. + return gql( + { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node("PR_b1"), + "pr_out1": None, + } + }, + errors=[ + { + "type": "RESOURCE_LIMITS_EXCEEDED", + "path": ["pr_out1", "pullRequest"], + "message": "Resource limits for this query exceeded.", + } + ], + ) + # Retry of just pr_out1. + return gql({"repository": {"pr_out1": make_pr_node("PR_b2")}}) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + repo_id, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["b1", "b2"], user_ids=[], labels=[], teams=[]) + ) + + assert repo_id == "R_1" + assert prs[0].id == "PR_b1" + assert prs[1].id == "PR_b2" + # The retry requested only the failed alias, never re-requesting pr_out0. + assert requested_aliases == [["pr_out0", "pr_out1"], ["pr_out1"]] + + def test_fatal_field_error_raises(self): + """A non-retryable field error (NOT_FOUND) surfaces as a forge exception.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql( + {"repository": {"id": "R_1", "pr_out0": None}}, + errors=[{"type": "NOT_FOUND", "path": ["pr_out0"], "message": "not found"}], + ) + ) + gh = make_github(endpoint) + + with pytest.raises(RevupForgeException): + asyncio.run(gh.query_everything(head_refs=["b1"], user_ids=[], labels=[], teams=[])) + + def test_empty_pr_result(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": {"nodes": [], "totalCount": 0}, + } + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything(head_refs=["no-pr"], user_ids=[], labels=[], teams=[]) + ) + + assert prs == [None] + + def test_multiple_prs_each_mapped_to_correct_branch(self): + """Each head_ref maps to its corresponding PR result by index.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "repository": { + "id": "R_1", + "pr_out0": make_pr_node(pr_id="PR_A", base_ref="develop"), + "pr_out1": {"nodes": [], "totalCount": 0}, + "pr_out2": make_pr_node(pr_id="PR_C", base_ref="main"), + }, + } + } + ) + ) + gh = make_github(endpoint) + + _, prs, _, _, _, _, _ = asyncio.run( + gh.query_everything( + head_refs=["branch-a", "branch-b", "branch-c"], + user_ids=[], + labels=[], + teams=[], + ) + ) + + assert prs[0].id == "PR_A" + assert prs[0].headRef == "branch-a" + assert prs[0].baseRef == "develop" + assert prs[1] is None + assert prs[2].id == "PR_C" + assert prs[2].headRef == "branch-c" + + +class TestCreatePullRequests: + def test_fork_mode_prefixes_head_ref(self): + """When fork_info.owner differs from repo_info.owner, headRef gets owner: prefix.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "pr_out0": {"pullRequest": {"id": "PR_1", "url": "url1"}}, + } + } + ) + ) + gh = make_github(endpoint, fork_owner="myfork") + + pr = PrInfo( + baseRef="main", headRef="feat1", baseRefOid=None, headRefOid=None, body="b", title="t" + ) + asyncio.run(gh.create_pull_requests("R_1", [pr])) + + _, kwargs = endpoint.graphql.call_args + assert kwargs["pr0"]["headRefName"] == "myfork:feat1" + assert kwargs["pr0"]["baseRefName"] == "main" + + def test_same_owner_no_prefix(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + {"data": {"pr_out0": {"pullRequest": {"id": "PR_1", "url": "url1"}}}} + ) + ) + gh = make_github(endpoint, fork_owner="owner") + + pr = PrInfo( + baseRef="main", headRef="feat1", baseRefOid=None, headRefOid=None, body="b", title="t" + ) + asyncio.run(gh.create_pull_requests("R_1", [pr])) + + _, kwargs = endpoint.graphql.call_args + assert kwargs["pr0"]["headRefName"] == "feat1" + + def test_draft_flag_passed_through(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw({"data": {"pr_out0": {"pullRequest": {"id": "PR_1", "url": "u"}}}}) + ) + gh = make_github(endpoint) + + pr = PrInfo( + baseRef="main", + headRef="feat1", + baseRefOid=None, + headRefOid=None, + body="b", + title="t", + is_draft=True, + ) + asyncio.run(gh.create_pull_requests("R_1", [pr])) + + _, kwargs = endpoint.graphql.call_args + assert kwargs["pr0"]["draft"] is True + + def test_already_exists_recovers_id_by_head_ref(self): + """A create that comes back null (already exists) must recover the real PR id + via a head-ref lookup, so downstream updates don't target an empty id.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + calls = [] + + async def mock_graphql(query, **kwargs): + calls.append(query) + if "createPullRequest" in query: + # Already exists: GitHub returns null for the created pullRequest. + return gql( + {"pr_out0": {"pullRequest": None}}, + errors=[ + { + "type": "UNPROCESSABLE", + "path": ["pr_out0"], + "message": "A pull request already exists for owner:feat1.", + } + ], + ) + # The follow-up lookup by head ref finds the existing PR. + return gql( + {"repository": {"id": "R_1", "pr_out0": make_pr_node("PR_existing", url="u_ex")}} + ) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + pr = PrInfo( + baseRef="main", headRef="feat1", baseRefOid=None, headRefOid=None, body="b", title="t" + ) + asyncio.run(gh.create_pull_requests("R_1", [pr])) + + assert pr.id == "PR_existing" + assert pr.url == "u_ex" + assert len(calls) == 2 # the create, then the head-ref lookup + + def test_multiple_prs_batched_in_one_call(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql_raw( + { + "data": { + "pr_out0": {"pullRequest": {"id": "PR_1", "url": "u1"}}, + "pr_out1": {"pullRequest": {"id": "PR_2", "url": "u2"}}, + "pr_out2": {"pullRequest": {"id": "PR_3", "url": "u3"}}, + } + } + ) + ) + gh = make_github(endpoint) + + prs = [ + PrInfo( + baseRef="main", + headRef=f"f{i}", + baseRefOid=None, + headRefOid=None, + body="b", + title=f"t{i}", + ) + for i in range(3) + ] + asyncio.run(gh.create_pull_requests("R_1", prs)) + + endpoint.graphql.assert_called_once() + assert prs[0].id == "PR_1" + assert prs[1].id == "PR_2" + assert prs[2].id == "PR_3" + + +class TestUpdatePullRequests: + def test_builds_all_mutation_types(self): + """Labels, reviewers, assignees, draft conversion, comments all go in one mutation.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock(return_value=gql_raw({"data": {}})) + gh = make_github(endpoint) + + updates = [ + PrUpdate( + id="PR_1", + title="new title", + body="new body", + label_ids={"L_1"}, + reviewer_ids={"U_1"}, + reviewer_team_ids={"T_1"}, + assignee_ids={"U_2"}, + is_draft=True, + comments=[PrComment("new comment"), PrComment("edit me", "C_1")], + ), + ] + + asyncio.run(gh.update_pull_requests(updates)) + + _, kwargs = endpoint.graphql.call_args + assert kwargs["pr0"]["title"] == "new title" + assert kwargs["pr0"]["body"] == "new body" + assert kwargs["label0"]["labelIds"] == ["L_1"] + assert kwargs["rev0"]["userIds"] == ["U_1"] + assert kwargs["rev0"]["teamIds"] == ["T_1"] + assert kwargs["asn0"]["assigneeIds"] == ["U_2"] + assert kwargs["to_d0"]["pullRequestId"] == "PR_1" + assert kwargs["com0"]["body"] == "new comment" + assert kwargs["com0"]["subjectId"] == "PR_1" + assert kwargs["edit_com0"]["body"] == "edit me" + assert kwargs["edit_com0"]["id"] == "C_1" + + def test_ready_for_review_mutation(self): + """is_draft=False generates markPullRequestReadyForReview, not convertToDraft.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock(return_value=gql_raw({"data": {}})) + gh = make_github(endpoint) + + asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", is_draft=False)])) + + _, kwargs = endpoint.graphql.call_args + assert kwargs["from_d0"]["pullRequestId"] == "PR_1" + assert "to_d0" not in kwargs + + def test_mutation_timeout_resubmits_only_unapplied(self): + """On a whole-request timeout, fields that applied (non-null data) are kept and + only the unapplied (null) fields are resubmitted — no duplicate side effects.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + seen = [] + + async def mock_graphql(query, **kwargs): + prs = sorted(k for k in kwargs if k.startswith("pr")) + seen.append(prs) + if prs == ["pr0", "pr1"]: + # pr0 applied before the timeout; pr1 didn't. Timeout names no field. + return gql( + {"pr_out0": {"clientMutationId": "revup"}, "pr_out1": None}, + errors=[{"message": "We couldn't respond to your request in time"}], + ) + return gql({"pr_out1": {"clientMutationId": "revup"}}) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + asyncio.run( + gh.update_pull_requests( + [PrUpdate(id="PR_1", title="a"), PrUpdate(id="PR_2", title="b")] + ) + ) + # pr0 already applied, so only pr1 is resubmitted. + assert seen == [["pr0", "pr1"], ["pr1"]] + + def test_fatal_mutation_error_raises(self): + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock( + return_value=gql( + {"pr_out0": None}, + errors=[{"type": "FORBIDDEN", "path": ["pr_out0"], "message": "no permission"}], + ) + ) + gh = make_github(endpoint) + + with pytest.raises(RevupForgeException): + asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", title="x")])) + + def test_already_exists_mutation_is_not_retried(self): + """UNPROCESSABLE 'already exists' is treated as done: no raise, no re-send.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + call_count = 0 + + async def mock_graphql(query, **kwargs): + nonlocal call_count + call_count += 1 + return gql( + {"com_out0": None}, + errors=[ + { + "type": "UNPROCESSABLE", + "path": ["com_out0"], + "message": "A pull request already exists", + } + ], + ) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", comments=[PrComment("hi")])])) + assert call_count == 1 + + def test_resource_limit_salvages_and_retries_only_failed(self): + """A partial mutation re-sends only the resource-limited field, not the applied one.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + seen = [] + + async def mock_graphql(query, **kwargs): + prs = sorted(k for k in kwargs if k.startswith("pr")) + seen.append(prs) + if prs == ["pr0", "pr1"]: + return gql( + {"pr_out0": {"clientMutationId": "revup"}, "pr_out1": None}, + errors=[ + { + "type": "RESOURCE_LIMITS_EXCEEDED", + "path": ["pr_out1"], + "message": "too big", + } + ], + ) + return gql({"pr_out1": {"clientMutationId": "revup"}}) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + asyncio.run( + gh.update_pull_requests( + [PrUpdate(id="PR_1", title="a"), PrUpdate(id="PR_2", title="b")] + ) + ) + # First the full pair, then a retry of only the failed pr1 (pr0 not re-sent). + assert seen == [["pr0", "pr1"], ["pr1"]] + + def test_resource_limit_with_no_progress_splits_in_half(self): + """When nothing completes and the request is too big, it halves and each half + (down to a single field) succeeds separately.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + seen = [] + + async def mock_graphql(query, **kwargs): + prs = sorted(k for k in kwargs if k.startswith("pr")) + seen.append(prs) + if len(prs) > 1: + # Nothing completes; every field hits the resource limit. + return gql( + {f"pr_out{k[2:]}": None for k in prs}, + errors=[ + { + "type": "RESOURCE_LIMITS_EXCEEDED", + "path": [f"pr_out{k[2:]}"], + "message": "big", + } + for k in prs + ], + ) + return gql({f"pr_out{prs[0][2:]}": {"clientMutationId": "revup"}}) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + asyncio.run( + gh.update_pull_requests( + [PrUpdate(id="PR_1", title="a"), PrUpdate(id="PR_2", title="b")] + ) + ) + # Full pair fails wholesale, then each field is sent alone and succeeds. + assert seen[0] == ["pr0", "pr1"] + assert sorted(seen[1:]) == [["pr0"], ["pr1"]] + + def test_repeated_timeout_gives_up_after_stall_limit(self): + """A lone field that keeps timing out (no progress) is retried a bounded number + of times, then surfaced as a fatal error rather than looping forever.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + call_count = 0 + + async def mock_graphql(query, **kwargs): + nonlocal call_count + call_count += 1 + return gql( + {"pr_out0": None}, + errors=[{"message": "We couldn't respond to your request in time"}], + ) + + endpoint.graphql = mock_graphql + gh = make_github(endpoint) + + with pytest.raises(RevupForgeException): + asyncio.run(gh.update_pull_requests([PrUpdate(id="PR_1", title="a")])) + # Initial attempt + _MAX_STALLED_RETRIES resubmits, then it gives up. + assert call_count == 1 + _MAX_STALLED_RETRIES + + def test_multiple_prs_combined_into_single_mutation(self): + """Multiple PrUpdates with different fields all batch into one graphql call.""" + endpoint = AsyncMock(spec=GitHubEndpoint) + endpoint.graphql = AsyncMock(return_value=gql_raw({"data": {}})) + gh = make_github(endpoint) + + updates = [ + PrUpdate(id="PR_1", title="a", reviewer_ids={"U_1"}), + PrUpdate(id="PR_2", body="b", assignee_ids={"U_2"}), + ] + asyncio.run(gh.update_pull_requests(updates)) + + endpoint.graphql.assert_called_once() + _, kwargs = endpoint.graphql.call_args + assert kwargs["pr0"]["title"] == "a" + assert kwargs["pr1"]["body"] == "b" + assert kwargs["rev0"]["pullRequestId"] == "PR_1" + assert kwargs["asn0"]["assignableId"] == "PR_2" + + +class TestMergeData: + def test_merges_nested_repository_and_top_level(self): + merged = {"repository": {"id": "R_1", "pr_out0": "x"}} + _merge_data(merged, {"repository": {"pr_out1": "y"}, "team_out0": "z"}) + assert merged["repository"] == {"id": "R_1", "pr_out0": "x", "pr_out1": "y"} + assert merged["team_out0"] == "z" + + def test_merges_top_level_keys(self): + merged: dict = {} + _merge_data(merged, {"team_out0": "a"}) + _merge_data(merged, {"team_out1": "b"}) + assert merged == {"team_out0": "a", "team_out1": "b"} + + def test_none_src_is_noop(self): + merged = {"repository": {"id": "R_1"}} + _merge_data(merged, None) + assert merged == {"repository": {"id": "R_1"}} + + +def _add_field(q, prefix, values, scope="repo", field_template="{}: f(a: {}) {{id}},"): + q.add( + prefix=prefix, + scope=scope, + field_template=field_template, + var_types=["String!"], + values=list(values), + ) + + +class TestGraphqlQuerySplit: + def test_split_preserves_alias_indices(self): + """Aliases are baked in at add time, so merged results don't collide.""" + from revup.github.graphql import GraphqlQuery + + q = GraphqlQuery(name="Test") + q.add_fixed_var("owner", "String!", "o") + q.add_fixed_var("name", "String!", "r") + q.fixed_repo_fields = "id\n" + + for i in range(4): + _add_field(q, "pr", [f"val{i}"]) + + left, right = q.split() + left_query, left_vars = left.build() + right_query, right_vars = right.build() + + # Left gets items 0,1 with aliases pr_out0, pr_out1 + assert "pr_out0" in left_query + assert "pr_out1" in left_query + assert left_vars["pr0"] == "val0" + assert left_vars["pr1"] == "val1" + + # Right gets items 2,3 with aliases pr_out2, pr_out3 (index-preserved) + assert "pr_out2" in right_query + assert "pr_out3" in right_query + assert right_vars["pr2"] == "val2" + assert right_vars["pr3"] == "val3" + + def test_split_multi_var_query(self): + from revup.github.graphql import GraphqlQuery + + q = GraphqlQuery(name="Test") + tmpl = "{}: org(login: {}) {{team(slug: {})}}," + teams = [("org1", "slug1"), ("org2", "slug2"), ("org3", "slug3"), ("org4", "slug4")] + for org, slug in teams: + q.add( + prefix="team", + scope="top", + field_template=tmpl, + var_types=["String!", "String!"], + values=[org, slug], + ) + + left, right = q.split() + _, left_vars = left.build() + _, right_vars = right.build() + + assert left_vars == { + "team0_0": "org1", + "team0_1": "slug1", + "team1_0": "org2", + "team1_1": "slug2", + } + assert right_vars == { + "team2_0": "org3", + "team2_1": "slug3", + "team3_0": "org4", + "team3_1": "slug4", + } + + def test_single_item_cannot_split_further(self): + from revup.github.graphql import GraphqlQuery + + q = GraphqlQuery() + _add_field(q, "x", ["only"]) + + left, right = q.split() + assert left.total_items() == 1 + assert right.total_items() == 0 + + def test_split_is_heterogeneous(self): + """Split halves the flat list regardless of query type, so a half may mix types.""" + from revup.github.graphql import GraphqlQuery + + q = GraphqlQuery() + q.add_fixed_var("owner", "String!", "o") + q.add_fixed_var("name", "String!", "r") + + # Interleave two types so a per-type split couldn't separate them, but a + # flat split can. Order added: pr0, user0, pr1, user1, pr2, user2. + for i in range(3): + _add_field(q, "pr", [f"pr{i}"]) + _add_field(q, "user", [f"user{i}"], field_template="{}: u(q: {}) {{id}},") + + left, right = q.split() + assert left.total_items() == 3 + assert right.total_items() == 3 + + _, left_vars = left.build() + _, right_vars = right.build() + + # Left half is the first 3 added (pr0, user0, pr1) — heterogeneous. + assert left_vars["pr0"] == "pr0" + assert left_vars["user0"] == "user0" + assert left_vars["pr1"] == "pr1" + # Right half is the last 3 (user1, pr2, user2) — also heterogeneous. + assert right_vars["user1"] == "user1" + assert right_vars["pr2"] == "pr2" + assert right_vars["user2"] == "user2" + + +def make_endpoint() -> GitHubEndpoint: + return GitHubEndpoint(oauth_token="tok", github_url="github.com") + + +def scripted_post(*responses): + """An async _post replacement yielding (status, headers, body) tuples in order.""" + it = iter(responses) + + async def _post(query, kwargs): + return next(it) + + return _post + + +class TestEndpointGraphqlPolicy: + """endpoint.graphql retry/error policy, mocking the _post network seam.""" + + def test_200_with_partial_data_and_errors_returns_response(self): + ep = make_endpoint() + body = { + "data": {"repository": {"pr_out0": {"id": "PR_1"}}}, + "errors": [{"type": "RESOURCE_LIMITS_EXCEEDED", "path": ["pr_out1"], "message": "x"}], + } + ep._post = scripted_post((200, {}, body)) + + resp = asyncio.run(ep.graphql("query")) + assert isinstance(resp, GraphqlResponse) + assert resp.data["repository"]["pr_out0"] == {"id": "PR_1"} + assert resp.errors[0].type == "RESOURCE_LIMITS_EXCEEDED" + + def test_200_request_error_no_data_raises_forge(self): + ep = make_endpoint() + body = {"errors": [{"message": "Field 'x' doesn't exist"}]} + ep._post = scripted_post((200, {}, body)) + + with pytest.raises(RevupForgeException): + asyncio.run(ep.graphql("query")) + + def test_200_non_json_body_raises_request(self): + ep = make_endpoint() + ep._post = scripted_post((200, {}, None)) + + with pytest.raises(RevupRequestException): + asyncio.run(ep.graphql("query")) + + def test_transient_status_then_success_retries(self): + ep = make_endpoint() + ep._post = scripted_post((502, {}, None), (200, {}, {"data": {"ok": 1}})) + + with patch("asyncio.sleep", new=AsyncMock()) as sleep: + resp = asyncio.run(ep.graphql("query")) + assert resp.data == {"ok": 1} + sleep.assert_awaited_once() + + def test_secondary_limit_403_is_retried(self): + ep = make_endpoint() + ep._post = scripted_post((403, {}, None), (200, {}, {"data": {"ok": 1}})) + + with patch("asyncio.sleep", new=AsyncMock()): + resp = asyncio.run(ep.graphql("query")) + assert resp.data == {"ok": 1} + + def test_non_retryable_status_raises_immediately(self): + ep = make_endpoint() + ep._post = scripted_post((401, {}, {"message": "bad creds"})) + + with patch("asyncio.sleep", new=AsyncMock()) as sleep: + with pytest.raises(RevupRequestException) as exc: + asyncio.run(ep.graphql("query")) + assert exc.value.status == 401 + sleep.assert_not_awaited() # 401 is not retried + + def test_retries_exhausted_raises(self): + ep = make_endpoint() + ep._post = scripted_post((503, {}, None), (503, {}, None)) + + with patch("asyncio.sleep", new=AsyncMock()): + with pytest.raises(RevupRequestException) as exc: + asyncio.run(ep.graphql("query", max_retries=2)) + assert exc.value.status == 503 + + +class TestEndpointBackoffDelay: + def test_retry_after_takes_precedence(self): + # Even with a huge exponential attempt, an explicit Retry-After wins. + assert _backoff_delay({"retry-after": "7"}, attempt=5, base_delay=1.0) == 7.0 + + def test_exhausted_budget_waits_until_reset(self): + reset = int(time.time()) + 30 + delay = _backoff_delay( + {"x-ratelimit-remaining": "0", "x-ratelimit-reset": str(reset)}, + attempt=0, + base_delay=1.0, + ) + assert 25 <= delay <= 31 + + def test_falls_back_to_exponential(self): + assert _backoff_delay({}, attempt=0, base_delay=1.0) == 1.0 + assert _backoff_delay({}, attempt=3, base_delay=1.0) == 8.0 + + def test_malformed_retry_after_falls_through(self): + # A non-numeric Retry-After is ignored, falling back to exponential. + assert _backoff_delay({"retry-after": "soon"}, attempt=1, base_delay=1.0) == 2.0 + + def test_remaining_zero_without_reset_falls_through(self): + assert _backoff_delay({"x-ratelimit-remaining": "0"}, attempt=2, base_delay=1.0) == 4.0