naudachu
f5977fa4fc
Two conflicts git could see (AGENTS.md, skills/sync/SKILL.md) and one it could not: the payload-root change removed api()'s out_root parameter, so close.py stops passing it, and its payload test now asserts PAYLOAD_ROOT instead of the deleted PAYLOAD_DIR.
642 lines
25 KiB
Python
642 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
close.py — the state changes in Gitea, and the local file follows it or nothing
|
|
happens at all.
|
|
|
|
Two halves, and the second is the one that matters:
|
|
|
|
1. **It closes.** A slug, a number, several of either in one run, and
|
|
`--reopen` going the other way. What goes out is a PATCH carrying `state`
|
|
and nothing else; what comes back is written into `state:` on the local
|
|
file, and the index is rebuilt so the store's own table agrees.
|
|
|
|
2. **It changes nothing local unless the tracker confirmed it.** A `tea` that
|
|
exited non-zero, an answer with no number, an answer for another issue, an
|
|
answer that still says `open`, an `origin: local` issue, a `--dry-run`: in
|
|
every one of those the file on disk is byte for byte what it was. A bug here
|
|
makes the store lie about the tracker, so each path is asserted on its own.
|
|
|
|
The transport is stubbed at `_gitea.api`, as `test_drop_after_push.py` does,
|
|
with the same deliberate exception: the non-2xx test stubs `_gitea.subprocess`
|
|
and lets the real `_gitea.api` run, so "tea exited 1" is proved end to end.
|
|
|
|
Nothing here touches a network, and nothing here touches the developer's store:
|
|
every test builds its own in a `tempfile.TemporaryDirectory()`.
|
|
"""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
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 close # noqa: E402
|
|
import issue # noqa: E402
|
|
import map as gmap # 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
|
|
|
|
BODY = """## Summary
|
|
Прозаическое описание задачи.
|
|
|
|
## Spec
|
|
skills/issue/references/format.md
|
|
|
|
## Acceptance criteria
|
|
- [x] что-нибудь работает
|
|
"""
|
|
|
|
|
|
class FakeTracker(object):
|
|
"""`tea api` answered from memory, for state writes only.
|
|
|
|
It keeps a `state` per number and flips it on a PATCH, which is the whole
|
|
contract close.py has with the far side."""
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.states = {} # number -> "open" / "closed"
|
|
self.raise_on_write = None # an exception instance to raise
|
|
self.answer_override = None # what a write answers instead
|
|
|
|
def payload_of(self, number):
|
|
return {"number": number, "state": self.states[number],
|
|
"title": "A thing", "updated_at": "2026-08-11T00:00:00Z",
|
|
"html_url": "https://git.example/%s/issues/%d" % (REPO, number)}
|
|
|
|
def writes(self):
|
|
return [c for c in self.calls if c[0] != "GET"]
|
|
|
|
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 "/issues/" in path and method == "PATCH":
|
|
number = int(path.rsplit("/", 1)[1])
|
|
if self.raise_on_write is not None:
|
|
raise self.raise_on_write
|
|
self.states.setdefault(number, "open")
|
|
if "state" in (payload or {}):
|
|
self.states[number] = payload["state"]
|
|
if self.answer_override is not None:
|
|
return self.answer_override
|
|
return self.payload_of(number)
|
|
|
|
if "/issues/" in path and method == "GET":
|
|
n = int(path.rsplit("/", 1)[1])
|
|
return self.payload_of(n) if n in self.states else None
|
|
|
|
raise AssertionError("unstubbed call: %s %s" % (method, endpoint))
|
|
|
|
|
|
class StoreTestCase(unittest.TestCase):
|
|
"""A temp store and a fake tracker."""
|
|
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory(prefix="tea-close-")
|
|
self.addCleanup(tmp.cleanup)
|
|
self.root = tmp.name
|
|
self.fake = FakeTracker()
|
|
for p in (mock.patch.object(_gitea, "api", self.fake.api),
|
|
mock.patch.object(_gitea, "require_login", lambda: "test-login")):
|
|
p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
# -- fixtures ----------------------------------------------------------
|
|
|
|
def synced(self, id="a-thing", number=101, state="open"):
|
|
"""An issue that is in the tracker and on disk, the way a pull leaves
|
|
it: `origin: gitea`, a `gitea:` field, and a ledger entry."""
|
|
key = gmap.remote_key(REPO, number)
|
|
iss = issue.Issue(id=id, title="A thing", body=BODY, state=state,
|
|
labels=["type/task"], origin=gmap.ORIGIN,
|
|
extra={"gitea": key, "url": "https://git.example/x",
|
|
"synced": "2026-08-10T00:00:00Z"})
|
|
issue.save(self.root, iss)
|
|
m = _gitea.load_map(self.root)
|
|
m[key] = id
|
|
_gitea.save_map(self.root, m)
|
|
self.fake.states[number] = state
|
|
return iss
|
|
|
|
def local_only(self, id="local-thing"):
|
|
"""An issue that has never left this machine."""
|
|
iss = issue.Issue(id=id, title="Local thing", body=BODY,
|
|
labels=["type/task"])
|
|
issue.save(self.root, iss)
|
|
return iss
|
|
|
|
def dropped(self, id="gone-thing", number=205, state="open"):
|
|
"""Pushed, and its file went with the push: ledger only."""
|
|
m = _gitea.load_map(self.root)
|
|
m[gmap.remote_key(REPO, number)] = id
|
|
_gitea.save_map(self.root, m)
|
|
self.fake.states[number] = state
|
|
return number
|
|
|
|
# -- runner ------------------------------------------------------------
|
|
|
|
def run_close(self, *argv):
|
|
self.out, self.err = io.StringIO(), io.StringIO()
|
|
args = ["close.py", "--repo", REPO, "--out", self.root] + list(argv)
|
|
with mock.patch.object(sys, "argv", args), \
|
|
contextlib.redirect_stdout(self.out), \
|
|
contextlib.redirect_stderr(self.err):
|
|
close.main()
|
|
return self.out.getvalue(), self.err.getvalue()
|
|
|
|
# -- assertions --------------------------------------------------------
|
|
|
|
def state_on_disk(self, id):
|
|
return issue.load(self.root, id).state
|
|
|
|
def raw(self, id):
|
|
with open(issue.path_of(self.root, id)) as f:
|
|
return f.read()
|
|
|
|
def assertUnchanged(self, id, before, why=""):
|
|
self.assertEqual(self.raw(id), before,
|
|
"%s.md was rewritten%s" % (id, why and " — " + why))
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# it closes
|
|
# --------------------------------------------------------------------------
|
|
|
|
class ClosesTest(StoreTestCase):
|
|
|
|
def test_a_slug_closes_the_issue_it_names(self):
|
|
self.synced("a-thing", 101)
|
|
out, _ = self.run_close("a-thing")
|
|
self.assertEqual(self.fake.states[101], "closed")
|
|
self.assertIn("closed a-thing #101", out)
|
|
|
|
def test_the_local_state_follows(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing")
|
|
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
|
|
|
def test_only_the_state_is_sent(self):
|
|
"""Closing is not an edit: no title, no body, no labels ride along."""
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing")
|
|
writes = self.fake.writes()
|
|
self.assertEqual(len(writes), 1)
|
|
method, endpoint, payload = writes[0]
|
|
self.assertEqual((method, endpoint), ("PATCH", "%s/issues/101" % BASE))
|
|
self.assertEqual(payload, {"state": "closed"})
|
|
|
|
def test_a_number_closes_it_too(self):
|
|
"""The normal case for a pushed issue — the file is long gone."""
|
|
self.synced("a-thing", 101)
|
|
self.run_close("101")
|
|
self.assertEqual(self.fake.states[101], "closed")
|
|
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
|
|
|
def test_every_key_form_is_accepted(self):
|
|
forms = {110: "110", 111: "#111", 112: "%s#112" % REPO,
|
|
113: "https://git.example/%s/issues/113" % REPO}
|
|
for n in forms:
|
|
self.fake.states[n] = "open"
|
|
for n, arg in forms.items():
|
|
with self.subTest(arg=arg):
|
|
self.run_close(arg)
|
|
self.assertEqual(self.fake.states[n], "closed")
|
|
|
|
def test_several_ids_in_one_run(self):
|
|
self.synced("a-thing", 101)
|
|
self.synced("b-thing", 102)
|
|
self.run_close("a-thing", "102")
|
|
self.assertEqual(self.fake.states, {101: "closed", 102: "closed"})
|
|
self.assertEqual(self.state_on_disk("a-thing"), "closed")
|
|
self.assertEqual(self.state_on_disk("b-thing"), "closed")
|
|
|
|
def test_the_same_issue_named_twice_is_written_once(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing", "#101")
|
|
self.assertEqual(len(self.fake.writes()), 1)
|
|
|
|
def test_the_index_is_rebuilt(self):
|
|
self.synced("a-thing", 101)
|
|
out, _ = self.run_close("a-thing")
|
|
self.assertIn("index:", out)
|
|
with open(os.path.join(self.root, "INDEX.md")) as f:
|
|
self.assertIn("closed", f.read())
|
|
|
|
def test_the_body_survives_untouched(self):
|
|
"""One metadata field changes; the prose and the ticks do not."""
|
|
self.synced("a-thing", 101)
|
|
before = issue.load(self.root, "a-thing").body
|
|
self.run_close("a-thing")
|
|
self.assertEqual(issue.load(self.root, "a-thing").body, before)
|
|
|
|
def test_synced_is_refreshed(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing")
|
|
iss = issue.load(self.root, "a-thing")
|
|
self.assertNotEqual(iss.extra.get("synced"), "2026-08-10T00:00:00Z")
|
|
self.assertEqual(iss.extra.get("remote-updated"), "2026-08-11T00:00:00Z")
|
|
|
|
def test_an_issue_whose_file_was_dropped_still_closes(self):
|
|
"""No local copy at all: the ledger names it, the tracker takes it, and
|
|
nothing is written locally."""
|
|
self.dropped("gone-thing", 205)
|
|
out, _ = self.run_close("gone-thing")
|
|
self.assertEqual(self.fake.states[205], "closed")
|
|
self.assertIn("no local copy", out)
|
|
self.assertNotIn("index:", out)
|
|
|
|
def test_a_number_nobody_here_knows_closes_without_a_slug(self):
|
|
self.fake.states[777] = "open"
|
|
out, _ = self.run_close("777")
|
|
self.assertEqual(self.fake.states[777], "closed")
|
|
self.assertIn("#777", out)
|
|
|
|
|
|
class ReopensTest(StoreTestCase):
|
|
|
|
def test_reopen_sends_open(self):
|
|
self.synced("a-thing", 101, state="closed")
|
|
out, _ = self.run_close("--reopen", "a-thing")
|
|
self.assertEqual(self.fake.writes()[0][2], {"state": "open"})
|
|
self.assertIn("reopened a-thing #101", out)
|
|
|
|
def test_reopen_writes_the_local_state_back(self):
|
|
self.synced("a-thing", 101, state="closed")
|
|
self.run_close("--reopen", "a-thing")
|
|
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
|
|
|
def test_close_then_reopen_is_a_round_trip(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing")
|
|
self.run_close("--reopen", "a-thing")
|
|
self.assertEqual(self.fake.states[101], "open")
|
|
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# it refuses
|
|
# --------------------------------------------------------------------------
|
|
|
|
class LocalOnlyTest(StoreTestCase):
|
|
"""An `origin: local` issue is not in the tracker, so it cannot be closed
|
|
there — and the local field is not quietly edited instead."""
|
|
|
|
def test_it_exits(self):
|
|
self.local_only("local-thing")
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("local-thing")
|
|
|
|
def test_the_error_names_the_id_and_says_it_is_not_in_the_tracker(self):
|
|
self.local_only("local-thing")
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("local-thing")
|
|
err = self.err.getvalue()
|
|
self.assertIn("local-thing", err)
|
|
self.assertIn("not in the tracker", err)
|
|
|
|
def test_nothing_is_sent(self):
|
|
self.local_only("local-thing")
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("local-thing")
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_the_file_is_untouched(self):
|
|
self.local_only("local-thing")
|
|
before = self.raw("local-thing")
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("local-thing")
|
|
self.assertUnchanged("local-thing", before)
|
|
|
|
def test_a_bad_id_stops_the_whole_run_before_anything_is_sent(self):
|
|
"""Resolution happens up front, so a typo in the second id does not
|
|
leave the first one closed."""
|
|
self.synced("a-thing", 101)
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("a-thing", "local-thing")
|
|
self.assertEqual(self.fake.states[101], "open")
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_an_unknown_slug_exits(self):
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("no-such-thing")
|
|
self.assertIn("no-such-thing", self.err.getvalue())
|
|
|
|
|
|
class DryRunTest(StoreTestCase):
|
|
|
|
def test_not_one_request_is_made(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("--dry-run", "a-thing")
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_the_file_is_untouched(self):
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.run_close("--dry-run", "a-thing")
|
|
self.assertUnchanged("a-thing", before, "--dry-run must write nothing")
|
|
|
|
def test_it_says_what_would_be_closed(self):
|
|
self.synced("a-thing", 101)
|
|
self.synced("b-thing", 102)
|
|
out, _ = self.run_close("--dry-run", "a-thing", "102")
|
|
self.assertIn("would close a-thing #101", out)
|
|
self.assertIn("would close b-thing #102", out)
|
|
self.assertIn("2 issue(s) would be closed", out)
|
|
|
|
def test_it_says_reopen_under_reopen(self):
|
|
self.synced("a-thing", 101, state="closed")
|
|
out, _ = self.run_close("--dry-run", "--reopen", "a-thing")
|
|
self.assertIn("would reopen a-thing #101", out)
|
|
self.assertIn("would be reopened", out)
|
|
|
|
def test_it_needs_no_login(self):
|
|
"""A dry run must work before /tea:auth has ever been run."""
|
|
self.synced("a-thing", 101)
|
|
with mock.patch.object(_gitea, "require_login",
|
|
lambda: self.fail("dry run asked for a login")):
|
|
self.run_close("--dry-run", "a-thing")
|
|
|
|
def test_a_local_only_issue_is_still_refused(self):
|
|
self.local_only("local-thing")
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("--dry-run", "local-thing")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# the tracker said no
|
|
# --------------------------------------------------------------------------
|
|
|
|
class TrackerFailureTest(StoreTestCase):
|
|
"""The criterion that matters most: a write that was not confirmed leaves
|
|
the local file exactly as it was."""
|
|
|
|
def test_a_non_2xx_answer_leaves_the_file(self):
|
|
"""The real `_gitea.api` against a `tea` that exits 1 — the path a 422
|
|
or a 500 actually takes, and it ends in `die()`."""
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
|
|
def fake_run(cmd, capture_output=False, text=False):
|
|
return types.SimpleNamespace(
|
|
returncode=1, stdout="",
|
|
stderr="422 Unprocessable Entity: issue is blocked")
|
|
|
|
with mock.patch.object(_gitea, "api", REAL_API), \
|
|
mock.patch.object(_gitea, "subprocess",
|
|
types.SimpleNamespace(run=fake_run)), \
|
|
self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
|
|
self.assertUnchanged("a-thing", before, "tea exited non-zero")
|
|
self.assertEqual(self.state_on_disk("a-thing"), "open")
|
|
|
|
def test_a_transport_exception_leaves_the_file(self):
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.fake.raise_on_write = OSError("tea: command not found")
|
|
with self.assertRaises(OSError):
|
|
self.run_close("a-thing")
|
|
self.assertUnchanged("a-thing", before, "the transport raised")
|
|
|
|
def test_an_answer_without_a_number_leaves_the_file(self):
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.fake.answer_override = {"ok": True, "state": "closed"}
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
self.assertUnchanged("a-thing", before)
|
|
|
|
def test_an_answer_for_another_issue_leaves_the_file(self):
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.fake.answer_override = {"number": 999, "state": "closed"}
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
self.assertUnchanged("a-thing", before)
|
|
|
|
def test_an_answer_that_did_not_change_the_state_leaves_the_file(self):
|
|
"""A 200 that still says `open` is not a close."""
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.fake.answer_override = {"number": 101, "state": "open"}
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
self.assertUnchanged("a-thing", before)
|
|
|
|
def test_an_empty_answer_leaves_the_file(self):
|
|
self.synced("a-thing", 101)
|
|
before = self.raw("a-thing")
|
|
self.fake.answer_override = None
|
|
real_api = self.fake.api
|
|
self.fake.api = lambda *a, **kw: (real_api(*a, **kw), None)[1]
|
|
with mock.patch.object(_gitea, "api", self.fake.api), \
|
|
self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
self.assertUnchanged("a-thing", before)
|
|
|
|
def test_the_error_says_nothing_local_changed(self):
|
|
self.synced("a-thing", 101)
|
|
self.fake.answer_override = {"ok": True}
|
|
with self.assertRaises(SystemExit):
|
|
self.run_close("a-thing")
|
|
self.assertIn("Nothing local was changed", self.err.getvalue())
|
|
|
|
def test_a_failure_partway_through_keeps_the_rest(self):
|
|
"""Two issues, the second one is not confirmed. The first is
|
|
legitimately closed; the second's file still says open."""
|
|
self.synced("aaa-thing", 101)
|
|
self.synced("zzz-thing", 102)
|
|
before = self.raw("zzz-thing")
|
|
|
|
real = self.fake.api
|
|
seen = []
|
|
|
|
def once(login, endpoint, method="GET", payload=None, **kw):
|
|
got = real(login, endpoint, method, payload, **kw)
|
|
if method != "GET":
|
|
seen.append(endpoint)
|
|
return {"nope": True} if len(seen) > 1 else got
|
|
|
|
with mock.patch.object(_gitea, "api", once), \
|
|
self.assertRaises(SystemExit):
|
|
self.run_close("aaa-thing", "zzz-thing")
|
|
|
|
self.assertEqual(self.state_on_disk("aaa-thing"), "closed")
|
|
self.assertUnchanged("zzz-thing", before, "its write was not confirmed")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# the pure parts
|
|
# --------------------------------------------------------------------------
|
|
|
|
class ConfirmedTest(unittest.TestCase):
|
|
"""The gate itself. Everything below it rewrites a file."""
|
|
|
|
def test_a_matching_close_is_confirmed(self):
|
|
self.assertTrue(close.confirmed({"number": 42, "state": "closed"}, 42, "closed"))
|
|
|
|
def test_a_mismatched_number_is_not(self):
|
|
self.assertFalse(close.confirmed({"number": 43, "state": "closed"}, 42, "closed"))
|
|
|
|
def test_the_wrong_state_is_not(self):
|
|
self.assertFalse(close.confirmed({"number": 42, "state": "open"}, 42, "closed"))
|
|
|
|
def test_a_missing_state_is_not(self):
|
|
self.assertFalse(close.confirmed({"number": 42}, 42, "closed"))
|
|
|
|
def test_none_and_lists_are_not(self):
|
|
self.assertFalse(close.confirmed(None, 42, "closed"))
|
|
self.assertFalse(close.confirmed([{"number": 42, "state": "closed"}], 42, "closed"))
|
|
|
|
def test_true_is_not_a_number(self):
|
|
self.assertFalse(close.confirmed({"number": True, "state": "closed"}, 1, "closed"))
|
|
|
|
def test_a_string_number_is_not(self):
|
|
self.assertFalse(close.confirmed({"number": "42", "state": "closed"}, 42, "closed"))
|
|
|
|
|
|
class KeyFormTest(unittest.TestCase):
|
|
"""A slug and a key are two vocabularies that must not collide."""
|
|
|
|
def test_keys_are_keys(self):
|
|
for k in ("42", "#42", "owner/repo#42",
|
|
"https://git.example/owner/repo/issues/42"):
|
|
self.assertTrue(close.looks_like_key(k), k)
|
|
|
|
def test_slugs_are_not_keys(self):
|
|
for s in ("a-thing", "wire-sqlc-appclick", "close-issues-through-a-script"):
|
|
self.assertFalse(close.looks_like_key(s), s)
|
|
|
|
|
|
class LedgerPairsTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.m = {"%s#7" % REPO: "a-thing", "other/repo#7": "b-thing",
|
|
"not-a-key": "c-thing"}
|
|
|
|
def test_it_filters_by_repo(self):
|
|
self.assertEqual(close.ledger_pairs(self.m, REPO), [(REPO, 7, "a-thing")])
|
|
|
|
def test_without_a_repo_it_keeps_everything_parseable(self):
|
|
got = close.ledger_pairs(self.m)
|
|
self.assertEqual(sorted(s for _r, _n, s in got), ["a-thing", "b-thing"])
|
|
|
|
def test_an_ambiguous_number_exits(self):
|
|
pairs = close.ledger_pairs(self.m)
|
|
with self.assertRaises(SystemExit):
|
|
with contextlib.redirect_stderr(io.StringIO()):
|
|
close.resolve("7", {}, pairs)
|
|
|
|
|
|
class AmbiguityTest(StoreTestCase):
|
|
"""Two repos, one number, no --repo: settle it rather than guess."""
|
|
|
|
def test_the_error_points_at_repo(self):
|
|
_gitea.save_map(self.root, {"%s#7" % REPO: "a-thing",
|
|
"other/repo#7": "b-thing"})
|
|
err = io.StringIO()
|
|
args = ["close.py", "--out", self.root, "7"]
|
|
with mock.patch.object(sys, "argv", args), \
|
|
contextlib.redirect_stdout(io.StringIO()), \
|
|
contextlib.redirect_stderr(err), \
|
|
self.assertRaises(SystemExit):
|
|
close.main()
|
|
self.assertIn("--repo", err.getvalue())
|
|
|
|
|
|
class RepoOfTheKeyTest(StoreTestCase):
|
|
"""A key that names its own repo is sent there, not to whatever repo the
|
|
CWD happens to be — otherwise `#42` closes somebody else's issue."""
|
|
|
|
def run_bare(self, *argv):
|
|
"""No `--repo`, so the ids have to say where they live."""
|
|
self.out, self.err = io.StringIO(), io.StringIO()
|
|
args = ["close.py", "--out", self.root] + list(argv)
|
|
with mock.patch.object(sys, "argv", args), \
|
|
contextlib.redirect_stdout(self.out), \
|
|
contextlib.redirect_stderr(self.err):
|
|
close.main()
|
|
return self.out.getvalue(), self.err.getvalue()
|
|
|
|
def test_a_foreign_key_goes_to_its_own_repo(self):
|
|
self.run_bare("other/repo#42")
|
|
self.assertEqual(self.fake.writes()[0][1], "repos/other/repo/issues/42")
|
|
|
|
def test_a_slug_goes_to_the_repo_its_gitea_field_names(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_bare("a-thing")
|
|
self.assertEqual(self.fake.writes()[0][1], "%s/issues/101" % BASE)
|
|
|
|
def test_two_repos_in_one_run_is_a_question_not_a_guess(self):
|
|
self.synced("a-thing", 101)
|
|
with self.assertRaises(SystemExit):
|
|
self.run_bare("a-thing", "other/repo#42")
|
|
self.assertIn("one repo", self.err.getvalue())
|
|
self.assertEqual(self.fake.calls, [])
|
|
|
|
def test_an_explicit_repo_settles_it(self):
|
|
self.synced("a-thing", 101)
|
|
self.run_close("a-thing", "other/repo#42")
|
|
self.assertEqual({c[1] for c in self.fake.writes()},
|
|
{"%s/issues/101" % BASE, "%s/issues/42" % BASE})
|
|
|
|
|
|
class NoStoreTest(StoreTestCase):
|
|
"""A number needs no local file, and a store that is not there is not an
|
|
error — closing an issue whose copy push dropped is the normal case."""
|
|
|
|
def test_a_number_closes_with_no_store_at_all(self):
|
|
missing = os.path.join(self.root, "nowhere")
|
|
self.fake.states[303] = "open"
|
|
args = ["close.py", "--repo", REPO, "--out", missing, "303"]
|
|
with mock.patch.object(sys, "argv", args), \
|
|
contextlib.redirect_stdout(io.StringIO()), \
|
|
contextlib.redirect_stderr(io.StringIO()):
|
|
close.main()
|
|
self.assertEqual(self.fake.states[303], "closed")
|
|
self.assertFalse(os.path.isdir(missing), "no store was conjured")
|
|
|
|
|
|
class PayloadFileTest(StoreTestCase):
|
|
"""The request body goes to the transport's own scratchpad.
|
|
|
|
Not to a directory this script picks: `close.py` names the payload and
|
|
nothing else, the way every other caller does. Where PAYLOAD_ROOT lands is
|
|
_gitea's business, and test_payload_root.py is where that is tested."""
|
|
|
|
def test_the_payload_lands_in_the_transports_scratchpad(self):
|
|
self.synced("a-thing", 101)
|
|
payloads = os.path.join(self.root, "payload")
|
|
with mock.patch.object(_gitea, "PAYLOAD_ROOT", payloads), \
|
|
mock.patch.object(_gitea, "api", REAL_API), \
|
|
mock.patch.object(
|
|
_gitea, "subprocess",
|
|
types.SimpleNamespace(run=lambda cmd, **kw: types.SimpleNamespace(
|
|
returncode=0, stderr="",
|
|
stdout=json.dumps({"number": 101, "state": "closed"})))):
|
|
self.run_close("a-thing")
|
|
p = os.path.join(payloads, "state-101.json")
|
|
self.assertTrue(os.path.isfile(p))
|
|
with open(p) as f:
|
|
self.assertEqual(json.load(f), {"state": "closed"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|