#!/usr/bin/env python3 """ Closed issues leave the store, and nothing else does. Two halves, and the second one is the one that matters: 1. **It evicts.** A closed issue whose `origin:` names a tracker is removed from `tmp/issues/` — the issue file and every sidecar under its slug — by one command, and `INDEX.md` is rebuilt so the directory and its table agree. `skills/sync/scripts/evict.py` does the same after refreshing `state:` from Gitea, so an issue closed in the web UI goes without a pull first. 2. **It evicts nothing else, ever.** `origin: local` is the only copy of the work there is: it stays in every state, including when it is closed and including when it is named on the command line. An open issue stays. A dry run stays. And a tracker call that fails leaves the whole store on disk — every candidate, not just the ones whose answers had not arrived yet. A bug in the second half destroys work, so each path is asserted separately and the assertion is always the same — `os.path.isfile`. Nothing here touches a network (the sync half stubs `_gitea.api`, and one test stubs `_gitea.subprocess` so a non-zero `tea` is proved end to end) and nothing here touches the developer's store: every test builds its own under `tempfile.TemporaryDirectory()`. """ import contextlib import io import os import shutil 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 evict # noqa: E402 import issue # noqa: E402 import issue_evict # noqa: E402 import map as gmap # noqa: E402 REAL_API = _gitea.api REPO = "claude-skills/tea" BODY = """## Summary Прозаическое описание задачи. ## Spec skills/issue/references/format.md ## Acceptance criteria - [x] сделано """ class StoreTestCase(unittest.TestCase): """A temp store, and fixtures for the three kinds of file that live in it.""" def setUp(self): self.root = tempfile.mkdtemp(prefix="tea-evict-") self.addCleanup(shutil.rmtree, self.root, True) self.numbers = {} # -- fixtures ---------------------------------------------------------- def local(self, id, state="open"): """An issue that exists nowhere but here.""" return self._write(id, state=state, origin=issue.LOCAL) def synced(self, id, state="open", number=None): """A working copy of something the tracker already has.""" n = number if number is not None else 100 + len(self.numbers) self.numbers[id] = n return self._write(id, state=state, origin=gmap.ORIGIN, extra={"gitea": gmap.remote_key(REPO, n), "url": "https://git.example/%s/issues/%d" % (REPO, n), "synced": "2026-08-10T00:00:00Z"}) def _write(self, id, state, origin, extra=None): iss = issue.Issue(id=id, title=id.replace("-", " ").capitalize(), body=BODY, labels=["type/task"], state=state, origin=origin, extra=dict(extra or {})) issue.save(self.root, iss) return iss def comments(self, id): p = _gitea.comments_path(self.root, id) with open(p, "w") as f: f.write("## comment 1 — someone — 2026-08-10\n\nтекст\n") return p # -- runners ----------------------------------------------------------- def run_evict(self, *argv): return self._run(issue_evict, "issue_evict.py", argv) def run_sync_evict(self, *argv): return self._run(evict, "evict.py", argv) def _run(self, mod, name, argv): self.out, self.err = io.StringIO(), io.StringIO() args = [name, "--out", self.root] + list(argv) with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(self.out), \ contextlib.redirect_stderr(self.err): mod.main() return self.out.getvalue(), self.err.getvalue() # -- assertions -------------------------------------------------------- def assertOnDisk(self, id, why=""): self.assertTrue(os.path.isfile(issue.path_of(self.root, id)), "%s.md was deleted%s" % (id, why and " — " + why)) def assertGone(self, id): self.assertFalse(os.path.isfile(issue.path_of(self.root, id)), "%s.md is still on disk" % id) def index(self): with open(os.path.join(self.root, "INDEX.md")) as f: return f.read() # -------------------------------------------------------------------------- # the domain: what belongs to a slug # -------------------------------------------------------------------------- class SlugFilesTest(StoreTestCase): """`issue.slug_files` — how the domain removes an issue completely without knowing what a comment thread is.""" def test_the_issue_file_comes_first(self): self.synced("a-thing") p = self.comments("a-thing") self.assertEqual(issue.slug_files(self.root, "a-thing"), [issue.path_of(self.root, "a-thing"), p]) def test_an_issue_with_no_sidecars_is_one_file(self): self.synced("a-thing") self.assertEqual(issue.slug_files(self.root, "a-thing"), [issue.path_of(self.root, "a-thing")]) def test_a_longer_slug_is_not_a_sidecar(self): """`a-thing-2` is another issue, not a companion of `a-thing`.""" self.synced("a-thing") self.synced("a-thing-2") self.assertEqual(issue.slug_files(self.root, "a-thing"), [issue.path_of(self.root, "a-thing")]) def test_a_missing_store_is_empty_not_an_error(self): self.assertEqual(issue.slug_files(os.path.join(self.root, "nope"), "x"), []) # -------------------------------------------------------------------------- # the domain: it evicts # -------------------------------------------------------------------------- class EvictsClosedTest(StoreTestCase): def test_a_closed_synced_issue_goes(self): self.synced("old-thing", state="closed") self.run_evict() self.assertGone("old-thing") def test_the_comment_thread_goes_with_it(self): self.synced("old-thing", state="closed") p = self.comments("old-thing") self.run_evict() self.assertFalse(os.path.isfile(p), "the thread outlived the issue") def test_the_store_of_open_and_closed_keeps_exactly_the_open_and_the_local(self): """The acceptance criterion, whole: a store of both kinds, one run, and what is left is the open issues and the local ones.""" self.synced("open-synced") self.synced("closed-synced", state="closed") self.local("open-local") self.local("closed-local", state="closed") self.run_evict() self.assertEqual(issue.all_ids(self.root), ["closed-local", "open-local", "open-synced"]) def test_the_output_names_every_file_removed(self): self.synced("old-thing", state="closed") p = self.comments("old-thing") out, _ = self.run_evict() self.assertIn("evicted", out) self.assertIn(issue.path_of(self.root, "old-thing"), out) self.assertIn(p, out) def test_the_index_is_rebuilt_to_match_the_directory(self): """`INDEX.md` and the directory agree afterwards — nothing to fix up.""" self.synced("old-thing", state="closed") self.synced("live-thing") self.run_evict() index = self.index() self.assertIn("live-thing", index) self.assertNotIn("old-thing", index) def test_only_the_named_issue_is_evicted(self): self.synced("first-old", state="closed") self.synced("second-old", state="closed") self.run_evict("first-old") self.assertGone("first-old") self.assertOnDisk("second-old", "it was not named") def test_the_ledger_is_not_pruned(self): """`.remote.json` is the number -> slug ledger, not an index over the files: an evicted issue is exactly as findable as a pushed one.""" self.synced("old-thing", state="closed") key = gmap.remote_key(REPO, self.numbers["old-thing"]) _gitea.save_map(self.root, {key: "old-thing"}) self.run_evict() self.assertEqual(_gitea.load_map(self.root), {key: "old-thing"}) class ClassifyTest(unittest.TestCase): """The decision itself, pure. Everything below it deletes a file.""" def issues(self, **kinds): return {id: issue.Issue(id=id, state=state, origin=origin) for id, (state, origin) in kinds.items()} def test_closed_and_synced_is_evicted(self): got = issue_evict.classify(self.issues(a=("closed", "gitea"))) self.assertEqual(got, (["a"], [], [])) def test_closed_and_local_is_protected(self): got = issue_evict.classify(self.issues(a=("closed", issue.LOCAL))) self.assertEqual(got, ([], ["a"], [])) def test_open_is_left_alone_whatever_its_origin(self): got = issue_evict.classify(self.issues(a=("open", "gitea"), b=("open", issue.LOCAL))) self.assertEqual(got, ([], [], ["a", "b"])) def test_naming_a_local_issue_does_not_make_it_evictable(self): got = issue_evict.classify(self.issues(a=("closed", issue.LOCAL)), ["a"]) self.assertEqual(got, ([], ["a"], [])) def test_ids_restrict_the_question(self): got = issue_evict.classify(self.issues(a=("closed", "gitea"), b=("closed", "gitea")), ["b"]) self.assertEqual(got, (["b"], [], [])) # -------------------------------------------------------------------------- # the domain: it evicts nothing else # -------------------------------------------------------------------------- class LocalIsNeverEvictedTest(StoreTestCase): """The criterion that matters most: `origin: local` IS the work.""" def test_a_closed_local_issue_stays(self): self.local("closed-local", state="closed") self.run_evict() self.assertOnDisk("closed-local", "origin: local is the only copy") def test_a_closed_local_issue_named_explicitly_still_stays(self): self.local("closed-local", state="closed") out, _ = self.run_evict("closed-local") self.assertOnDisk("closed-local", "naming it does not make deleting it safe") self.assertIn("kept", out) def test_the_receipt_says_why_it_was_kept(self): self.local("closed-local", state="closed") out, _ = self.run_evict() self.assertIn("origin: local", out) self.assertIn("this file IS the issue", out) def test_its_sidecars_stay_too(self): self.local("closed-local", state="closed") p = self.comments("closed-local") self.run_evict() self.assertTrue(os.path.isfile(p)) class DryRunTouchesNothingTest(StoreTestCase): def test_nothing_is_deleted(self): self.synced("old-thing", state="closed") p = self.comments("old-thing") self.run_evict("--dry-run") self.assertOnDisk("old-thing", "--dry-run must not delete") self.assertTrue(os.path.isfile(p)) def test_it_prints_what_would_go(self): self.synced("old-thing", state="closed") p = self.comments("old-thing") out, _ = self.run_evict("--dry-run") self.assertIn("would evict", out) self.assertIn(issue.path_of(self.root, "old-thing"), out) self.assertIn(p, out) self.assertIn("nothing was touched", out) def test_the_index_is_not_written(self): """`INDEX.md` is a write like any other — a dry run makes none.""" self.synced("old-thing", state="closed") self.run_evict("--dry-run") self.assertFalse(os.path.isfile(os.path.join(self.root, "INDEX.md"))) class NoOpRunsWriteNothingTest(StoreTestCase): def test_a_store_with_nothing_to_evict_is_not_rewritten(self): self.synced("live-thing") out, _ = self.run_evict() self.assertIn("0 issue(s) evicted", out) self.assertFalse(os.path.isfile(os.path.join(self.root, "INDEX.md"))) def test_an_unknown_id_stops_the_run(self): self.synced("old-thing", state="closed") with self.assertRaises(SystemExit): self.run_evict("no-such-thing") self.assertOnDisk("old-thing", "the run stopped before anything went") def test_a_missing_store_is_an_error_and_not_a_directory_to_create(self): missing = os.path.join(self.root, "nope") self.out, self.err = io.StringIO(), io.StringIO() argv = ["issue_evict.py", "--out", missing] with mock.patch.object(sys, "argv", argv), \ contextlib.redirect_stdout(self.out), \ contextlib.redirect_stderr(self.err), \ self.assertRaises(SystemExit): issue_evict.main() self.assertFalse(os.path.isdir(missing)) # -------------------------------------------------------------------------- # the bridge: the state comes from the tracker # -------------------------------------------------------------------------- class FakeTracker(object): """`tea api` answered from memory. GET on an issue, and nothing else.""" def __init__(self): self.calls = [] self.states = {} # number -> "open" | "closed" self.answer_override = {} # number -> whatever it should answer instead self.raise_on = None # number -> exception to raise instead def api(self, login, endpoint, method="GET", payload=None, payload_name=None, out_root=None, allow_fail=False): self.calls.append((method, endpoint)) number = int(endpoint.rstrip("/").rsplit("/", 1)[1]) if self.raise_on == number: raise OSError("tea: command not found") if number in self.answer_override: return self.answer_override[number] return {"number": number, "state": self.states.get(number, "open"), "title": "Whatever", "body": "текст"} class SyncEvictTestCase(StoreTestCase): def setUp(self): StoreTestCase.setUp(self) 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) def close_in_gitea(self, id): self.fake.states[self.numbers[id]] = "closed" def state_on_disk(self, id): return issue.load(self.root, id).state class TrackerStateWinsTest(SyncEvictTestCase): def test_an_issue_closed_upstream_is_evicted_without_a_pull_first(self): """The observed workflow, in one command: the file still says `open`.""" self.synced("old-thing", state="open") self.close_in_gitea("old-thing") self.run_sync_evict() self.assertGone("old-thing") def test_an_issue_still_open_upstream_stays(self): self.synced("live-thing", state="open") self.run_sync_evict() self.assertOnDisk("live-thing", "Gitea says it is open") def test_a_stale_closed_file_is_corrected_and_kept(self): """Reopened in the web UI: the local `state:` stops lying, and the file is not evicted on the strength of what it used to say.""" self.synced("back-thing", state="closed") self.run_sync_evict() self.assertOnDisk("back-thing", "Gitea says it is open again") self.assertEqual(self.state_on_disk("back-thing"), "open") def test_a_local_issue_is_never_asked_about(self): self.local("closed-local", state="closed") out, _ = self.run_sync_evict() self.assertEqual(self.fake.calls, []) self.assertOnDisk("closed-local") def test_an_issue_with_no_handle_is_reported_and_kept(self): """`origin: gitea` and nothing to reach it by: a guess would delete a file nobody can get back.""" issue.save(self.root, issue.Issue(id="orphan-thing", title="Orphan thing", body=BODY, labels=["type/task"], state="closed", origin=gmap.ORIGIN)) _, err = self.run_sync_evict() self.assertIn("orphan-thing", err) self.assertOnDisk("orphan-thing", "it could not be verified") def test_the_index_matches_the_directory_afterwards(self): self.synced("old-thing", state="open") self.synced("live-thing", state="open") self.close_in_gitea("old-thing") self.run_sync_evict() self.assertNotIn("old-thing", self.index()) self.assertIn("live-thing", self.index()) def test_dry_run_asks_but_neither_writes_nor_deletes(self): self.synced("old-thing", state="open") self.close_in_gitea("old-thing") out, _ = self.run_sync_evict("--dry-run") self.assertTrue(self.fake.calls, "it should still have asked") self.assertOnDisk("old-thing", "--dry-run must not delete") self.assertEqual(self.state_on_disk("old-thing"), "open", "--dry-run must not write the refreshed state either") self.assertIn("would evict", out) class SurvivesEveryTrackerFailureTest(SyncEvictTestCase): """A failed call evicts nothing — including the candidates whose answers had already arrived.""" def two_closed(self): self.synced("aaa-thing", state="closed", number=11) self.synced("zzz-thing", state="closed", number=12) self.close_in_gitea("aaa-thing") self.close_in_gitea("zzz-thing") def test_a_transport_exception_evicts_nothing(self): self.two_closed() self.fake.raise_on = 12 with self.assertRaises(OSError): self.run_sync_evict() self.assertOnDisk("aaa-thing", "its answer arrived, but the run failed") self.assertOnDisk("zzz-thing") def test_a_non_2xx_answer_evicts_nothing(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.two_closed() def fake_run(cmd, capture_output=False, text=False): return types.SimpleNamespace(returncode=1, stdout="", stderr="500 Internal Server Error") with mock.patch.object(_gitea, "api", REAL_API), \ mock.patch.object(_gitea, "subprocess", types.SimpleNamespace(run=fake_run)), \ self.assertRaises(SystemExit): self.run_sync_evict() self.assertOnDisk("aaa-thing", "tea exited non-zero") self.assertOnDisk("zzz-thing", "tea exited non-zero") def test_an_answer_for_another_issue_evicts_nothing(self): self.two_closed() self.fake.answer_override[12] = {"number": 999, "state": "closed"} with self.assertRaises(SystemExit): self.run_sync_evict() self.assertOnDisk("aaa-thing") self.assertOnDisk("zzz-thing", "the tracker answered for a different issue") def test_an_answer_without_a_state_evicts_nothing(self): self.two_closed() self.fake.answer_override[12] = {"number": 12} with self.assertRaises(SystemExit): self.run_sync_evict() self.assertOnDisk("aaa-thing") self.assertOnDisk("zzz-thing") def test_an_empty_answer_evicts_nothing(self): """`tea` exited 0 and printed nothing — api returns None.""" self.two_closed() self.fake.answer_override[12] = None with self.assertRaises(SystemExit): self.run_sync_evict() self.assertOnDisk("aaa-thing") self.assertOnDisk("zzz-thing") def test_the_error_says_nothing_was_evicted(self): self.two_closed() self.fake.answer_override[12] = {"ok": True} with self.assertRaises(SystemExit): self.run_sync_evict() self.assertIn("Nothing was evicted", self.err.getvalue()) def test_no_state_is_written_back_before_the_failure_either(self): """The write-back happens after every answer is in, so a run that dies leaves the files exactly as it found them.""" self.synced("aaa-thing", state="closed", number=11) self.synced("zzz-thing", state="closed", number=12) self.fake.states[11] = "open" # would be corrected on a good run self.fake.answer_override[12] = {"nope": True} with self.assertRaises(SystemExit): self.run_sync_evict() self.assertEqual(self.state_on_disk("aaa-thing"), "closed") class ConfirmedStateTest(unittest.TestCase): """The gate itself, in the shape of `push.confirmed_number`.""" def test_a_matching_answer_is_confirmed(self): self.assertEqual(evict.confirmed_state({"number": 42, "state": "closed"}, 42), "closed") self.assertEqual(evict.confirmed_state({"number": 42, "state": "open"}, 42), "open") def test_another_issue_is_not(self): self.assertIsNone(evict.confirmed_state({"number": 43, "state": "closed"}, 42)) def test_none_is_not(self): self.assertIsNone(evict.confirmed_state(None, 42)) def test_a_list_is_not(self): self.assertIsNone(evict.confirmed_state([{"number": 42, "state": "closed"}], 42)) def test_a_missing_state_is_not(self): self.assertIsNone(evict.confirmed_state({"number": 42}, 42)) def test_an_unknown_state_is_not(self): self.assertIsNone(evict.confirmed_state({"number": 42, "state": "merged"}, 42)) def test_a_string_number_is_not(self): self.assertIsNone(evict.confirmed_state({"number": "42", "state": "closed"}, 42)) def test_true_is_not_a_number(self): self.assertIsNone(evict.confirmed_state({"number": True, "state": "closed"}, 1)) class CandidatesTest(StoreTestCase): """Who the tracker is asked about at all.""" def test_a_synced_issue_is_asked_about_in_its_own_repo(self): self.synced("a-thing", number=7) checkable, unverifiable = evict.candidates(issue.load_all(self.root)) self.assertEqual(checkable, [("a-thing", REPO, 7)]) self.assertEqual(unverifiable, []) def test_a_local_issue_is_in_neither_list(self): self.local("local-thing", state="closed") self.assertEqual(evict.candidates(issue.load_all(self.root)), ([], [])) def test_a_handle_that_cannot_be_parsed_is_unverifiable(self): issue.save(self.root, issue.Issue(id="bad-thing", origin=gmap.ORIGIN, extra={"gitea": "not-a-key"})) checkable, unverifiable = evict.candidates(issue.load_all(self.root)) self.assertEqual(checkable, []) self.assertEqual([id for id, _ in unverifiable], ["bad-thing"]) if __name__ == "__main__": unittest.main()