marketplace/tests/test_store_path.py

#!/usr/bin/env python3
"""
Where the issue store is, and that the answer does not depend on cwd.

    python3 -m unittest discover -s tests -v

Stdlib unittest, no third-party anything — the same rule the scripts under test
live by. `skills/*/scripts/` are not packages, so the domain module is imported
by path.

Most of these tests do not touch this repository at all. They build a throwaway
repo in a temp directory — a `.git` marker, a copy of both script layers, a
store with two issues — and run the real scripts inside it as subprocesses with
different working directories. That is the only honest way to test a cwd bug:
importing the module would resolve the store once, against the wrong tree.
"""
import os
import shutil
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")

sys.path.insert(0, ISSUE_SCRIPTS)
import issue  # noqa: E402


ALPHA = """\
---
id: alpha-issue
state: open
labels: [type/task]
assignees: []
milestone: none
depends: []
origin: local
---
# Alpha issue

## Summary
Первый issue фикстуры.

## Spec
none

## Motivation
Нужен, чтобы в store что-то лежало.

## Acceptance criteria
- [ ] проверяемое условие
"""

BETA = """\
---
id: beta-issue
state: open
labels: [type/task]
assignees: []
milestone: none
depends: [alpha-issue]
origin: local
---
# Beta issue

## Summary
Второй issue фикстуры, зависит от первого.

## Spec
none

## Depends on
- alpha-issue

## Motivation
Нужен, чтобы у графа было ребро.

## Acceptance criteria
- [ ] проверяемое условие
"""


def run(script, *args, **kw):
    """Run one of the plugin's scripts and return (rc, stdout, stderr)."""
    cwd = kw.pop("cwd")
    env = dict(os.environ)
    env.pop("PYTHONPATH", None)  # no leakage from the harness into the child
    p = subprocess.run([sys.executable, script] + list(args), cwd=cwd, env=env,
                       capture_output=True, text=True)
    return p.returncode, p.stdout, p.stderr


class FakeRepo(object):
    """A self-contained repository in a temp directory.

    Both script layers are copied in, so `__file__`-anchored resolution lands
    inside the fixture and never on the developer's real store.
    """

    def __init__(self, with_store=True, issues=(ALPHA, BETA)):
        self._tmp = tempfile.TemporaryDirectory()
        # realpath: on macOS $TMPDIR is a symlink, and a child process 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)
        os.makedirs(self.path("sub", "deeper"))

        if with_store:
            os.makedirs(self.store)
            for text in issues:
                id = text.split("id: ", 1)[1].split("\n", 1)[0]
                with open(os.path.join(self.store, "%s.md" % id), "w") as f:
                    f.write(text)

    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")

    def script(self, layer, name):
        return self.path("skills", layer, "scripts", name)

    def everywhere(self):
        """Working directories that must all produce the same answer: the repo
        root, a plain subdirectory, a deeper one, the script directory itself,
        and — the case from the bug report — inside the store."""
        return [self.root, self.path("sub"), self.path("sub", "deeper"),
                self.path("skills", "issue", "scripts"), self.store]


# --------------------------------------------------------------------------
# resolution, in isolation
# --------------------------------------------------------------------------

class TestResolution(unittest.TestCase):

    def setUp(self):
        self.repo = FakeRepo()
        self.addCleanup(self.repo.cleanup)

    def test_repo_root_found_from_any_depth(self):
        for start in self.repo.everywhere():
            self.assertEqual(issue.repo_root(start), self.repo.root, start)

    def test_agents_md_works_as_a_marker(self):
        """A checkout without .git — the plugin copied out of git — still
        resolves, because AGENTS.md marks the root too."""
        shutil.rmtree(self.repo.path(".git"))
        open(self.repo.path("AGENTS.md"), "w").close()
        self.assertEqual(issue.repo_root(self.repo.path("sub", "deeper")),
                         self.repo.root)

    def test_nearest_marker_wins(self):
        """A repo inside a repo (a worktree, a vendored copy) resolves to the
        inner one, not the outer."""
        inner = self.repo.path("sub", "inner")
        os.makedirs(os.path.join(inner, ".git"))
        self.assertEqual(issue.repo_root(inner), inner)
        self.assertEqual(issue.repo_root(self.repo.root), self.repo.root)

    def test_store_root_is_repo_root_plus_tmp_issues(self):
        self.assertEqual(issue.store_root(self.repo.path("sub", "deeper")),
                         self.repo.store)

    def test_default_root_is_absolute(self):
        """The whole point: a default that cannot mean two directories."""
        self.assertTrue(os.path.isabs(issue.ISSUE_ROOT), issue.ISSUE_ROOT)
        self.assertEqual(issue.ISSUE_ROOT,
                         os.path.join(REPO, "tmp", "issues"))


# --------------------------------------------------------------------------
# the acceptance criterion: same answer from any subdirectory
# --------------------------------------------------------------------------

class TestSameFromAnywhere(unittest.TestCase):

    def setUp(self):
        self.repo = FakeRepo()
        self.addCleanup(self.repo.cleanup)

    def assertSameEverywhere(self, layer, name, *args):
        """Run the script from the repo root and from every other directory;
        every result must be byte-identical to the one from the root."""
        dirs = self.repo.everywhere()
        base = run(self.repo.script(layer, name), *args, cwd=dirs[0])
        self.assertEqual(base[0], 0, "%s failed at the repo root:\n%s" % (name, base[2]))
        for d in dirs[1:]:
            self.assertEqual(run(self.repo.script(layer, name), *args, cwd=d), base,
                             "%s disagrees when run from %s" % (name, d))
        return base

    def test_issue_check(self):
        rc, out, _ = self.assertSameEverywhere("issue", "issue_check.py")
        self.assertIn("ok    alpha-issue", out)
        self.assertIn("2 issue(s) checked, 0 with errors", out)

    def test_issue_tree(self):
        _, out, _ = self.assertSameEverywhere("issue", "issue_tree.py")
        self.assertIn("beta-issue", out)
        self.assertIn("alpha-issue", out)

    def test_issue_index(self):
        _, out, _ = self.assertSameEverywhere("issue", "issue_index.py")
        self.assertIn("2 issue(s)", out)
        self.assertIn(os.path.join(self.repo.store, "INDEX.md"), out)

    def test_no_second_store_is_ever_created(self):
        """The bug's worst symptom: `issue_index.py` run from inside the store
        used to leave tmp/issues/tmp/issues/ behind, silently."""
        for d in self.repo.everywhere():
            for name in ("issue_index.py", "issue_check.py", "issue_tree.py"):
                run(self.repo.script("issue", name), cwd=d)

        found = []
        for dirpath, dirnames, filenames in os.walk(self.repo.root):
            if "__pycache__" in dirnames:
                dirnames.remove("__pycache__")
            if "INDEX.md" in filenames:
                found.append(dirpath)
        self.assertEqual(found, [self.repo.store],
                         "a second store appeared: %s" % found)


# --------------------------------------------------------------------------
# missing is not empty
# --------------------------------------------------------------------------

class TestMissingVersusEmpty(unittest.TestCase):

    def test_missing_store_says_missing(self):
        repo = FakeRepo(with_store=False)
        self.addCleanup(repo.cleanup)
        for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
            rc, out, err = run(repo.script("issue", name), cwd=repo.root)
            msg = out + err
            self.assertNotEqual(rc, 0, "%s should fail on a missing store" % name)
            self.assertIn("does not exist", msg, name)
            self.assertNotIn("is empty", msg, name)

    def test_empty_store_says_empty(self):
        repo = FakeRepo(issues=())
        self.addCleanup(repo.cleanup)
        for name in ("issue_check.py", "issue_tree.py"):
            rc, out, err = run(repo.script("issue", name), cwd=repo.root)
            msg = out + err
            self.assertNotEqual(rc, 0, name)
            self.assertIn("is empty", msg, name)
            self.assertNotIn("does not exist", msg, name)

    def test_index_of_an_empty_store_is_legitimate(self):
        """An existing store with nothing in it gets an index saying so. Only a
        missing directory is an error."""
        repo = FakeRepo(issues=())
        self.addCleanup(repo.cleanup)
        rc, out, err = run(repo.script("issue", "issue_index.py"), cwd=repo.root)
        self.assertEqual(rc, 0, err)
        self.assertIn("0 issue(s)", out)
        with open(os.path.join(repo.store, "INDEX.md")) as f:
            self.assertIn("_empty_", f.read())


# --------------------------------------------------------------------------
# nothing conjures a store
# --------------------------------------------------------------------------

class TestNoSilentCreation(unittest.TestCase):

    def setUp(self):
        self.repo = FakeRepo(with_store=False)
        self.addCleanup(self.repo.cleanup)

    def test_readers_and_the_indexer_create_nothing(self):
        for d in (self.repo.root, self.repo.path("sub")):
            for name in ("issue_check.py", "issue_tree.py", "issue_index.py"):
                run(self.repo.script("issue", name), cwd=d)
        self.assertFalse(os.path.exists(self.repo.path("tmp")),
                         "the store was created by a read")
        self.assertFalse(os.path.exists(self.repo.path("sub", "tmp")),
                         "a store was created relative to cwd")

    def test_explicit_out_pointing_nowhere_is_an_error_not_a_mkdir(self):
        target = self.repo.path("sub", "nowhere")
        rc, out, err = run(self.repo.script("issue", "issue_index.py"),
                           "--out", target, cwd=self.repo.root)
        self.assertNotEqual(rc, 0)
        self.assertIn("does not exist", out + err)
        self.assertFalse(os.path.exists(target))

    def test_issue_new_creates_the_store_and_says_so(self):
        """Creating the first issue in a fresh checkout must still work — but
        out loud, and at the repo root, not below whatever cwd happens to be."""
        rc, out, err = run(self.repo.script("issue", "issue_new.py"),
                           "--type", "task", "--title", "Bootstrap the store",
                           cwd=self.repo.path("sub", "deeper"))
        self.assertEqual(rc, 0, err)
        self.assertIn("created store", err)
        self.assertIn(self.repo.store, err)
        self.assertTrue(os.path.isfile(
            os.path.join(self.repo.store, "bootstrap-the-store.md")))
        self.assertFalse(os.path.exists(self.repo.path("sub", "deeper", "tmp")),
                         "a store was created relative to cwd")


# --------------------------------------------------------------------------
# an explicit --out is the operator's, not ours to rewrite
# --------------------------------------------------------------------------

class TestExplicitOutWins(unittest.TestCase):

    def setUp(self):
        self.repo = FakeRepo()
        self.addCleanup(self.repo.cleanup)

    def test_absolute_out_is_honored(self):
        other = self.repo.path("sub", "other-store")
        os.makedirs(other)
        shutil.copy(os.path.join(self.repo.store, "alpha-issue.md"), other)
        rc, out, err = run(self.repo.script("issue", "issue_check.py"),
                           "--out", other, cwd=self.repo.root)
        self.assertEqual(rc, 0, err)
        self.assertIn("1 issue(s) checked", out)

    def test_relative_out_stays_relative_to_cwd(self):
        """`--out tmp/issues` typed from a subdirectory means that
        subdirectory's tmp/issues — which is not there. Auto-resolution must
        not step in and "fix" what the operator typed."""
        rc, out, err = run(self.repo.script("issue", "issue_check.py"),
                           "--out", os.path.join("tmp", "issues"),
                           cwd=self.repo.path("sub"))
        self.assertNotEqual(rc, 0)
        self.assertIn("does not exist", out + err)

        # the same relative path from the root does resolve, by cwd alone
        rc, out, err = run(self.repo.script("issue", "issue_check.py"),
                           "--out", os.path.join("tmp", "issues"),
                           cwd=self.repo.root)
        self.assertEqual(rc, 0, err)
        self.assertIn("2 issue(s) checked", out)

    def test_relative_out_can_climb(self):
        rc, out, err = run(self.repo.script("issue", "issue_check.py"),
                           "--out", os.path.join("..", "tmp", "issues"),
                           cwd=self.repo.path("sub"))
        self.assertEqual(rc, 0, err)
        self.assertIn("2 issue(s) checked", out)


# --------------------------------------------------------------------------
# both layers, one root
# --------------------------------------------------------------------------

class TestSyncLayerAgrees(unittest.TestCase):

    def setUp(self):
        self.repo = FakeRepo()
        self.addCleanup(self.repo.cleanup)

    def _probe(self, layer, cwd):
        """Ask one layer, from `cwd`, which module defines the store and where
        it lands. The sync scripts put the issue scripts on sys.path themselves
        — `import map` is how they do it — so each layer is asked its own way.
        """
        scripts = self.repo.path("skills", layer, "scripts")
        entry = "import map, issue" if layer == "sync" else "import issue"
        code = ("import sys; sys.path.insert(0, %r)\n%s\n"
                "print(issue.__file__)\nprint(issue.ISSUE_ROOT)\n") % (scripts, entry)
        env = dict(os.environ)
        env.pop("PYTHONPATH", None)
        p = subprocess.run([sys.executable, "-c", code], cwd=cwd, env=env,
                           capture_output=True, text=True)
        self.assertEqual(p.returncode, 0, p.stderr)
        return p.stdout.strip().splitlines()

    def test_both_layers_resolve_the_same_store_from_anywhere(self):
        for d in self.repo.everywhere():
            mod_i, root_i = self._probe("issue", d)
            mod_s, root_s = self._probe("sync", d)
            # sync does not redefine the store; it imports the domain module
            self.assertEqual(os.path.realpath(mod_i), os.path.realpath(mod_s), d)
            self.assertEqual(root_i, self.repo.store, d)
            self.assertEqual(root_s, self.repo.store, d)

    def test_every_out_flag_defers_to_the_domain_layer(self):
        """Both layers agree by construction, not by coincidence: no script
        spells the default out for itself."""
        for layer, names in (("issue", ("issue_new.py", "issue_check.py",
                                        "issue_tree.py", "issue_index.py",
                                        "issue_evict.py")),
                             ("sync", ("pull.py", "push.py", "remote.py",
                                       "comment.py", "evict.py"))):
            for name in names:
                with open(os.path.join(REPO, "skills", layer, "scripts", name)) as f:
                    src = f.read()
                self.assertIn('"--out", default=issue.ISSUE_ROOT', src,
                              "%s/%s does not take its --out default from the "
                              "domain layer" % (layer, name))


# --------------------------------------------------------------------------
# the layering rule, mechanically
# --------------------------------------------------------------------------

class TestLayering(unittest.TestCase):

    def test_domain_layer_is_stdlib_only(self):
        """skills/issue must keep working with skills/sync deleted — so no
        transport, and above all no subprocess, in the domain layer."""
        imported = set()
        for name in sorted(os.listdir(ISSUE_SCRIPTS)):
            if not name.endswith(".py"):
                continue
            with open(os.path.join(ISSUE_SCRIPTS, name)) as f:
                for line in f:
                    if line.startswith(("import ", "from ")):
                        imported.add(line.split()[1].split(".")[0])
        local = {"issue", "issue_ac", "issue_index"}
        foreign = imported - local - sys.stdlib_module_names
        self.assertEqual(foreign, set(),
                         "non-stdlib import in the domain layer: %s"
                         % ", ".join(sorted(foreign)))
        self.assertNotIn("subprocess", imported)


if __name__ == "__main__":
    unittest.main()