#!/usr/bin/env python3 """ Where request bodies land, and that writing one never conjures a store. python3 -m unittest discover -s tests -v Stdlib unittest, no third-party anything. The bug these tests pin down: `labels.py --bootstrap` on a fresh checkout left `tmp/issues/.payload/` behind, because the only place `_gitea.api` had to put a request file was whatever root the caller handed it — and the label bootstrap, which touches no issue at all, handed it the issue store. A store materialized as a side effect of an operation that has nothing to do with issues. Every run here is against a throwaway repository with a FAKE `tea` first on PATH, so nothing reaches the network and the developer's own store is never in the blast radius. """ import json import os import shutil import stat import subprocess import sys import tempfile import unittest REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ISSUE_SCRIPTS = os.path.join(REPO, "skills", "issue", "scripts") SYNC_SCRIPTS = os.path.join(REPO, "skills", "sync", "scripts") AUTH_SCRIPTS = os.path.join(REPO, "skills", "auth", "scripts") sys.path.insert(0, SYNC_SCRIPTS) sys.path.insert(0, ISSUE_SCRIPTS) import _gitea # noqa: E402 import issue # noqa: E402 # A `tea` that answers without a network: an empty list for every GET (so the # repository looks like it has no labels yet) and a created object for every # write. It also records its own argv, which is how a test can tell that the # payload file the script wrote is the one the call actually referenced. FAKE_TEA = '''#!%s import json, os, sys with open(os.path.join(os.environ["TEA_CALL_LOG"], "calls.txt"), "a") as f: f.write("\\t".join(sys.argv[1:]) + "\\n") sys.stdout.write(json.dumps({"id": 1, "name": "created"}) if "-X" in sys.argv else "[]") ''' class FakeRepo(object): """A self-contained repository with no store and no tmp/ at all.""" def __init__(self): self._tmp = tempfile.TemporaryDirectory() # realpath: on macOS $TMPDIR is a symlink, and a child reporting its # own cwd would otherwise disagree with the path we handed it. self.root = os.path.realpath(self._tmp.name) os.makedirs(os.path.join(self.root, ".git")) # the repo marker skip = shutil.ignore_patterns("__pycache__") shutil.copytree(ISSUE_SCRIPTS, self.path("skills", "issue", "scripts"), ignore=skip) shutil.copytree(SYNC_SCRIPTS, self.path("skills", "sync", "scripts"), ignore=skip) # the transport resolves the login pin through skills/auth/scripts shutil.copytree(AUTH_SCRIPTS, self.path("skills", "auth", "scripts"), ignore=skip) os.makedirs(self.path("sub", "deeper")) # the login pin the transport insists on, local to this fixture os.makedirs(self.path(".claude")) with open(self.path(".claude", "settings.local.json"), "w") as f: json.dump({"env": {"GITEA_LOGIN": "fixture/user"}}, f) self.bin = self.path("fakebin") os.makedirs(self.bin) tea = os.path.join(self.bin, "tea") with open(tea, "w") as f: f.write(FAKE_TEA % sys.executable) os.chmod(tea, os.stat(tea).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) def cleanup(self): self._tmp.cleanup() def path(self, *parts): return os.path.join(self.root, *parts) @property def store(self): return self.path("tmp", "issues") @property def payloads(self): return self.path("tmp", "payload") def script(self, layer, name): return self.path("skills", layer, "scripts", name) def run(self, script, *args, **kw): env = dict(os.environ) env.pop("PYTHONPATH", None) # no leakage from the harness into the child env["PATH"] = self.bin + os.pathsep + env["PATH"] env["TEA_CALL_LOG"] = self.root p = subprocess.run([sys.executable, script] + list(args), cwd=kw.pop("cwd", self.root), env=env, capture_output=True, text=True) return p.returncode, p.stdout, p.stderr def calls(self): p = os.path.join(self.root, "calls.txt") if not os.path.isfile(p): return [] with open(p) as f: return [line.rstrip("\n").split("\t") for line in f if line.strip()] # -------------------------------------------------------------------------- # resolution # -------------------------------------------------------------------------- class TestPayloadRoot(unittest.TestCase): def test_root_is_absolute_and_repo_anchored(self): self.assertTrue(os.path.isabs(_gitea.PAYLOAD_ROOT), _gitea.PAYLOAD_ROOT) self.assertEqual(_gitea.PAYLOAD_ROOT, os.path.join(REPO, "tmp", "payload")) def test_it_is_not_the_issue_store_and_not_inside_one(self): """The acceptance criterion, as a path fact: a request body is not store content, so it may not live in a store or under one.""" self.assertNotEqual(_gitea.PAYLOAD_ROOT, issue.ISSUE_ROOT) self.assertFalse(_gitea.PAYLOAD_ROOT.startswith(issue.ISSUE_ROOT + os.sep)) self.assertFalse(issue.ISSUE_ROOT.startswith(_gitea.PAYLOAD_ROOT + os.sep)) def test_the_name_says_what_it_holds(self): """Named so the distinction is visible: a top-level directory called `payload`, not a dotdir hiding among an issue's files.""" self.assertEqual(os.path.basename(_gitea.PAYLOAD_ROOT), "payload") self.assertFalse(os.path.basename(_gitea.PAYLOAD_ROOT).startswith(".")) def test_gitignore_covers_it(self): with open(os.path.join(REPO, ".gitignore")) as f: ignored = {line.strip() for line in f} self.assertEqual(_gitea.PAYLOAD_PARTS[0], "tmp") self.assertIn("tmp/", ignored, "the payload directory is not covered by .gitignore") def test_resolution_is_anchored_on_the_module_not_on_cwd(self): repo = FakeRepo() self.addCleanup(repo.cleanup) self.assertEqual(_gitea.payload_root(repo.path("sub", "deeper")), repo.payloads) # -------------------------------------------------------------------------- # the bug: a label bootstrap that materialized the store # -------------------------------------------------------------------------- class TestLabelsTouchesNoStore(unittest.TestCase): def setUp(self): self.repo = FakeRepo() self.addCleanup(self.repo.cleanup) def bootstrap(self, *args, **kw): rc, out, err = self.repo.run(self.repo.script("sync", "labels.py"), "--repo", "fixture/repo", *args, **kw) self.assertEqual(rc, 0, "labels.py failed:\n%s%s" % (out, err)) return out, err def test_bootstrap_creates_no_store(self): """The reproduction from the report, run for real: no tmp/issues, and no complaint about one either.""" out, _ = self.bootstrap() self.assertIn("created", out) self.assertFalse(os.path.exists(self.repo.store), "labels.py created the issue store") def test_bootstrap_writes_its_payloads_to_the_payload_root(self): self.bootstrap() self.assertTrue(os.path.isdir(self.repo.payloads), "no payload directory: %s" % self.repo.payloads) written = os.listdir(self.repo.payloads) self.assertIn("label-type-bug.json", written) for name in written: self.assertTrue(name.startswith("label-"), name) # and the file named on the command line is the one that was written sent = [a[a.index("-d") + 1][1:] for a in self.repo.calls() if "-d" in a] self.assertTrue(sent) for path in sent: self.assertEqual(os.path.dirname(path), self.repo.payloads) self.assertTrue(os.path.isfile(path), path) def test_the_payload_is_the_request_body(self): self.bootstrap() with open(os.path.join(self.repo.payloads, "label-type-bug.json")) as f: body = json.load(f) self.assertEqual(body.get("name"), "type/bug") self.assertTrue(body.get("color")) def test_a_dry_run_writes_nothing_at_all(self): out, _ = self.bootstrap("--dry-run") self.assertIn("nothing was written", out) self.assertFalse(os.path.exists(self.repo.path("tmp")), "a dry run left something behind in tmp/") def test_the_directory_does_not_follow_cwd(self): """Run from a subdirectory: still one payload root, at the repo root. A cwd-relative directory is how the store ended up with a second copy of itself, and this one is resolved the same way to avoid the same class of bug.""" self.bootstrap(cwd=self.repo.path("sub", "deeper")) self.assertTrue(os.path.isdir(self.repo.payloads)) self.assertFalse(os.path.exists(self.repo.path("sub", "deeper", "tmp"))) self.assertFalse(os.path.exists(self.repo.store)) # -------------------------------------------------------------------------- # one place, every caller # -------------------------------------------------------------------------- class TestOnePlaceForEveryCaller(unittest.TestCase): def hits(self, needle, skip_transport=False): """Every `layer/script.py:line` mentioning `needle`.""" out = [] d = SYNC_SCRIPTS layer = os.path.basename(os.path.dirname(d)) for name in sorted(os.listdir(d)): if not name.endswith(".py") or (skip_transport and name == "_gitea.py"): continue with open(os.path.join(d, name)) as f: for n, line in enumerate(f, 1): if needle in line: out.append("%s/%s:%d" % (layer, name, n)) return out def test_no_caller_chooses_where_its_payload_goes(self): """Whatever the answer is, it has to be the same for all of them — payload files scattered across the stores of whichever command wrote them is the state this replaced.""" self.assertEqual(self.hits("out_root"), [], "a caller still picks a payload directory of its own") def test_only_the_transport_names_the_directory(self): self.assertEqual(self.hits("PAYLOAD", skip_transport=True), [], "the payload directory is named outside the transport") if __name__ == "__main__": unittest.main()