#!/usr/bin/env python3 """ close.py — the state changes in Gitea, and the local file follows it or nothing happens at all. Two halves, and the second is the one that matters: 1. **It closes.** A slug, a number, several of either in one run, and `--reopen` going the other way. What goes out is a PATCH carrying `state` and nothing else; what comes back is written into `state:` on the local file, and the index is rebuilt so the store's own table agrees. 2. **It changes nothing local unless the tracker confirmed it.** A `tea` that exited non-zero, an answer with no number, an answer for another issue, an answer that still says `open`, an `origin: local` issue, a `--dry-run`: in every one of those the file on disk is byte for byte what it was. A bug here makes the store lie about the tracker, so each path is asserted on its own. The transport is stubbed at `_gitea.api`, as `test_drop_after_push.py` does, with the same deliberate exception: the non-2xx test stubs `_gitea.subprocess` and lets the real `_gitea.api` run, so "tea exited 1" is proved end to end. Nothing here touches a network, and nothing here touches the developer's store: every test builds its own in a `tempfile.TemporaryDirectory()`. """ import contextlib import io import json import os import sys import tempfile import types 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 close # noqa: E402 import issue # noqa: E402 import map as gmap # noqa: E402 # Captured before any test patches it — the non-2xx test needs the real thing. REAL_API = _gitea.api REPO = "claude-skills/tea" BASE = "repos/%s" % REPO BODY = """## Summary Прозаическое описание задачи. ## Spec skills/issue/references/format.md ## Acceptance criteria - [x] что-нибудь работает """ class FakeTracker(object): """`tea api` answered from memory, for state writes only. It keeps a `state` per number and flips it on a PATCH, which is the whole contract close.py has with the far side.""" def __init__(self): self.calls = [] self.states = {} # number -> "open" / "closed" self.raise_on_write = None # an exception instance to raise self.answer_override = None # what a write answers instead def payload_of(self, number): return {"number": number, "state": self.states[number], "title": "A thing", "updated_at": "2026-08-11T00:00:00Z", "html_url": "https://git.example/%s/issues/%d" % (REPO, number)} def writes(self): return [c for c in self.calls if c[0] != "GET"] def api(self, login, endpoint, method="GET", payload=None, payload_name=None, out_root=None, allow_fail=False): self.calls.append((method, endpoint, payload)) path = endpoint.split("?")[0] if "/issues/" in path and method == "PATCH": number = int(path.rsplit("/", 1)[1]) if self.raise_on_write is not None: raise self.raise_on_write self.states.setdefault(number, "open") if "state" in (payload or {}): self.states[number] = payload["state"] if self.answer_override is not None: return self.answer_override return self.payload_of(number) if "/issues/" in path and method == "GET": n = int(path.rsplit("/", 1)[1]) return self.payload_of(n) if n in self.states else None raise AssertionError("unstubbed call: %s %s" % (method, endpoint)) class StoreTestCase(unittest.TestCase): """A temp store and a fake tracker.""" def setUp(self): tmp = tempfile.TemporaryDirectory(prefix="tea-close-") self.addCleanup(tmp.cleanup) self.root = tmp.name self.fake = FakeTracker() for p in (mock.patch.object(_gitea, "api", self.fake.api), mock.patch.object(_gitea, "require_login", lambda: "test-login")): p.start() self.addCleanup(p.stop) # -- fixtures ---------------------------------------------------------- def synced(self, id="a-thing", number=101, state="open"): """An issue that is in the tracker and on disk, the way a pull leaves it: `origin: gitea`, a `gitea:` field, and a ledger entry.""" key = gmap.remote_key(REPO, number) iss = issue.Issue(id=id, title="A thing", body=BODY, state=state, labels=["type/task"], origin=gmap.ORIGIN, extra={"gitea": key, "url": "https://git.example/x", "synced": "2026-08-10T00:00:00Z"}) issue.save(self.root, iss) m = _gitea.load_map(self.root) m[key] = id _gitea.save_map(self.root, m) self.fake.states[number] = state return iss def local_only(self, id="local-thing"): """An issue that has never left this machine.""" iss = issue.Issue(id=id, title="Local thing", body=BODY, labels=["type/task"]) issue.save(self.root, iss) return iss def dropped(self, id="gone-thing", number=205, state="open"): """Pushed, and its file went with the push: ledger only.""" m = _gitea.load_map(self.root) m[gmap.remote_key(REPO, number)] = id _gitea.save_map(self.root, m) self.fake.states[number] = state return number # -- runner ------------------------------------------------------------ def run_close(self, *argv): self.out, self.err = io.StringIO(), io.StringIO() args = ["close.py", "--repo", REPO, "--out", self.root] + list(argv) with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(self.out), \ contextlib.redirect_stderr(self.err): close.main() return self.out.getvalue(), self.err.getvalue() # -- assertions -------------------------------------------------------- def state_on_disk(self, id): return issue.load(self.root, id).state def raw(self, id): with open(issue.path_of(self.root, id)) as f: return f.read() def assertUnchanged(self, id, before, why=""): self.assertEqual(self.raw(id), before, "%s.md was rewritten%s" % (id, why and " — " + why)) # -------------------------------------------------------------------------- # it closes # -------------------------------------------------------------------------- class ClosesTest(StoreTestCase): def test_a_slug_closes_the_issue_it_names(self): self.synced("a-thing", 101) out, _ = self.run_close("a-thing") self.assertEqual(self.fake.states[101], "closed") self.assertIn("closed a-thing #101", out) def test_the_local_state_follows(self): self.synced("a-thing", 101) self.run_close("a-thing") self.assertEqual(self.state_on_disk("a-thing"), "closed") def test_only_the_state_is_sent(self): """Closing is not an edit: no title, no body, no labels ride along.""" self.synced("a-thing", 101) self.run_close("a-thing") writes = self.fake.writes() self.assertEqual(len(writes), 1) method, endpoint, payload = writes[0] self.assertEqual((method, endpoint), ("PATCH", "%s/issues/101" % BASE)) self.assertEqual(payload, {"state": "closed"}) def test_a_number_closes_it_too(self): """The normal case for a pushed issue — the file is long gone.""" self.synced("a-thing", 101) self.run_close("101") self.assertEqual(self.fake.states[101], "closed") self.assertEqual(self.state_on_disk("a-thing"), "closed") def test_every_key_form_is_accepted(self): forms = {110: "110", 111: "#111", 112: "%s#112" % REPO, 113: "https://git.example/%s/issues/113" % REPO} for n in forms: self.fake.states[n] = "open" for n, arg in forms.items(): with self.subTest(arg=arg): self.run_close(arg) self.assertEqual(self.fake.states[n], "closed") def test_several_ids_in_one_run(self): self.synced("a-thing", 101) self.synced("b-thing", 102) self.run_close("a-thing", "102") self.assertEqual(self.fake.states, {101: "closed", 102: "closed"}) self.assertEqual(self.state_on_disk("a-thing"), "closed") self.assertEqual(self.state_on_disk("b-thing"), "closed") def test_the_same_issue_named_twice_is_written_once(self): self.synced("a-thing", 101) self.run_close("a-thing", "#101") self.assertEqual(len(self.fake.writes()), 1) def test_the_index_is_rebuilt(self): self.synced("a-thing", 101) out, _ = self.run_close("a-thing") self.assertIn("index:", out) with open(os.path.join(self.root, "INDEX.md")) as f: self.assertIn("closed", f.read()) def test_the_body_survives_untouched(self): """One metadata field changes; the prose and the ticks do not.""" self.synced("a-thing", 101) before = issue.load(self.root, "a-thing").body self.run_close("a-thing") self.assertEqual(issue.load(self.root, "a-thing").body, before) def test_synced_is_refreshed(self): self.synced("a-thing", 101) self.run_close("a-thing") iss = issue.load(self.root, "a-thing") self.assertNotEqual(iss.extra.get("synced"), "2026-08-10T00:00:00Z") self.assertEqual(iss.extra.get("remote-updated"), "2026-08-11T00:00:00Z") def test_an_issue_whose_file_was_dropped_still_closes(self): """No local copy at all: the ledger names it, the tracker takes it, and nothing is written locally.""" self.dropped("gone-thing", 205) out, _ = self.run_close("gone-thing") self.assertEqual(self.fake.states[205], "closed") self.assertIn("no local copy", out) self.assertNotIn("index:", out) def test_a_number_nobody_here_knows_closes_without_a_slug(self): self.fake.states[777] = "open" out, _ = self.run_close("777") self.assertEqual(self.fake.states[777], "closed") self.assertIn("#777", out) class ReopensTest(StoreTestCase): def test_reopen_sends_open(self): self.synced("a-thing", 101, state="closed") out, _ = self.run_close("--reopen", "a-thing") self.assertEqual(self.fake.writes()[0][2], {"state": "open"}) self.assertIn("reopened a-thing #101", out) def test_reopen_writes_the_local_state_back(self): self.synced("a-thing", 101, state="closed") self.run_close("--reopen", "a-thing") self.assertEqual(self.state_on_disk("a-thing"), "open") def test_close_then_reopen_is_a_round_trip(self): self.synced("a-thing", 101) self.run_close("a-thing") self.run_close("--reopen", "a-thing") self.assertEqual(self.fake.states[101], "open") self.assertEqual(self.state_on_disk("a-thing"), "open") # -------------------------------------------------------------------------- # it refuses # -------------------------------------------------------------------------- class LocalOnlyTest(StoreTestCase): """An `origin: local` issue is not in the tracker, so it cannot be closed there — and the local field is not quietly edited instead.""" def test_it_exits(self): self.local_only("local-thing") with self.assertRaises(SystemExit): self.run_close("local-thing") def test_the_error_names_the_id_and_says_it_is_not_in_the_tracker(self): self.local_only("local-thing") with self.assertRaises(SystemExit): self.run_close("local-thing") err = self.err.getvalue() self.assertIn("local-thing", err) self.assertIn("not in the tracker", err) def test_nothing_is_sent(self): self.local_only("local-thing") with self.assertRaises(SystemExit): self.run_close("local-thing") self.assertEqual(self.fake.calls, []) def test_the_file_is_untouched(self): self.local_only("local-thing") before = self.raw("local-thing") with self.assertRaises(SystemExit): self.run_close("local-thing") self.assertUnchanged("local-thing", before) def test_a_bad_id_stops_the_whole_run_before_anything_is_sent(self): """Resolution happens up front, so a typo in the second id does not leave the first one closed.""" self.synced("a-thing", 101) with self.assertRaises(SystemExit): self.run_close("a-thing", "local-thing") self.assertEqual(self.fake.states[101], "open") self.assertEqual(self.fake.calls, []) def test_an_unknown_slug_exits(self): with self.assertRaises(SystemExit): self.run_close("no-such-thing") self.assertIn("no-such-thing", self.err.getvalue()) class DryRunTest(StoreTestCase): def test_not_one_request_is_made(self): self.synced("a-thing", 101) self.run_close("--dry-run", "a-thing") self.assertEqual(self.fake.calls, []) def test_the_file_is_untouched(self): self.synced("a-thing", 101) before = self.raw("a-thing") self.run_close("--dry-run", "a-thing") self.assertUnchanged("a-thing", before, "--dry-run must write nothing") def test_it_says_what_would_be_closed(self): self.synced("a-thing", 101) self.synced("b-thing", 102) out, _ = self.run_close("--dry-run", "a-thing", "102") self.assertIn("would close a-thing #101", out) self.assertIn("would close b-thing #102", out) self.assertIn("2 issue(s) would be closed", out) def test_it_says_reopen_under_reopen(self): self.synced("a-thing", 101, state="closed") out, _ = self.run_close("--dry-run", "--reopen", "a-thing") self.assertIn("would reopen a-thing #101", out) self.assertIn("would be reopened", out) def test_it_needs_no_login(self): """A dry run must work before /tea:auth has ever been run.""" self.synced("a-thing", 101) with mock.patch.object(_gitea, "require_login", lambda: self.fail("dry run asked for a login")): self.run_close("--dry-run", "a-thing") def test_a_local_only_issue_is_still_refused(self): self.local_only("local-thing") with self.assertRaises(SystemExit): self.run_close("--dry-run", "local-thing") # -------------------------------------------------------------------------- # the tracker said no # -------------------------------------------------------------------------- class TrackerFailureTest(StoreTestCase): """The criterion that matters most: a write that was not confirmed leaves the local file exactly as it was.""" def test_a_non_2xx_answer_leaves_the_file(self): """The real `_gitea.api` against a `tea` that exits 1 — the path a 422 or a 500 actually takes, and it ends in `die()`.""" self.synced("a-thing", 101) before = self.raw("a-thing") def fake_run(cmd, capture_output=False, text=False): return types.SimpleNamespace( returncode=1, stdout="", stderr="422 Unprocessable Entity: issue is blocked") with mock.patch.object(_gitea, "api", REAL_API), \ mock.patch.object(_gitea, "subprocess", types.SimpleNamespace(run=fake_run)), \ self.assertRaises(SystemExit): self.run_close("a-thing") self.assertUnchanged("a-thing", before, "tea exited non-zero") self.assertEqual(self.state_on_disk("a-thing"), "open") def test_a_transport_exception_leaves_the_file(self): self.synced("a-thing", 101) before = self.raw("a-thing") self.fake.raise_on_write = OSError("tea: command not found") with self.assertRaises(OSError): self.run_close("a-thing") self.assertUnchanged("a-thing", before, "the transport raised") def test_an_answer_without_a_number_leaves_the_file(self): self.synced("a-thing", 101) before = self.raw("a-thing") self.fake.answer_override = {"ok": True, "state": "closed"} with self.assertRaises(SystemExit): self.run_close("a-thing") self.assertUnchanged("a-thing", before) def test_an_answer_for_another_issue_leaves_the_file(self): self.synced("a-thing", 101) before = self.raw("a-thing") self.fake.answer_override = {"number": 999, "state": "closed"} with self.assertRaises(SystemExit): self.run_close("a-thing") self.assertUnchanged("a-thing", before) def test_an_answer_that_did_not_change_the_state_leaves_the_file(self): """A 200 that still says `open` is not a close.""" self.synced("a-thing", 101) before = self.raw("a-thing") self.fake.answer_override = {"number": 101, "state": "open"} with self.assertRaises(SystemExit): self.run_close("a-thing") self.assertUnchanged("a-thing", before) def test_an_empty_answer_leaves_the_file(self): self.synced("a-thing", 101) before = self.raw("a-thing") self.fake.answer_override = None real_api = self.fake.api self.fake.api = lambda *a, **kw: (real_api(*a, **kw), None)[1] with mock.patch.object(_gitea, "api", self.fake.api), \ self.assertRaises(SystemExit): self.run_close("a-thing") self.assertUnchanged("a-thing", before) def test_the_error_says_nothing_local_changed(self): self.synced("a-thing", 101) self.fake.answer_override = {"ok": True} with self.assertRaises(SystemExit): self.run_close("a-thing") self.assertIn("Nothing local was changed", self.err.getvalue()) def test_a_failure_partway_through_keeps_the_rest(self): """Two issues, the second one is not confirmed. The first is legitimately closed; the second's file still says open.""" self.synced("aaa-thing", 101) self.synced("zzz-thing", 102) before = self.raw("zzz-thing") real = self.fake.api seen = [] def once(login, endpoint, method="GET", payload=None, **kw): got = real(login, endpoint, method, payload, **kw) if method != "GET": seen.append(endpoint) return {"nope": True} if len(seen) > 1 else got with mock.patch.object(_gitea, "api", once), \ self.assertRaises(SystemExit): self.run_close("aaa-thing", "zzz-thing") self.assertEqual(self.state_on_disk("aaa-thing"), "closed") self.assertUnchanged("zzz-thing", before, "its write was not confirmed") # -------------------------------------------------------------------------- # the pure parts # -------------------------------------------------------------------------- class ConfirmedTest(unittest.TestCase): """The gate itself. Everything below it rewrites a file.""" def test_a_matching_close_is_confirmed(self): self.assertTrue(close.confirmed({"number": 42, "state": "closed"}, 42, "closed")) def test_a_mismatched_number_is_not(self): self.assertFalse(close.confirmed({"number": 43, "state": "closed"}, 42, "closed")) def test_the_wrong_state_is_not(self): self.assertFalse(close.confirmed({"number": 42, "state": "open"}, 42, "closed")) def test_a_missing_state_is_not(self): self.assertFalse(close.confirmed({"number": 42}, 42, "closed")) def test_none_and_lists_are_not(self): self.assertFalse(close.confirmed(None, 42, "closed")) self.assertFalse(close.confirmed([{"number": 42, "state": "closed"}], 42, "closed")) def test_true_is_not_a_number(self): self.assertFalse(close.confirmed({"number": True, "state": "closed"}, 1, "closed")) def test_a_string_number_is_not(self): self.assertFalse(close.confirmed({"number": "42", "state": "closed"}, 42, "closed")) class KeyFormTest(unittest.TestCase): """A slug and a key are two vocabularies that must not collide.""" def test_keys_are_keys(self): for k in ("42", "#42", "owner/repo#42", "https://git.example/owner/repo/issues/42"): self.assertTrue(close.looks_like_key(k), k) def test_slugs_are_not_keys(self): for s in ("a-thing", "wire-sqlc-appclick", "close-issues-through-a-script"): self.assertFalse(close.looks_like_key(s), s) class LedgerPairsTest(unittest.TestCase): def setUp(self): self.m = {"%s#7" % REPO: "a-thing", "other/repo#7": "b-thing", "not-a-key": "c-thing"} def test_it_filters_by_repo(self): self.assertEqual(close.ledger_pairs(self.m, REPO), [(REPO, 7, "a-thing")]) def test_without_a_repo_it_keeps_everything_parseable(self): got = close.ledger_pairs(self.m) self.assertEqual(sorted(s for _r, _n, s in got), ["a-thing", "b-thing"]) def test_an_ambiguous_number_exits(self): pairs = close.ledger_pairs(self.m) with self.assertRaises(SystemExit): with contextlib.redirect_stderr(io.StringIO()): close.resolve("7", {}, pairs) class AmbiguityTest(StoreTestCase): """Two repos, one number, no --repo: settle it rather than guess.""" def test_the_error_points_at_repo(self): _gitea.save_map(self.root, {"%s#7" % REPO: "a-thing", "other/repo#7": "b-thing"}) err = io.StringIO() args = ["close.py", "--out", self.root, "7"] with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(io.StringIO()), \ contextlib.redirect_stderr(err), \ self.assertRaises(SystemExit): close.main() self.assertIn("--repo", err.getvalue()) class RepoOfTheKeyTest(StoreTestCase): """A key that names its own repo is sent there, not to whatever repo the CWD happens to be — otherwise `#42` closes somebody else's issue.""" def run_bare(self, *argv): """No `--repo`, so the ids have to say where they live.""" self.out, self.err = io.StringIO(), io.StringIO() args = ["close.py", "--out", self.root] + list(argv) with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(self.out), \ contextlib.redirect_stderr(self.err): close.main() return self.out.getvalue(), self.err.getvalue() def test_a_foreign_key_goes_to_its_own_repo(self): self.run_bare("other/repo#42") self.assertEqual(self.fake.writes()[0][1], "repos/other/repo/issues/42") def test_a_slug_goes_to_the_repo_its_gitea_field_names(self): self.synced("a-thing", 101) self.run_bare("a-thing") self.assertEqual(self.fake.writes()[0][1], "%s/issues/101" % BASE) def test_two_repos_in_one_run_is_a_question_not_a_guess(self): self.synced("a-thing", 101) with self.assertRaises(SystemExit): self.run_bare("a-thing", "other/repo#42") self.assertIn("one repo", self.err.getvalue()) self.assertEqual(self.fake.calls, []) def test_an_explicit_repo_settles_it(self): self.synced("a-thing", 101) self.run_close("a-thing", "other/repo#42") self.assertEqual({c[1] for c in self.fake.writes()}, {"%s/issues/101" % BASE, "%s/issues/42" % BASE}) class NoStoreTest(StoreTestCase): """A number needs no local file, and a store that is not there is not an error — closing an issue whose copy push dropped is the normal case.""" def test_a_number_closes_with_no_store_at_all(self): missing = os.path.join(self.root, "nowhere") self.fake.states[303] = "open" args = ["close.py", "--repo", REPO, "--out", missing, "303"] with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(io.StringIO()), \ contextlib.redirect_stderr(io.StringIO()): close.main() self.assertEqual(self.fake.states[303], "closed") self.assertFalse(os.path.isdir(missing), "no store was conjured") class PayloadFileTest(StoreTestCase): """The request body is written to .payload/, as every other write is.""" def test_the_payload_lands_beside_the_store(self): self.synced("a-thing", 101) with mock.patch.object(_gitea, "api", REAL_API), \ mock.patch.object( _gitea, "subprocess", types.SimpleNamespace(run=lambda cmd, **kw: types.SimpleNamespace( returncode=0, stderr="", stdout=json.dumps({"number": 101, "state": "closed"})))): self.run_close("a-thing") p = os.path.join(self.root, _gitea.PAYLOAD_DIR, "state-101.json") self.assertTrue(os.path.isfile(p)) with open(p) as f: self.assertEqual(json.load(f), {"state": "closed"}) if __name__ == "__main__": unittest.main()