"""Provider-shaped rework readback fixture with explicit resource ownership.""" import copy import hashlib import importlib.util import json import tempfile from argparse import Namespace from pathlib import Path from urllib.parse import parse_qs, urlsplit from harness.tests._support.paths import ROOT DIRECTORY = ROOT / "harness/agents/impl/skills/gitea-issue-rework/scripts" ORIGIN = "https://gitea.jsfamily.quest" T0 = "2026-01-01T00:00:00Z" T1 = "2026-01-02T00:00:00Z" class ReworkProviderFixture: """Own synthetic provider state and its temporary resources.""" def __init__(self): self._temp = None self._cleaned = False def setup(self): if self._temp is not None: raise RuntimeError("fixture already set up") self._temp = tempfile.TemporaryDirectory() root = Path(self._temp.name) self.legacy = self.load("rework_comment") data = { "owner": "acme", "repository": "demo", "issue": 7, "pr": 9, "review": 301, "reviewed_head": "a" * 40, "final_head": "b" * 40, "contract_sha256": "c" * 64, "result": "partial", "findings": [{"id": "F-1", "status": "resolved", "classification": "valid_fix", "changes": "bounded fix", "verification": "local checks pass", "reason": "regression"}], "local_checks": "pass", "runtime_browser": "not applicable", "commit_status": "success", "actions_runs": "없음", } (root / "input.json").write_text(json.dumps(data)) self.legacy.do_render(Namespace(input=str(root / "input.json"), no_delegated_evidence=True, delegation_adoption=None, comment=str(root / "comment.md"), expected=str(root / "expected.json"))) self.expected = json.loads((root / "expected.json").read_text()) self.contract = {"schema": "gitea-agent.rework-provider/v1", "version": "1.24.6", "aggregate": {"pr": {"field": "comments", "components": ["top_level", "reviews"]}, "pr_issue": {"field": "comments", "components": ["top_level", "reviews"]}}, "inline": {"mode": "per_review", "field": "review_comments", "components": ["reviews", "inline"]}} self.contract["aggregate"]["pr"]["components"] = ["top_level", "reviews", "inline"] self.contract["aggregate"]["pr_issue"]["components"] = ["top_level", "reviews", "inline"] self.repo = {"id": 3, "full_name": "acme/demo", "html_url": ORIGIN + "/acme/demo"} self.source = {"id": 707, "number": 7, "html_url": ORIGIN + "/acme/demo/issues/7", "body": "source contract", "title": "Source", "state": "open", "comments": 0, "updated_at": T0} self.pr = {"id": 99, "number": 9, "html_url": ORIGIN + "/acme/demo/pulls/9", "url": ORIGIN + "/acme/demo/pulls/9", "state": "open", "head": {"sha": "b" * 40, "repo": {"id": 12, "full_name": "contributor/demo"}}, "base": {"repo": copy.deepcopy(self.repo)}, "title": "PR", "body": "Closes #7", "comments": 1, "review_comments": 1, "updated_at": T0} self.underlying = {"id": 99, "number": 9, "html_url": self.pr["html_url"], "pull_request": {"html_url": self.pr["html_url"], "merged": False}, "title": "PR", "body": "Closes #7", "state": "open", "comments": 1, "updated_at": T0} self.reviews = [{"id": 301, "commit_id": "a" * 40, "comments_count": 0, "state": "REQUEST_CHANGES", "user": {"id": 33, "login": "gitea-review-agent"}}] self.comments = [] self.source_comments = [] self.inline = [] self.overrides = {} self.calls = [] self._request_override = None for record in (self.source, self.pr, self.underlying): record.update(created_at=T0, labels=[], assignees=[], milestone=None, user={"id": 22, "login": "genichin"} if record is self.source else {"id": 42, "login": "gitea-impl-agent"}) self.pr["base"].update(ref="main", label="main", sha="d" * 40, repo_id=3) self.pr["head"].update(ref="fix/issue-7", label="contributor:fix/issue-7", repo_id=12) self.reviews[0].update(body="Fix F-1", submitted_at=T0, updated_at=T0, stale=True, dismissed=False, html_url=self.pr["html_url"] + "#pullreview-301", pull_request_url=self.pr["html_url"]) return self def cleanup(self): if self._temp is not None and not self._cleaned: self._temp.cleanup() self._cleaned = True def load(self, name): path = DIRECTORY / (name + ".py") if not path.is_file(): raise AssertionError("canonical readback core missing") spec = importlib.util.spec_from_file_location(name, path) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def request(self, path): self.calls.append(path) if path in self.overrides: return copy.deepcopy(self.overrides[path]) parsed = urlsplit(path) objects = {"/user": {"id": 42, "login": "gitea-impl-agent"}, "/version": {"version": "1.24.6"}, "/repos/acme/demo": self.repo, "/repos/acme/demo/issues/7": self.source, "/repos/acme/demo/pulls/9": self.pr, "/repos/acme/demo/issues/9": self.underlying} if parsed.path in objects: return copy.deepcopy(objects[parsed.path]), {}, 200 collections = {"/repos/acme/demo/issues/7/comments": self.source_comments, "/repos/acme/demo/issues/9/comments": self.comments, "/repos/acme/demo/pulls/9/reviews": self.reviews, "/repos/acme/demo/pulls/9/reviews/301/comments": self.inline} if parsed.path not in collections: raise AssertionError("unexpected GET endpoint") rows = collections[parsed.path] page = int(parse_qs(parsed.query)["page"][0]) return copy.deepcopy(rows[(page - 1) * 50:page * 50]), {"X-Total-Count": str(len(rows))}, 200 def adapter(self): module = self.load("rework_readback") try: return module.ReworkReadbackAdapter(self._request_override or self.request, origin=ORIGIN + "/api/v1", credential_source="$HERMES_HOME/.env:GITEA_TOKEN", actor_id=42, provider_contract=self.contract) except Exception as error: raise AssertionError("canonical CLI transport contract rejected: " + str(error)) from error def add_result(self): self.comments.append({"id": 501, "html_url": self.pr["html_url"] + "#issuecomment-501", "body": self.expected["body"], "created_at": T1, "updated_at": T1, "user": {"id": 42, "login": "gitea-impl-agent"}}) self.pr["comments"] += 1 self.underlying["comments"] += 1 self.pr["updated_at"] = self.underlying["updated_at"] = T1