claude-skills/marketplace
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

merge: drop the local copy after a successful push

Browse Source 
# Conflicts:
#	AGENTS.md
#	agents/tea-runner.md
This commit is contained in:
naudachu
2026-08-10 16:40:21 +05:00
parent 1815d91cdf e629d14585
commit 2ac301550e
14 changed files with 1410 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_drop_after_push.py
+805
View File
@@ -0,0 +1,805 @@
#!/usr/bin/env python3
"""
The local copy is dropped after a successful push, and pulled back on demand.
Two halves, and the second one is the one that matters:
1. **It deletes.** A confirmed create or PATCH removes `tmp/issues/<id>.md` and
`<id>.comments.md`, prints where the issue lives now, and leaves the ledger
behind so the slug can be found again. A pull puts the same file back —
same slug, same `depends:`, same body — including after a rename in Gitea
and on a machine that never had the file.
2. **It does not delete anything else, ever.** A transport that raised, a `tea`
that exited non-zero, an answer without a number, an answer for the wrong
issue, an `origin: local` issue nobody pushed: the file is still on disk.
A bug here destroys work, so every one of those paths is asserted
separately, and the assertion is always the same — `os.path.isfile`.
The transport is stubbed at `_gitea.api`, as `test_push_dependencies.py` does,
with one deliberate exception: the non-2xx test stubs `_gitea.subprocess`
instead and lets the REAL `_gitea.api` run, so "tea exited 1" is proved end to
end rather than assumed.
Nothing here touches a network, and nothing here touches the developer's store:
every test builds its own in a `tempfile.mkdtemp()`.
"""
import contextlib
import io
import json
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 issue # noqa: E402
import map as gmap # noqa: E402
import pull # noqa: E402
import push # 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
LABELS = {"type/task": 901, "type/bug": 902}
LABEL_NAMES = {v: k for k, v in LABELS.items()}
BODY = """## Summary
Прозаическое описание задачи.
## Spec
skills/issue/references/format.md
## Acceptance criteria
- [ ] что-нибудь работает
"""
BODY_WITH_DEPS = """## Summary
Прозаическое описание задачи.
## Spec
skills/issue/references/format.md
## Depends on
- first-thing — ставит фундамент, без него второй не собрать
## Acceptance criteria
- [ ] что-нибудь работает
"""
# --------------------------------------------------------------------------
# a tracker that can be both pushed to and pulled from
# --------------------------------------------------------------------------
class FakeTracker(object):
"""`tea api` answered from memory, for push AND pull.
It keeps bodies the way Gitea does — verbatim, marker and all — which is
what makes the round-trip tests real: the slug that comes back is the one
that was actually stored on the far side, not one the test handed over."""
def __init__(self, next_number=101):
self.calls = []
self.next_number = next_number
self.issues = {} # number -> payload
self.deps = {} # number -> {(repo, number)}
# Failure injection, one write at a time.
self.raise_on_write = None # an exception instance to raise
self.answer_override = None # what a write answers instead
# -- state -------------------------------------------------------------
def store(self, number, title, body, **kw):
p = {"number": number, "title": title, "body": body, "state": "open",
"comments": 0, "labels": [{"name": "type/task"}], "assignees": [],
"milestone": None, "ref": "test-branch",
"html_url": "https://git.example/%s/issues/%d" % (REPO, number),
"updated_at": "2026-08-10T00:00:00Z",
"repository": {"full_name": REPO}}
p.update(kw)
self.issues[number] = p
return p
def body_of(self, number):
return self.issues[number]["body"]
def rename(self, number, title):
self.issues[number]["title"] = title
def writes(self):
return [c for c in self.calls if c[0] != "GET"]
# -- the seam ----------------------------------------------------------
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 path == "%s/labels" % BASE and method == "GET":
return [{"name": n, "id": i} for n, i in LABELS.items()]
if path.endswith("/comments"):
return []
if path.endswith("/dependencies"):
number = int(path.split("/issues/")[1].split("/")[0])
if method == "GET":
return [dict(self.issues[n], repository={"full_name": r})
for r, n in sorted(self.deps.get(number, set()))
if n in self.issues]
if method == "POST":
self.deps.setdefault(number, set()).add(
("%s/%s" % (payload["owner"], payload["repo"]),
int(payload["index"])))
return {"number": number}
if path == "%s/issues" % BASE and method == "POST":
return self._write(
lambda: self.store(self._next(), payload.get("title", ""),
payload.get("body", ""),
labels=self._labels(payload),
ref=payload.get("ref", "")))
if "/issues/" in path and method == "PATCH":
number = int(path.rsplit("/", 1)[1])
return self._write(
lambda: self.store(number, payload.get("title", ""),
payload.get("body", ""),
labels=self._labels(payload),
ref=payload.get("ref", "")))
if "/issues/" in path and method == "GET":
return self.issues.get(int(path.rsplit("/", 1)[1]))
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
# -- helpers -----------------------------------------------------------
def _next(self):
n = self.next_number
self.next_number += 1
return n
def _labels(self, payload):
return [{"name": LABEL_NAMES[i]} for i in (payload or {}).get("labels") or []
if i in LABEL_NAMES]
def _write(self, do):
"""Every create and update goes through here, so a test can make one
fail without knowing which verb it was."""
if self.raise_on_write is not None:
raise self.raise_on_write
got = do()
if self.answer_override is not None:
return self.answer_override
return got
class StoreTestCase(unittest.TestCase):
"""A temp store, a fake tracker, and no git."""
def setUp(self):
self.root = tempfile.mkdtemp(prefix="tea-drop-")
self.fake = FakeTracker()
for p in (mock.patch.object(_gitea, "api", self.fake.api),
mock.patch.object(_gitea, "require_login", lambda: "test-login"),
mock.patch.object(push, "git_branch", lambda: "test-branch")):
p.start()
self.addCleanup(p.stop)
self.addCleanup(shutil.rmtree, self.root, True)
# -- fixtures ----------------------------------------------------------
def write_issue(self, id, title, body=BODY, depends=(), origin=issue.LOCAL,
extra=None):
iss = issue.Issue(id=id, title=title, body=body, labels=["type/task"],
depends=list(depends), origin=origin,
extra=dict(extra or {}))
issue.save(self.root, iss)
return iss
def write_comments(self, id, text="## comment 1 — someone — 2026-08-10\n\nтекст\n"):
p = _gitea.comments_path(self.root, id)
with open(p, "w") as f:
f.write(text)
return p
# -- runners -----------------------------------------------------------
def run_push(self, *argv):
return self._run(push, "push.py", argv)
def run_pull(self, *argv):
return self._run(pull, "pull.py", argv)
def _run(self, mod, name, argv):
# Kept on self so a test that expects SystemExit can still read what
# went to stderr — the run never returns in that case.
self.out, self.err = io.StringIO(), io.StringIO()
args = [name, "--repo", REPO, "--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 ledger(self):
return _gitea.load_map(self.root)
def number_of(self, id):
for key, slug in self.ledger().items():
if slug == id:
return gmap.parse_remote_key(key)[1]
return None
# --------------------------------------------------------------------------
# it deletes
# --------------------------------------------------------------------------
class DropsAfterCreateTest(StoreTestCase):
def test_the_issue_file_is_gone(self):
self.write_issue("a-thing", "A thing")
self.run_push()
self.assertGone("a-thing")
def test_the_comment_thread_goes_with_it(self):
self.write_issue("a-thing", "A thing")
cpath = self.write_comments("a-thing")
self.run_push()
self.assertFalse(os.path.isfile(cpath), "the thread outlived the issue")
def test_a_missing_thread_is_not_an_error(self):
"""Most issues have no comments file. Dropping must not care."""
self.write_issue("a-thing", "A thing")
out, _ = self.run_push()
self.assertIn("dropped", out)
def test_the_output_names_the_number_and_the_url(self):
"""The local path is gone, so this line is the only address left."""
self.write_issue("a-thing", "A thing")
out, _ = self.run_push()
n = self.number_of("a-thing")
self.assertIn("#%d" % n, out)
self.assertIn("https://git.example/%s/issues/%d" % (REPO, n), out)
self.assertIn("pull.py %d" % n, out)
def test_the_ledger_outlives_the_file(self):
"""`.remote.json` does not become garbage when the files go — it
becomes the only local record of which slug this number is."""
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.assertIsNotNone(n)
self.assertEqual(self.ledger(), {gmap.remote_key(REPO, n): "a-thing"})
def test_the_ledger_is_written_before_the_file_is_removed(self):
"""Ordering, asserted rather than trusted: if the two were swapped, an
interrupted run would cost the slug and not just a re-pull."""
seen = {}
real_drop = push.drop_local
def spy(root, id):
seen["ledger"] = json.load(open(_gitea.map_path(root)))
return real_drop(root, id)
self.write_issue("a-thing", "A thing")
with mock.patch.object(push, "drop_local", spy):
self.run_push()
self.assertIn("a-thing", (seen.get("ledger") or {}).values())
class DropsAfterUpdateTest(StoreTestCase):
"""One rule, no exception: `--update` deletes too."""
def pushed_then_pulled(self, id="a-thing", body=BODY):
self.write_issue(id, "A thing", body=body)
self.run_push()
self.run_pull(str(self.number_of(id)))
self.assertOnDisk(id, "the pull should have put it back")
return id
def test_patch_deletes_the_file_too(self):
id = self.pushed_then_pulled()
out, _ = self.run_push("--update", id)
self.assertIn("updated", out)
self.assertGone(id)
def test_patch_deletes_the_thread_too(self):
id = self.pushed_then_pulled()
cpath = self.write_comments(id)
self.run_push("--update", id)
self.assertFalse(os.path.isfile(cpath))
def test_the_patch_really_went_out(self):
id = self.pushed_then_pulled()
self.run_push("--update", id)
self.assertTrue([c for c in self.fake.calls if c[0] == "PATCH"])
# --------------------------------------------------------------------------
# it deletes nothing else
# --------------------------------------------------------------------------
class NeverPushedIsNeverDroppedTest(StoreTestCase):
def test_a_local_issue_nobody_selected_stays(self):
self.write_issue("pushed-thing", "Pushed thing")
self.write_issue("kept-thing", "Kept thing")
self.run_push("pushed-thing")
self.assertGone("pushed-thing")
self.assertOnDisk("kept-thing", "it was never pushed")
def test_a_local_only_dependency_stays(self):
"""It is read (for the warning) but never sent, so never dropped."""
self.write_issue("first-thing", "First thing")
self.write_issue("second-thing", "Second thing", body=BODY_WITH_DEPS,
depends=["first-thing"])
_, err = self.run_push("second-thing")
self.assertIn("depends on local-only issue(s) first-thing", err)
self.assertOnDisk("first-thing", "it was never sent")
def test_dry_run_deletes_nothing(self):
self.write_issue("a-thing", "A thing")
self.run_push("--dry-run")
self.assertOnDisk("a-thing", "--dry-run must not write or delete")
self.assertEqual(self.fake.calls, [])
def test_a_format_violation_stops_before_anything_is_sent(self):
"""No type/* label: validation fails, nothing is sent, nothing goes."""
issue.save(self.root, issue.Issue(id="bad-thing", title="Bad thing",
body=BODY, labels=[]))
with self.assertRaises(SystemExit):
self.run_push("bad-thing")
self.assertOnDisk("bad-thing")
self.assertEqual(self.fake.writes(), [])
class SurvivesEveryFailureTest(StoreTestCase):
"""The criterion that matters most. Each path is asserted on its own."""
def test_a_transport_exception_leaves_the_file(self):
"""`tea` could not be run at all — the exception propagates out of the
push and the delete is never reached."""
self.write_issue("a-thing", "A thing")
self.fake.raise_on_write = OSError("tea: command not found")
with self.assertRaises(OSError):
self.run_push()
self.assertOnDisk("a-thing", "the transport raised")
self.assertEqual(self.ledger(), {})
def test_a_non_2xx_answer_leaves_the_file(self):
"""The real `_gitea.api` against a `tea` that exits 1.
Stubbed one layer lower than every other test here on purpose: this is
the path a 422 or a 500 actually takes, and it ends in `die()`."""
self.write_issue("a-thing", "A thing")
def fake_run(cmd, capture_output=False, text=False):
creating = "-X" in cmd and cmd[cmd.index("-X") + 1] == "POST"
if creating:
return types.SimpleNamespace(
returncode=1, stdout="",
stderr="422 Unprocessable Entity: validation failed")
if cmd[-1].split("?")[0].endswith("/labels"):
return types.SimpleNamespace(
returncode=0, stderr="",
stdout=json.dumps([{"name": n, "id": i}
for n, i in LABELS.items()]))
return types.SimpleNamespace(returncode=0, stdout="", stderr="")
with mock.patch.object(_gitea, "api", REAL_API), \
mock.patch.object(_gitea, "subprocess",
types.SimpleNamespace(run=fake_run)), \
self.assertRaises(SystemExit):
self.run_push()
self.assertOnDisk("a-thing", "tea exited non-zero")
def test_an_answer_without_a_number_leaves_the_file(self):
self.write_issue("a-thing", "A thing")
self.fake.answer_override = {"ok": True, "message": "created"}
with self.assertRaises(SystemExit):
self.run_push()
self.assertOnDisk("a-thing", "the answer carried no number")
def test_an_answer_that_is_not_an_object_leaves_the_file(self):
self.write_issue("a-thing", "A thing")
self.fake.answer_override = ["something", "else"]
with self.assertRaises(SystemExit):
self.run_push()
self.assertOnDisk("a-thing")
def test_an_empty_answer_leaves_the_file(self):
"""`tea` exited 0 and printed nothing — api returns None."""
self.write_issue("a-thing", "A thing")
self.fake.answer_override = None
real_write = self.fake._write
self.fake._write = lambda do: (real_write(do), None)[1]
with self.assertRaises(SystemExit):
self.run_push()
self.assertOnDisk("a-thing")
def test_a_patch_answering_for_another_issue_leaves_the_file(self):
"""The mismatched-body case: we PATCHed #101 and #999 answered."""
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.run_pull(str(n))
self.assertOnDisk("a-thing")
self.fake.answer_override = {"number": 999, "html_url": "https://x"}
with self.assertRaises(SystemExit):
self.run_push("--update", "a-thing")
self.assertOnDisk("a-thing", "the tracker answered for a different issue")
def test_the_error_says_the_file_is_untouched(self):
self.write_issue("a-thing", "A thing")
self.fake.answer_override = {"ok": True}
with self.assertRaises(SystemExit):
self.run_push()
self.assertIn("untouched", self.err.getvalue())
def test_a_failure_partway_through_keeps_what_has_not_been_sent(self):
"""Two issues, the second one fails. The first is legitimately gone —
Gitea confirmed it — and the second is still here."""
self.write_issue("aaa-thing", "Aaa thing")
self.write_issue("zzz-thing", "Zzz thing")
real_write = self.fake._write
seen = []
def once(do):
seen.append(1)
if len(seen) > 1:
return {"nope": True}
return real_write(do)
self.fake._write = once
with self.assertRaises(SystemExit):
self.run_push()
self.assertGone("aaa-thing")
self.assertOnDisk("zzz-thing", "its write never succeeded")
# And the one that did go up is in the ledger, so it is findable.
self.assertEqual(list(self.ledger().values()), ["aaa-thing"])
class ConfirmedNumberTest(unittest.TestCase):
"""The gate itself. Everything below it deletes a file."""
def test_a_plain_create_is_confirmed(self):
self.assertEqual(push.confirmed_number({"number": 42}), 42)
def test_a_matching_patch_is_confirmed(self):
self.assertEqual(push.confirmed_number({"number": 42}, 42), 42)
def test_a_mismatched_patch_is_not(self):
self.assertIsNone(push.confirmed_number({"number": 43}, 42))
def test_none_is_not(self):
self.assertIsNone(push.confirmed_number(None))
def test_a_list_is_not(self):
self.assertIsNone(push.confirmed_number([{"number": 42}]))
def test_a_missing_number_is_not(self):
self.assertIsNone(push.confirmed_number({"html_url": "https://x"}))
def test_a_string_number_is_not(self):
self.assertIsNone(push.confirmed_number({"number": "42"}))
def test_true_is_not_a_number(self):
"""`True` is an `int` in Python; `number: true` confirms nothing."""
self.assertIsNone(push.confirmed_number({"number": True}))
def test_zero_and_negatives_are_not(self):
self.assertIsNone(push.confirmed_number({"number": 0}))
self.assertIsNone(push.confirmed_number({"number": -1}))
# --------------------------------------------------------------------------
# the id marker
# --------------------------------------------------------------------------
class IdMarkerTest(unittest.TestCase):
"""map.py, pure — no store, no tracker."""
def test_the_marker_is_the_first_line(self):
got = gmap.with_id_marker("## Summary\nтекст", "a-thing")
self.assertEqual(got.splitlines()[0], "<!-- tea:id a-thing -->")
self.assertEqual(got.splitlines()[1], "")
def test_strip_is_the_exact_inverse(self):
for body in ("## Summary\nтекст", "", "one line",
"## Summary\n\n- [ ] пункт\n\n## Spec\nnone"):
self.assertEqual(gmap.strip_id_marker(gmap.with_id_marker(body, "x")),
body)
def test_a_body_with_no_marker_comes_back_byte_for_byte(self):
body = "## Summary\n\n весь текст \n\n\n"
self.assertEqual(gmap.strip_id_marker(body), body)
def test_marking_twice_still_leaves_one(self):
once = gmap.with_id_marker("текст", "a-thing")
twice = gmap.with_id_marker(once, "a-thing")
self.assertEqual(once, twice)
self.assertEqual(twice.count("tea:id"), 1)
def test_remarking_under_a_new_slug_replaces_rather_than_adds(self):
got = gmap.with_id_marker(gmap.with_id_marker("текст", "old"), "new")
self.assertEqual(got.count("tea:id"), 1)
self.assertEqual(gmap.id_in_body(got), "new")
def test_every_marker_is_removed_not_just_the_first(self):
"""A body hand-edited in the web UI could hold two. It comes back with
none, and the next push writes exactly one."""
mangled = ("<!-- tea:id one -->\n\nтекст\n\n<!-- tea:id two -->\nещё")
self.assertEqual(gmap.strip_id_marker(mangled), "текст\n\nещё")
self.assertEqual(gmap.with_id_marker(mangled, "one").count("tea:id"), 1)
def test_id_in_body_reads_the_first_marker(self):
self.assertEqual(gmap.id_in_body("<!-- tea:id one -->\n\nx"), "one")
self.assertIsNone(gmap.id_in_body("## Summary\nтекст"))
self.assertIsNone(gmap.id_in_body(""))
def test_a_marker_that_is_not_a_slug_is_ignored(self):
"""Better to fall back to the title than to name a file after junk."""
for junk in ("Not A Slug", "../etc/passwd", "-leading", "два-слова"):
self.assertIsNone(gmap.id_in_body("<!-- tea:id %s -->\n\nx" % junk))
def test_the_marker_tolerates_spacing(self):
self.assertEqual(gmap.id_in_body("<!--tea:id a-thing-->"), "a-thing")
self.assertEqual(gmap.id_in_body(" <!-- tea:id a-thing --> "),
"a-thing")
def test_a_marker_inside_prose_is_not_one(self):
"""Only a line that is nothing but the marker counts."""
self.assertIsNone(gmap.id_in_body("см. <!-- tea:id a-thing --> выше"))
def test_to_payload_marks_and_from_api_unmarks(self):
iss = issue.Issue(id="a-thing", title="A thing", body="## Summary\nтекст")
sent = gmap.to_payload(iss)["body"]
self.assertTrue(sent.startswith("<!-- tea:id a-thing -->"))
back, _ = gmap.from_api({"number": 1, "title": "A thing", "body": sent},
"a-thing", REPO)
self.assertEqual(back.body, "## Summary\nтекст")
class MarkerStaysOffDiskTest(StoreTestCase):
def test_the_local_file_never_holds_a_marker(self):
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.assertIn("tea:id a-thing", self.fake.body_of(n))
self.run_pull(str(n))
with open(issue.path_of(self.root, "a-thing")) as f:
self.assertNotIn("tea:id", f.read())
def test_repeated_round_trips_do_not_accumulate_markers(self):
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
for _ in range(3):
self.run_pull(str(n))
self.run_push("--update", "a-thing")
self.assertEqual(self.fake.body_of(n).count("tea:id"), 1)
# --------------------------------------------------------------------------
# the round trip
# --------------------------------------------------------------------------
class RoundTripTest(StoreTestCase):
"""push -> the file is gone -> pull -> the same file is back."""
def two_issues(self):
self.write_issue("first-thing", "First thing")
self.write_issue("second-thing", "Second thing", body=BODY_WITH_DEPS,
depends=["first-thing"])
def snapshot(self, id):
iss = issue.load(self.root, id)
return (iss.id, iss.title, iss.body, sorted(iss.depends),
sorted(iss.labels), iss.state)
def test_the_file_comes_back_identical(self):
self.two_issues()
before = self.snapshot("second-thing")
self.run_push()
self.assertGone("second-thing")
self.run_pull(str(self.number_of("second-thing")), "--deps")
self.assertEqual(self.snapshot("second-thing"), before)
def test_depends_survives_the_round_trip(self):
"""The edge lives in Gitea's own graph while the files do not exist —
push wrote it, `pull --deps` reads it back, and the ledger turns the
number back into the slug it had here."""
self.two_issues()
self.run_push()
self.assertGone("first-thing")
self.assertGone("second-thing")
self.run_pull(str(self.number_of("second-thing")), "--deps")
self.assertEqual(issue.load(self.root, "second-thing").depends,
["first-thing"])
def test_the_prose_dependency_is_still_the_authors_words(self):
self.two_issues()
self.run_push()
self.run_pull(str(self.number_of("second-thing")), "--deps")
self.assertIn("- first-thing — ставит фундамент",
issue.load(self.root, "second-thing").body)
def test_a_rename_in_gitea_does_not_change_the_slug(self):
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.fake.rename(n, "Completely different title now")
self.run_pull(str(n))
self.assertOnDisk("a-thing")
self.assertFalse(os.path.isfile(
issue.path_of(self.root, "completely-different-title-now")))
self.assertEqual(issue.load(self.root, "a-thing").title,
"Completely different title now")
def test_the_slug_survives_a_rename_with_the_ledger_thrown_away(self):
"""The case `.remote.json` cannot cover: a fresh clone, or another
machine. The marker is the only thing left, and it is enough."""
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.fake.rename(n, "Completely different title now")
os.remove(_gitea.map_path(self.root))
self.run_pull(str(n))
self.assertOnDisk("a-thing")
self.assertEqual(self.ledger(), {gmap.remote_key(REPO, n): "a-thing"})
def test_depends_survives_a_lost_ledger_when_both_come_back(self):
self.two_issues()
self.run_push()
first, second = self.number_of("first-thing"), self.number_of("second-thing")
os.remove(_gitea.map_path(self.root))
self.run_pull(str(first), str(second), "--deps")
self.assertEqual(issue.load(self.root, "second-thing").depends,
["first-thing"])
def test_an_issue_filed_in_the_web_ui_still_gets_a_slug(self):
"""No marker, no ledger entry — the title is the fallback, as before."""
self.fake.store(500, "Filed in the web ui", "## Summary\nтекст")
self.run_pull("500")
self.assertOnDisk("filed-in-the-web-ui")
def test_a_marker_colliding_with_a_local_issue_does_not_overwrite_it(self):
"""A slug is only taken at its word when it is free."""
self.write_issue("a-thing", "A thing", body="## Summary\nмоя локальная")
self.fake.store(500, "Something else",
gmap.with_id_marker("## Summary\nчужая", "a-thing"))
self.run_pull("500")
self.assertIn("моя локальная", issue.load(self.root, "a-thing").body)
self.assertIn("чужая", issue.load(self.root, "a-thing-2").body)
def test_the_branch_ref_comes_back_with_the_issue(self):
"""`branch:` is not written back to a file that is being deleted; it
goes up in the payload and comes down again on the next pull."""
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.run_pull(str(n))
self.assertEqual(issue.load(self.root, "a-thing").extra.get("branch"),
"test-branch")
def test_pushing_the_pulled_copy_back_is_a_no_op_on_the_body(self):
self.write_issue("a-thing", "A thing")
self.run_push()
n = self.number_of("a-thing")
self.run_pull(str(n))
before = self.fake.body_of(n)
self.run_push("--update", "a-thing")
self.assertEqual(self.fake.body_of(n), before)
# --------------------------------------------------------------------------
# the ledger
# --------------------------------------------------------------------------
class StoreListingTest(StoreTestCase):
"""The store layout the drop depends on."""
def test_a_comment_thread_is_not_an_issue(self):
"""`<id>.comments.md` sits in the store beside the issue. A slug has no
dot in it, so it is not a slug and not a unit of work — otherwise a bare
`push.py` files the comment thread as an issue of its own."""
self.write_issue("a-thing", "A thing")
self.write_comments("a-thing")
self.assertEqual(issue.all_ids(self.root), ["a-thing"])
def test_a_bare_push_with_threads_in_the_store_still_works(self):
self.write_issue("a-thing", "A thing")
self.write_comments("a-thing")
self.run_push()
self.assertGone("a-thing")
class LedgerTest(StoreTestCase):
"""`.remote.json` after the files it used to index are gone."""
def test_rebuild_keeps_entries_whose_files_no_longer_exist(self):
"""It used to reconstruct the map from the files and save the result,
which would now silently drop every pushed issue."""
_gitea.save_map(self.root, {gmap.remote_key(REPO, 7): "gone-thing"})
self.write_issue("here-thing", "Here thing", origin="gitea",
extra={"gitea": gmap.remote_key(REPO, 8)})
got = _gitea.rebuild_map(self.root, issue.load_all(self.root))
self.assertEqual(got, {gmap.remote_key(REPO, 7): "gone-thing",
gmap.remote_key(REPO, 8): "here-thing"})
self.assertEqual(_gitea.load_map(self.root), got)
def test_a_second_push_reuses_the_ledger_not_the_files(self):
"""Two pushes, no pull in between for the blocker: its file is gone, so
its number can only come from the ledger — and the link is still made."""
self.write_issue("first-thing", "First thing")
self.run_push("first-thing")
self.assertGone("first-thing")
self.write_issue("second-thing", "Second thing", body=BODY_WITH_DEPS,
depends=["first-thing"])
out, err = self.run_push("second-thing")
first, second = self.number_of("first-thing"), self.number_of("second-thing")
self.assertEqual(self.fake.deps.get(second), {(REPO, first)})
self.assertIn("depends on %s#%d (first-thing)" % (REPO, first), out)
self.assertNotIn("local-only", err)
def test_the_dry_run_resolves_a_dropped_blocker_from_the_ledger(self):
self.write_issue("first-thing", "First thing")
self.run_push("first-thing")
first = self.number_of("first-thing")
self.write_issue("second-thing", "Second thing", body=BODY_WITH_DEPS,
depends=["first-thing"])
out, _ = self.run_push("--dry-run", "second-thing")
self.assertIn("link -> %s#%d (first-thing)" % (REPO, first), out)
def test_ledger_keys_prefers_the_current_repo(self):
m = {"other/repo#7": "a-thing", "%s#9" % REPO: "a-thing"}
self.assertEqual(push.ledger_keys(m, REPO), {"a-thing": "%s#9" % REPO})
if __name__ == "__main__":
unittest.main()
tests/test_push_dependencies.py
+41 -11
View File
@@ -153,12 +153,27 @@ class PushTestCase(unittest.TestCase):
# -- fixtures ----------------------------------------------------------
def write_issue(self, id, title, body=BODY_NO_DEPS, depends=(), extra=None):
def write_issue(self, id, title, body=BODY_NO_DEPS, depends=(), extra=None,
origin=issue.LOCAL):
iss = issue.Issue(id=id, title=title, body=body, labels=["type/task"],
depends=list(depends), extra=dict(extra or {}))
depends=list(depends), origin=origin,
extra=dict(extra or {}))
issue.save(self.root, iss)
return iss
def repull(self, id, body=BODY_NO_DEPS, depends=()):
"""Put a pushed issue back the way `pull.py` would.
Push deletes the file, so anything that pushes the same issue twice has
to fetch it in between — which is the workflow, not a test artifact.
The slug and the number come from the ledger, exactly as `pull.id_for`
would resolve them."""
number = self.number_of(id)
self.assertIsNotNone(number, "%s was never pushed" % id)
return self.write_issue(id, self.fake.titles[number], body=body,
depends=depends, origin="gitea",
extra={"gitea": "%s#%d" % (REPO, number)})
def two_issues(self):
"""first-thing, and second-thing which depends on it."""
self.write_issue("first-thing", "First thing")
@@ -175,7 +190,14 @@ class PushTestCase(unittest.TestCase):
return out.getvalue(), err.getvalue()
def number_of(self, id):
return gmap.number_of(issue.load(self.root, id))
"""The number an id was pushed under, or None.
Read off `.remote.json` rather than the issue file: a successful push
deletes the file, and the ledger is what is left behind."""
for key, got in _gitea.load_map(self.root).items():
if got == id:
return gmap.parse_remote_key(key)[1]
return None
# --------------------------------------------------------------------------
@@ -293,6 +315,8 @@ class IdempotenceTest(PushTestCase):
self.run_push()
self.assertEqual(len(self.fake.dep_posts()), 1)
self.repull("first-thing")
self.repull("second-thing", body=BODY, depends=["first-thing"])
self.run_push("--update")
self.assertEqual(len(self.fake.dep_posts()), 1, "link re-POSTed")
self.assertEqual(self.fake.deps[self.number_of("second-thing")],
@@ -318,10 +342,9 @@ class UpdateCarriesNewLinksTest(PushTestCase):
self.run_push()
self.assertEqual(self.fake.dep_posts(), [])
iss = issue.load(self.root, "second-thing")
iss.depends = ["first-thing"]
iss.body = BODY
issue.save(self.root, iss)
# The issue comes back from Gitea, and the dependency is added to the
# copy that came back — there is no other copy to add it to.
self.repull("second-thing", body=BODY, depends=["first-thing"])
self.run_push("--update", "second-thing")
self.assertEqual(self.fake.deps[self.number_of("second-thing")],
@@ -356,10 +379,13 @@ class DryRunTest(PushTestCase):
class BodyIsVerbatimTest(PushTestCase):
"""The prose is untouched. The id marker is the one thing push adds, and it
comes straight back off — `strip_id_marker` is the inverse."""
def test_depends_on_prose_is_not_rewritten_to_numbers(self):
"""map.py deliberately never edits the prose. Linking must not start."""
self.two_issues()
before = issue.load(self.root, "second-thing").body
self.run_push()
created = [c for c in self.fake.calls
@@ -369,16 +395,20 @@ class BodyIsVerbatimTest(PushTestCase):
self.assertIn("- first-thing — ставит фундамент", second_body)
self.assertNotIn("#101", second_body)
self.assertEqual(second_body, issue.load(self.root, "second-thing").body)
self.assertEqual(gmap.strip_id_marker(second_body), before)
def test_body_survives_a_second_push_unchanged(self):
self.two_issues()
self.run_push()
before = issue.load(self.root, "second-thing").body
self.run_push()
self.repull("first-thing")
self.repull("second-thing", body=before, depends=["first-thing"])
self.assertEqual(issue.load(self.root, "second-thing").body, before)
self.run_push("--update")
patched = [c for c in self.fake.calls if c[0] == "PATCH"]
self.assertIn(before, [c[2]["body"] for c in patched])
self.assertEqual(before, issue.load(self.root, "second-thing").body)
self.assertIn(before, [gmap.strip_id_marker(c[2]["body"]) for c in patched])
class DepStateTest(PushTestCase):