#!/usr/bin/env python3 """ Native Gitea dependency links, written by push.py. The transport is stubbed at exactly one seam — `_gitea.api`, the single function that shells out to `tea` — so everything above it runs for real: argument parsing, validation, topological order, the id map, map.py's payload shapes and _gitea's own endpoint/body construction. Nothing here touches a network, and no test may ever be made to. `skills/*/scripts/` are not packages; they go on sys.path by hand. """ import contextlib import io import os import shutil import sys import tempfile import unittest from unittest import mock _ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) for _p in (os.path.join(_ROOT, "skills", "sync", "scripts"), os.path.join(_ROOT, "skills", "issue", "scripts")): if _p not in sys.path: sys.path.insert(0, _p) import _gitea # noqa: E402 import issue # noqa: E402 import map as gmap # noqa: E402 import push # noqa: E402 REPO = "claude-skills/tea" BASE = "repos/%s" % REPO LABELS = {"type/task": 901, "type/bug": 902, "severity/medium": 903, "comp/sync": 904} LABEL_NAMES = {v: k for k, v in LABELS.items()} BODY = """## Summary Прозаическое описание. ## Spec skills/issue/references/format.md ## Depends on - first-thing — ставит фундамент, без него второй не собрать ## Acceptance criteria - [ ] что-нибудь работает """ BODY_NO_DEPS = """## Summary Прозаическое описание. ## Spec skills/issue/references/format.md ## Acceptance criteria - [ ] что-нибудь работает """ class FakeGitea(object): """A `tea api` that answers from memory and remembers what it was asked. Dependency links are kept the way Gitea keeps them: per blocked issue, a set of (repo, number) blockers. That is what makes the idempotence test meaningful — the second push sees the link the first one made.""" def __init__(self, next_number=101): self.calls = [] # (method, endpoint, payload) self.next_number = next_number self.deps = {} # number -> {(repo, number)} self.titles = {} # number -> title self.fail_dependency_post = False # -- helpers ----------------------------------------------------------- @property def writes(self): """Every non-GET call. `--dry-run` must produce an empty list.""" return [c for c in self.calls if c[0] != "GET"] def dep_posts(self): return [c for c in self.calls if c[0] == "POST" and c[1].endswith("/dependencies")] def issue_payload(self, number, labels=()): return {"number": number, "html_url": "https://git.example/%s/issues/%d" % (REPO, number), "title": self.titles.get(number, ""), "labels": [{"name": LABEL_NAMES[i]} for i in labels if i in LABEL_NAMES], "updated_at": "2026-08-10T00:00:00Z", "repository": {"full_name": REPO}} # -- the seam ---------------------------------------------------------- def api(self, login, endpoint, method="GET", payload=None, payload_name=None, allow_fail=False): self.calls.append((method, endpoint, payload)) path = endpoint.split("?")[0] if path == "%s/labels" % BASE and method == "GET": # Every label the run could ask for, so nothing is ever created. return [{"name": n, "id": i} for n, i in LABELS.items()] if path == "%s/issues" % BASE and method == "POST": number = self.next_number self.next_number += 1 self.titles[number] = (payload or {}).get("title", "") # Echo the labels back, or push re-applies them with a PUT. return self.issue_payload(number, (payload or {}).get("labels") or []) if path.endswith("/dependencies"): number = int(path.split("/issues/")[1].split("/")[0]) if method == "GET": return [dict(self.issue_payload(n), repository={"full_name": r}) for r, n in sorted(self.deps.get(number, set()))] if method == "POST": if self.fail_dependency_post: return None key = ("%s/%s" % (payload["owner"], payload["repo"]), int(payload["index"])) self.deps.setdefault(number, set()).add(key) return self.issue_payload(number) if "/issues/" in path and method == "PATCH": number = int(path.rsplit("/", 1)[1]) self.titles[number] = (payload or {}).get("title", self.titles.get(number, "")) return self.issue_payload(number, (payload or {}).get("labels") or []) raise AssertionError("unstubbed call: %s %s" % (method, endpoint)) class PushTestCase(unittest.TestCase): """A temp store, a fake transport, and no git.""" def setUp(self): self.root = tempfile.mkdtemp(prefix="tea-store-") self.fake = FakeGitea() patches = [ mock.patch.object(_gitea, "api", self.fake.api), mock.patch.object(_gitea, "require_login", lambda: "test-login"), # push reads the current branch from git; a temp store has none and # the runner's branch would leak into the payload. mock.patch.object(push, "git_branch", lambda: "test-branch"), ] for p in patches: p.start() self.addCleanup(p.stop) self.addCleanup(shutil.rmtree, self.root, True) # -- fixtures ---------------------------------------------------------- def write_issue(self, id, title, body=BODY_NO_DEPS, depends=(), extra=None, origin=issue.LOCAL): iss = issue.Issue(id=id, title=title, body=body, labels=["type/task"], depends=list(depends), origin=origin, extra=dict(extra or {})) issue.save(self.root, iss) return iss def repull(self, id, body=BODY_NO_DEPS, depends=()): """Put a pushed issue back the way `pull.py` would. Push deletes the file, so anything that pushes the same issue twice has to fetch it in between — which is the workflow, not a test artifact. The slug and the number come from the ledger, exactly as `pull.id_for` would resolve them.""" number = self.number_of(id) self.assertIsNotNone(number, "%s was never pushed" % id) return self.write_issue(id, self.fake.titles[number], body=body, depends=depends, origin="gitea", extra={"gitea": "%s#%d" % (REPO, number)}) def two_issues(self): """first-thing, and second-thing which depends on it.""" self.write_issue("first-thing", "First thing") self.write_issue("second-thing", "Second thing", body=BODY, depends=["first-thing"]) def run_push(self, *argv): out, err = io.StringIO(), io.StringIO() args = ["push.py", "--repo", REPO, "--out", self.root] + list(argv) with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(out), \ contextlib.redirect_stderr(err): push.main() return out.getvalue(), err.getvalue() def number_of(self, id): """The number an id was pushed under, or None. Read off `.remote.json` rather than the issue file: a successful push deletes the file, and the ledger is what is left behind.""" for key, got in _gitea.load_map(self.root).items(): if got == id: return gmap.parse_remote_key(key)[1] return None # -------------------------------------------------------------------------- # _gitea: the POST body, and the pre-check that reads links back # -------------------------------------------------------------------------- class AddDependencyTest(unittest.TestCase): def test_post_body_is_issue_meta(self): """POST /issues/{index}/dependencies with IssueMeta for the BLOCKER. Confirmed against the instance's swagger.v1.json (Gitea 1.26.1): "Make the issue in the url depend on the issue in the form." """ calls = [] def fake_api(login, endpoint, method="GET", payload=None, **kw): calls.append((method, endpoint, payload)) return {"number": 102} with mock.patch.object(_gitea, "api", fake_api): ok = _gitea.add_dependency("l", BASE, 102, REPO, 101) self.assertTrue(ok) method, endpoint, payload = calls[0] self.assertEqual(method, "POST") self.assertEqual(endpoint, "%s/issues/102/dependencies" % BASE) self.assertEqual(payload, {"index": 101, "owner": "claude-skills", "repo": "tea"}) def test_blocker_may_live_in_another_repo(self): """IssueMeta carries owner/repo precisely so it can.""" seen = {} def fake_api(login, endpoint, method="GET", payload=None, **kw): seen.update(payload or {}) return {"number": 1} with mock.patch.object(_gitea, "api", fake_api): _gitea.add_dependency("l", BASE, 102, "other-org/infra", 7) self.assertEqual(seen, {"index": 7, "owner": "other-org", "repo": "infra"}) def test_failure_is_reported_not_raised(self): """409 (link already there) and friends come back as False.""" with mock.patch.object(_gitea, "api", lambda *a, **k: None): self.assertFalse(_gitea.add_dependency("l", BASE, 102, REPO, 101)) def test_unparseable_repo_makes_no_request(self): called = [] with mock.patch.object(_gitea, "api", lambda *a, **k: called.append(1)): self.assertFalse(_gitea.add_dependency("l", BASE, 102, "tea", 101)) self.assertEqual(called, []) def test_native_dep_pairs_reads_repo_and_number(self): payload = [{"number": 101, "repository": {"full_name": REPO}}, {"number": 7, "repository": {"full_name": "other-org/infra"}}] with mock.patch.object(_gitea, "api", lambda *a, **k: payload): got = _gitea.native_dep_pairs("l", BASE, 102) self.assertEqual(got, {(REPO, 101), ("other-org/infra", 7)}) def test_native_dep_pairs_empty_when_unsupported(self): with mock.patch.object(_gitea, "api", lambda *a, **k: None): self.assertEqual(_gitea.native_dep_pairs("l", BASE, 102), set()) # -------------------------------------------------------------------------- # push: the whole run # -------------------------------------------------------------------------- class PushCreatesLinksTest(PushTestCase): def test_link_created_after_both_have_numbers(self): """One run, topological order, one native link — no second pass.""" self.two_issues() out, _ = self.run_push() first, second = self.number_of("first-thing"), self.number_of("second-thing") self.assertLess(first, second, "blocker must be created first") self.assertEqual(self.fake.deps.get(second), {(REPO, first)}) self.assertIn("depends on %s#%d (first-thing)" % (REPO, first), out) def test_link_direction_matches_what_pull_reads_back(self): """The link hangs off the BLOCKED issue, which is where native_deps looks — push and `pull.py --deps` must agree or the round trip lies.""" self.two_issues() self.run_push() second = self.number_of("second-thing") with mock.patch.object(_gitea, "api", self.fake.api): self.assertEqual(_gitea.native_deps("l", BASE, second), [self.number_of("first-thing")]) def test_issue_without_dependencies_makes_no_dependency_request(self): """Not even the idempotence GET — it is skipped when there is nothing to link, so the common case costs no extra round trip.""" self.write_issue("lonely-thing", "Lonely thing") self.run_push() self.assertEqual([c for c in self.fake.calls if "dependencies" in c[1]], []) class LocalOnlyDependencyTest(PushTestCase): def test_local_dependency_is_warned_and_not_linked(self): self.two_issues() out, err = self.run_push("second-thing") self.assertEqual(self.fake.dep_posts(), []) self.assertIn("depends on local-only issue(s) first-thing", err) self.assertNotIn("depends on ", out) self.assertIsNone(self.number_of("first-thing")) class IdempotenceTest(PushTestCase): def test_repeat_push_does_not_duplicate_the_link(self): self.two_issues() self.run_push() self.assertEqual(len(self.fake.dep_posts()), 1) self.repull("first-thing") self.repull("second-thing", body=BODY, depends=["first-thing"]) self.run_push("--update") self.assertEqual(len(self.fake.dep_posts()), 1, "link re-POSTed") self.assertEqual(self.fake.deps[self.number_of("second-thing")], {(REPO, self.number_of("first-thing"))}) def test_a_failing_link_warns_and_the_run_finishes(self): """A 409 or any other refusal must not abort a push that has already created issues.""" self.two_issues() self.fake.fail_dependency_post = True out, err = self.run_push() self.assertIn("could not link", err) self.assertIn("index:", out) # the run completed self.assertIsNotNone(self.number_of("second-thing")) class UpdateCarriesNewLinksTest(PushTestCase): def test_dependency_added_after_the_first_push_is_linked_by_update(self): self.write_issue("first-thing", "First thing") self.write_issue("second-thing", "Second thing") self.run_push() self.assertEqual(self.fake.dep_posts(), []) # The issue comes back from Gitea, and the dependency is added to the # copy that came back — there is no other copy to add it to. self.repull("second-thing", body=BODY, depends=["first-thing"]) self.run_push("--update", "second-thing") self.assertEqual(self.fake.deps[self.number_of("second-thing")], {(REPO, self.number_of("first-thing"))}) class DryRunTest(PushTestCase): def test_dry_run_names_the_links_and_writes_nothing(self): self.two_issues() out, _ = self.run_push("--dry-run") self.assertEqual(self.fake.calls, [], "--dry-run made a request") self.assertIn("link -> #? (first-thing, created by this run)", out) self.assertIn("1 dependency link(s) would be created", out) def test_dry_run_shows_a_known_number_when_the_blocker_is_pushed(self): self.write_issue("first-thing", "First thing", extra={"gitea": "%s#101" % REPO}) self.write_issue("second-thing", "Second thing", body=BODY, depends=["first-thing"]) out, _ = self.run_push("--dry-run") self.assertIn("link -> %s#101 (first-thing)" % REPO, out) self.assertEqual(self.fake.writes, []) def test_dry_run_says_a_local_dependency_gets_no_link(self): self.two_issues() out, _ = self.run_push("--dry-run", "second-thing") self.assertIn("no link: first-thing is local-only", out) self.assertIn("0 dependency link(s) would be created", out) class BodyIsVerbatimTest(PushTestCase): """The prose is untouched. The id marker is the one thing push adds, and it comes straight back off — `strip_id_marker` is the inverse.""" def test_depends_on_prose_is_not_rewritten_to_numbers(self): """map.py deliberately never edits the prose. Linking must not start.""" self.two_issues() before = issue.load(self.root, "second-thing").body self.run_push() created = [c for c in self.fake.calls if c[0] == "POST" and c[1] == "%s/issues" % BASE] sent = [c[2]["body"] for c in created] second_body = [b for b in sent if "Depends on" in b][0] self.assertIn("- first-thing — ставит фундамент", second_body) self.assertNotIn("#101", second_body) self.assertEqual(gmap.strip_id_marker(second_body), before) def test_body_survives_a_second_push_unchanged(self): self.two_issues() before = issue.load(self.root, "second-thing").body self.run_push() self.repull("first-thing") self.repull("second-thing", body=before, depends=["first-thing"]) self.assertEqual(issue.load(self.root, "second-thing").body, before) self.run_push("--update") patched = [c for c in self.fake.calls if c[0] == "PATCH"] self.assertIn(before, [gmap.strip_id_marker(c[2]["body"]) for c in patched]) class DepStateTest(PushTestCase): """The classifier both the dry run and the real run read from.""" def test_classifies_linked_in_run_and_local(self): issues = { "pushed": issue.Issue(id="pushed", extra={"gitea": "%s#101" % REPO}), "coming": issue.Issue(id="coming"), "local": issue.Issue(id="local"), } iss = issue.Issue(id="dependent", depends=["pushed", "coming", "local", "ghost"]) got = push.dep_state(iss, issues, {"coming", "dependent"}) self.assertEqual(got, [("pushed", "%s#101" % REPO, False), ("coming", None, True), ("local", None, False)]) if __name__ == "__main__": unittest.main()