naudachu
d4c43464e5
A tick was lost in both directions: pull wrote the server body as-is, push sent the local body as-is, last writer won. Tick it in the web UI and the first `push.py --update` dropped it; tick it locally and the first pull dropped it. The usual answer is drift tracking and a three-way merge, which this repo rejected on purpose. It is not needed. A tick is monotone — an item only travels `[ ]` -> `[x]` — so unioning the two sides is a set union, not conflict resolution. One rule for one line type replaces the whole mechanism, and the store stays "not a mirror". `map.merge_checkbox_state` is pure and does the work; `from_api` takes the local body as an optional argument; `pull.py` hands it the copy already on disk. Checkbox parsing is imported from `skills/issue` (`checkboxes` / `set_checkbox`), never redefined here — the domain layer is untouched. The same item text more than once is read as a set: one ticked local item ticks every server line with that text. Pairing duplicates up by order is the alternative, and it can still drop a tick — which is the bug being fixed. The price is documented, not hidden: unticking is not monotone, so a box unticked in the web UI comes back on the next pull. Untick locally, then push. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
408 lines
19 KiB
Python
408 lines
19 KiB
Python
#!/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))
|
|
|
|
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()
|