claude-skills/marketplace
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects 1 Releases Wiki Activity
Files
main
marketplace/plugins/tea/tests/test_evict_closed.py
T
naudachu 83f73c5cea refactor: turn the repo into a two-plugin marketplace 
tea and tdl were two repositories, each carrying its own
.claude-plugin/marketplace.json — two marketplaces to register for what
is one collection. Fold them into one.

The repo root is now the marketplace and nothing else: a single
.claude-plugin/marketplace.json whose entries point at ./plugins/tea and
./plugins/tdl. A plugin's root is its own directory under plugins/, so
${CLAUDE_PLUGIN_ROOT} still resolves inside it and every path a plugin
uses stays relative to itself — the hooks and the test roots needed no
adjustment beyond the move.

tea's files move with git mv, so its history and blame follow. tdl
arrives as a plain copy; its history stays in claude-skills/threedotslab.

test_payload_root asserted `tmp/` was ignored by REPO/.gitignore. The
rule is that tmp/ is ignored, not which file says so, and git reads every
.gitignore on the way up — so the test now walks up to the repo root the
same way git does.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-11 00:25:28 +05:00

571 lines
23 KiB
Python
Raw Permalink Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink