#!/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/.md` and `.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, 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], "") 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 = ("\n\nтекст\n\n\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("\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("\n\nx" % junk)) def test_the_marker_tolerates_spacing(self): self.assertEqual(gmap.id_in_body(""), "a-thing") self.assertEqual(gmap.id_in_body(" "), "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("см. выше")) 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("")) 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): """`.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()