#!/usr/bin/env python3 """ Checkbox state survives a pull; everything else in the body does not. Two levels, on purpose. `map.merge_checkbox_state` is pure, so most of the rule is pinned down with plain strings and no store anywhere. The pull tests then prove the rule is actually wired into the write path, with the transport stubbed at the one seam `test_push_dependencies.py` uses — `_gitea.api`, the single function that shells out to `tea`. 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 re import shutil import sys import tempfile import unittest import urllib.parse 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 pull # noqa: E402 REPO = "claude-skills/tea" BASE = "repos/%s" % REPO def body(*criteria, **kw): """A body in the canonical shape, with the given `## Acceptance criteria`.""" summary = kw.get("summary", "Прозаическое описание.") return ("## Summary\n%s\n\n## Spec\nnone\n\n## Acceptance criteria\n%s\n" % (summary, "\n".join(criteria))).strip() # -------------------------------------------------------------------------- # the rule itself — pure, no store, no tracker # -------------------------------------------------------------------------- class MergeCheckboxStateTest(unittest.TestCase): """`[x]` wins from whichever side has it, for a matching item text.""" def test_a_local_tick_survives_the_overwrite(self): got = gmap.merge_checkbox_state(body("- [ ] первое", "- [ ] второе"), body("- [x] первое", "- [ ] второе")) self.assertEqual(got, body("- [x] первое", "- [ ] второе")) def test_a_remote_tick_is_kept(self): got = gmap.merge_checkbox_state(body("- [x] первое", "- [ ] второе"), body("- [ ] первое", "- [ ] второе")) self.assertEqual(got, body("- [x] первое", "- [ ] второе")) def test_both_sides_ticked_is_still_ticked(self): one = body("- [x] первое") self.assertEqual(gmap.merge_checkbox_state(one, one), one) def test_the_union_is_taken_item_by_item(self): got = gmap.merge_checkbox_state( body("- [x] первое", "- [ ] второе", "- [ ] третье"), body("- [ ] первое", "- [x] второе", "- [ ] третье")) self.assertEqual(got, body("- [x] первое", "- [x] второе", "- [ ] третье")) def test_an_item_the_local_copy_does_not_have_comes_from_the_server(self): """Including its state — both states, in both directions.""" got = gmap.merge_checkbox_state( body("- [x] новое сверху", "- [ ] новое снизу"), body("- [x] что-то совсем другое")) self.assertEqual(got, body("- [x] новое сверху", "- [ ] новое снизу")) def test_prose_is_not_merged(self): got = gmap.merge_checkbox_state( body("- [ ] пункт", summary="Новый текст с сервера."), body("- [x] пункт", summary="Старый локальный текст.")) self.assertIn("Новый текст с сервера.", got) self.assertNotIn("Старый локальный текст.", got) self.assertIn("- [x] пункт", got) def test_a_heading_the_local_copy_added_is_gone(self): remote = body("- [x] пункт") got = gmap.merge_checkbox_state(remote, remote + "\n\n## Notes\nмои заметки\n") self.assertEqual(got, remote) def test_no_local_copy_returns_the_server_body_untouched(self): remote = body("- [ ] пункт") for local in (None, "", " \n"): self.assertIs(gmap.merge_checkbox_state(remote, local), remote, "local_body=%r rewrote the body" % local) def test_nothing_ticked_locally_returns_the_same_object(self): remote = body("- [ ] пункт") self.assertIs(gmap.merge_checkbox_state(remote, body("- [ ] пункт")), remote) def test_no_matching_item_returns_the_same_object(self): remote = body("- [ ] пункт") self.assertIs(gmap.merge_checkbox_state(remote, body("- [x] другой")), remote) def test_a_body_with_no_checkboxes_at_all(self): remote = "## Summary\nодна проза\n" self.assertIs(gmap.merge_checkbox_state(remote, "- [x] пункт"), remote) self.assertEqual(gmap.merge_checkbox_state("- [ ] пункт", remote), "- [ ] пункт") def test_exactly_one_character_changes(self): """Ticking a box must not produce a diff wider than the state.""" remote = body("- [ ] пункт", "- [ ] второй") got = gmap.merge_checkbox_state(remote, body("- [x] пункт", "- [ ] второй")) diff = [i for i, (a, b) in enumerate(zip(remote, got)) if a != b] self.assertEqual(len(remote), len(got)) self.assertEqual(len(diff), 1) self.assertEqual((remote[diff[0]], got[diff[0]]), (" ", "x")) def test_a_rewrapped_item_keeps_its_tick(self): """`Checkbox.text` joins continuation lines with one space, which is the whole reason matching survives a reflow.""" remote = body("- [ ] длинный пункт, который сервер\n" " перенёс на две строки") got = gmap.merge_checkbox_state( remote, body("- [x] длинный пункт, который сервер перенёс на две строки")) self.assertIn("- [x] длинный пункт", got) def test_a_reworded_item_does_not_keep_its_tick(self): """Different text is a different item. The tick stays with the wording it was put on — this is a match, not a guess.""" got = gmap.merge_checkbox_state(body("- [ ] пункт про sqlc"), body("- [x] пункт про SQLC")) self.assertEqual(got, body("- [ ] пункт про sqlc")) def test_the_marker_style_does_not_have_to_match(self): """`-`, `*` and `1.` are all checkbox markers to the domain parser, so the item is the same item however the two sides chose to render it.""" got = gmap.merge_checkbox_state(body("1. [ ] пункт"), body("* [x] пункт")) self.assertEqual(got, body("1. [x] пункт")) def test_a_moved_item_keeps_its_tick(self): """Matching is on text alone; the section is not part of the key. An item promoted out of `## Acceptance criteria` is the same item.""" remote = "## Issues\n- [ ] пункт\n" got = gmap.merge_checkbox_state(remote, body("- [x] пункт")) self.assertEqual(got, "## Issues\n- [x] пункт\n") def test_duplicate_text_is_read_as_a_set(self): """The documented reading: one ticked local item ticks every remote line with that text. Pairing duplicates by order is the alternative, and it is the one that can still drop a tick.""" got = gmap.merge_checkbox_state(body("- [ ] пункт", "- [ ] пункт"), body("- [ ] пункт", "- [x] пункт")) self.assertEqual(got, body("- [x] пункт", "- [x] пункт")) def test_duplicate_text_never_loses_the_second_tick(self): """Two local lines, one remote: order-pairing would drop this tick.""" got = gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [ ] пункт", "- [x] пункт")) self.assertEqual(got, body("- [x] пункт")) def test_an_example_inside_a_code_fence_is_not_ticked(self): """The domain parser skips fences whole, and so does the merge: a `- [ ]` in a fence is markup being shown, not a box anyone may tick.""" remote = "## Spec\n```md\n- [ ] пункт\n```\n\n## Acceptance criteria\n- [ ] пункт\n" got = gmap.merge_checkbox_state(remote, body("- [x] пункт")) self.assertEqual(got, remote.replace("## Acceptance criteria\n- [ ] пункт", "## Acceptance criteria\n- [x] пункт")) self.assertIn("```md\n- [ ] пункт\n```", got) def test_an_existing_capital_X_is_left_alone(self): remote = body("- [X] пункт") self.assertIs(gmap.merge_checkbox_state(remote, body("- [x] пункт")), remote) def test_a_capital_X_locally_still_counts_as_ticked(self): got = gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [X] пункт")) self.assertEqual(got, body("- [x] пункт")) def test_unticking_in_the_web_does_not_survive(self): """The accepted price, pinned so nobody 'fixes' it by accident: unticking is not monotone, so a box unticked upstream comes back. Untick locally and push.""" got = gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [x] пункт")) self.assertEqual(got, body("- [x] пункт")) def test_parsing_is_the_domain_layer_s(self): """The acceptance criterion, asserted rather than eyeballed: the rule calls into skills/issue and defines no checkbox syntax of its own.""" with mock.patch.object(issue, "checkboxes", wraps=issue.checkboxes) as cb, \ mock.patch.object(issue, "set_checkbox", wraps=issue.set_checkbox) as sc: gmap.merge_checkbox_state(body("- [ ] пункт"), body("- [x] пункт")) self.assertTrue(cb.called) self.assertTrue(sc.called) def test_no_checkbox_markup_is_spelled_out_in_the_sync_layer(self): """The same criterion from the other side: the bracket markup itself appears nowhere under skills/sync/scripts. Knowing what `[ ]` looks like is the domain's job, and there is only one copy of it.""" sync = os.path.join(_ROOT, "skills", "sync", "scripts") for name in sorted(f for f in os.listdir(sync) if f.endswith(".py")): with open(os.path.join(sync, name)) as f: code = f.read().split('"""')[0::2] # docstrings dropped for chunk in code: for markup in ("[ xX]", "[xX]", "- [ ]", "- [x]"): self.assertNotIn(markup, chunk, "%s spells out %r" % (name, markup)) class FromApiTest(unittest.TestCase): """The seam between the rule and the translation.""" PAYLOAD = {"number": 42, "title": "T", "html_url": "u", "body": body("- [ ] пункт")} def test_local_body_is_optional_and_defaults_to_no_merge(self): iss, _ = gmap.from_api(dict(self.PAYLOAD), "an-issue", REPO) self.assertEqual(iss.body, body("- [ ] пункт")) def test_local_body_contributes_its_ticks(self): iss, _ = gmap.from_api(dict(self.PAYLOAD), "an-issue", REPO, local_body=body("- [x] пункт")) self.assertEqual(iss.body, body("- [x] пункт")) def test_an_empty_remote_body_does_not_crash(self): iss, _ = gmap.from_api({"number": 42, "title": "T", "body": None}, "an-issue", REPO, local_body=body("- [x] пункт")) self.assertEqual(iss.body, "") # -------------------------------------------------------------------------- # pull.py — the rule wired into the write path # -------------------------------------------------------------------------- class FakeGitea(object): """A `tea api` that answers issues from memory and remembers the calls.""" def __init__(self): self.calls = [] self.issues = {} def add(self, number, title, text, **kw): p = {"number": number, "title": title, "body": text, "state": "open", "comments": 0, "html_url": "https://git.example/%s/issues/%d" % (REPO, number), "labels": [{"name": "type/task"}], "assignees": [], "milestone": None, "updated_at": "2026-08-10T00:00:00Z"} p.update(kw) self.issues[number] = p return p def api(self, login, endpoint, method="GET", payload=None, **kw): self.calls.append((method, endpoint)) path, _, query = endpoint.partition("?") params = dict(urllib.parse.parse_qsl(query)) # Every pull asks for an issue's native links now (dependencies are the # default). Nothing here has any; the answer just has to exist. if path.endswith("/dependencies"): return [] m = re.match(r"^%s/issues/(\d+)$" % re.escape(BASE), path) if m and method == "GET": return self.issues.get(int(m.group(1))) if path == "%s/issues" % BASE and method == "GET": if int(params.get("page", 1)) > 1: return [] return [self.issues[n] for n in sorted(self.issues)] raise AssertionError("unstubbed call: %s %s" % (method, endpoint)) class PullTestCase(unittest.TestCase): """A temp store and a fake transport.""" def setUp(self): self.root = tempfile.mkdtemp(prefix="tea-store-") self.fake = FakeGitea() 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) self.addCleanup(shutil.rmtree, self.root, True) def write_local(self, id, text, number=42): issue.save(self.root, issue.Issue( id=id, title="An issue", body=text, labels=["type/task"], origin="gitea", extra={"gitea": "%s#%d" % (REPO, number), "url": "https://git.example/x", "synced": "old"})) def run_pull(self, *argv): out, err = io.StringIO(), io.StringIO() args = ["pull.py", "--repo", REPO, "--out", self.root] + list(argv) with mock.patch.object(sys, "argv", args), \ contextlib.redirect_stdout(out), \ contextlib.redirect_stderr(err): pull.main() return out.getvalue(), err.getvalue() def stored_body(self, id): return issue.load(self.root, id).body def raw(self, id): """The file on disk, byte for byte — metadata included.""" with open(issue.path_of(self.root, id)) as f: return f.read() class PullMergesTicksTest(PullTestCase): def test_a_tick_made_locally_survives_the_pull(self): self.write_local("an-issue", body("- [x] первое", "- [ ] второе")) self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое", "- [ ] второе")) def test_a_tick_made_in_the_web_lands_locally(self): self.write_local("an-issue", body("- [ ] первое", "- [ ] второе")) self.fake.add(42, "An issue", body("- [x] первое", "- [ ] второе")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое", "- [ ] второе")) def test_the_rest_of_the_body_is_still_overwritten(self): self.write_local("an-issue", body("- [x] первое", summary="Локальная правка прозы.")) self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] новое с сервера", summary="Серверная проза.")) self.run_pull("42") got = self.stored_body("an-issue") self.assertIn("Серверная проза.", got) self.assertNotIn("Локальная правка прозы.", got) self.assertIn("- [x] первое", got) self.assertIn("- [ ] новое с сервера", got) def test_an_item_the_server_added_arrives_ticked_if_the_server_ticked_it(self): self.write_local("an-issue", body("- [ ] первое")) self.fake.add(42, "An issue", body("- [ ] первое", "- [x] новое с сервера")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [ ] первое", "- [x] новое с сервера")) def test_filter_mode_merges_too(self): self.write_local("an-issue", body("- [x] первое")) self.fake.add(42, "An issue", body("- [ ] первое")) self.run_pull("--label", "type/task") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое")) def test_a_retitled_issue_keeps_its_slug_and_its_ticks(self): """The merge hangs off the local id, which is resolved from the number — a title change must not orphan the ticks.""" self.write_local("an-issue", body("- [x] первое")) self.fake.add(42, "Completely different title", body("- [ ] первое")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое")) self.assertEqual(issue.load(self.root, "an-issue").title, "Completely different title") class PullIntoAnEmptyStoreTest(PullTestCase): def test_no_local_file_writes_the_server_body_unchanged(self): self.fake.add(42, "An issue", body("- [x] первое", "- [ ] второе")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое", "- [ ] второе")) def test_a_store_that_does_not_exist_yet_is_created_and_not_merged(self): shutil.rmtree(self.root) self.fake.add(42, "An issue", body("- [ ] первое")) _out, err = self.run_pull("42") self.assertIn("created store", err) self.assertEqual(self.stored_body("an-issue"), body("- [ ] первое")) class CachedIsUnchangedTest(PullTestCase): def test_a_skipped_issue_is_neither_read_nor_merged(self): self.write_local("an-issue", body("- [x] первое")) self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе")) before = self.raw("an-issue") with mock.patch.object(gmap, "merge_checkbox_state") as merge: out, _ = self.run_pull("42", "--cached") merge.assert_not_called() self.assertIn("(cached)", out) self.assertEqual(self.raw("an-issue"), before) def test_without_cached_the_same_issue_is_merged(self): self.write_local("an-issue", body("- [x] первое")) self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе")) self.run_pull("42") self.assertEqual(self.stored_body("an-issue"), body("- [x] первое", "- [ ] второе")) class RoundTripTest(PullTestCase): """Pull twice with no change in between: the second is a no-op.""" def test_a_repeat_pull_does_not_churn_the_file(self): self.write_local("an-issue", body("- [x] первое", "- [ ] второе")) self.fake.add(42, "An issue", body("- [ ] первое", "- [ ] второе")) self.run_pull("42") first = self.raw("an-issue") self.run_pull("42") self.assertEqual(self.raw("an-issue"), first) if __name__ == "__main__": unittest.main()