marketplace/plugins/tea/tests/test_payload_root.py

#!/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):
    """An initialized project with no store and nothing under `.tea/` yet.

    The scripts are NOT copied in: they stay at their real installed path, so
    what these tests exercise is a plugin operating on somebody else's project
    — which is every use of it but this repository's own."""

    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, issue.MARKER))  # the project marker
        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(*issue.STORE_PARTS)

    @property
    def payloads(self):
        return self.path(*_gitea.PAYLOAD_PARTS)

    def script(self, layer, name):
        """The real script, several directories away from this fixture."""
        return os.path.join(REPO, "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
        # the first anchor of both walks: left in place, every fixture would
        # resolve to this repository instead of itself
        env.pop("CLAUDE_PROJECT_DIR", None)
        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 setUp(self):
        self.repo = FakeRepo()
        self.addCleanup(self.repo.cleanup)

    def test_root_is_absolute_and_project_anchored(self):
        root = _gitea.payload_root(self.repo.path("sub", "deeper"))
        self.assertTrue(os.path.isabs(root), root)
        self.assertEqual(root, self.repo.payloads)

    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."""
        start = self.repo.path("sub", "deeper")
        payload = _gitea.payload_root(start)
        store = issue.store_root(start)
        self.assertNotEqual(payload, store)
        self.assertFalse(payload.startswith(store + os.sep))
        self.assertFalse(store.startswith(payload + os.sep))

    def test_the_name_says_what_it_holds(self):
        """Named so the distinction is visible: `payload`, a sibling of the
        store under the marker, not a dotdir hiding among an issue's files."""
        payload = _gitea.payload_root(self.repo.root)
        self.assertEqual(os.path.basename(payload), "payload")
        self.assertEqual(os.path.dirname(payload),
                         self.repo.path(issue.MARKER))

    def test_no_project_means_no_payload_root(self):
        """Same answer as the store gives: nothing, rather than a directory
        picked because it was the only one at hand."""
        plain = tempfile.TemporaryDirectory()
        self.addCleanup(plain.cleanup)
        self.assertIsNone(_gitea.payload_root(os.path.realpath(plain.name)))

    def test_gitignore_covers_it(self):
        """The rule is that the marker is ignored, not which file says so: this
        plugin lives under `plugins/` in a marketplace repo, and git reads every
        .gitignore on the way up. So walk up the same way git does."""
        ignored = set()
        d = REPO
        while True:
            p = os.path.join(d, ".gitignore")
            if os.path.isfile(p):
                with open(p) as f:
                    ignored |= {line.strip() for line in f}
            parent = os.path.dirname(d)
            if parent == d or os.path.isdir(os.path.join(d, ".git")):
                break
            d = parent
        self.assertEqual(_gitea.PAYLOAD_PARTS[0], issue.MARKER)
        self.assertIn(issue.MARKER + "/", ignored,
                      "the payload directory is not covered by .gitignore")

    def test_resolution_follows_the_project_not_the_module(self):
        """The bug, as a path fact: the scripts live somewhere else entirely,
        and the answer is still this project's directory."""
        self.assertEqual(_gitea.payload_root(self.repo.path("sub", "deeper")),
                         self.repo.payloads)
        self.assertFalse(self.repo.payloads.startswith(REPO + os.sep))


# --------------------------------------------------------------------------
# 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.assertEqual(os.listdir(self.repo.path(issue.MARKER)), [],
                         "a dry run left something behind under the marker")

    def test_the_directory_does_not_follow_cwd(self):
        """Run from a subdirectory: still one payload root, at the project
        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", issue.MARKER)))
        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()