agent-orchestrator/tests/test_tools.py

#!/usr/bin/env python3
"""Unit tests for the standalone tools (the tangled_* scripts, secrets, gateway-domain).

Run:  python3 -m unittest tests.test_tools     (from the repo root)
      ./tests/run.sh                            (discovers this file automatically)

NO NETWORK. Every HTTP boundary is injected as a fake `_fetch`, so these run offline and
in CI. The point is the DECISION logic — "did the pull get created", "is this backend
valid" — because that is where the defects have actually been.

The reason this file exists: tangled_pr.py judged success only by a response header, so a
create that answered without one read as a failure, the caller retried, and duplicate
pulls #397-#399 were filed for one branch. The tool was never tested; the bug survived
months of daily use and was found by counting pulls, not by running the script.
"""
import os, sys, unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import tangled_pr  # noqa: E402


class TestPullNumbers(unittest.TestCase):
    def test_extracts_every_linked_pull(self):
        html = '<a href="/o/r/pulls/417">x</a><a href="/o/r/pulls/416">y</a>'
        self.assertEqual(tangled_pr.pull_numbers(html), {417, 416})

    def test_deduplicates_repeated_links(self):
        html = '/pulls/12 /pulls/12 /pulls/12'
        self.assertEqual(tangled_pr.pull_numbers(html), {12})

    def test_ignores_the_new_pull_form(self):
        # /pulls/new must not parse as a number — it is the page you POST to, not a pull
        self.assertEqual(tangled_pr.pull_numbers('/pulls/new /pulls/9'), {9})

    def test_empty_and_none_are_empty_not_an_error(self):
        # an unreadable page must yield "cannot confirm", never a crash mid-create
        self.assertEqual(tangled_pr.pull_numbers(""), set())
        self.assertEqual(tangled_pr.pull_numbers(None), set())


class TestPageNamesBranch(unittest.TestCase):
    def test_finds_the_branch(self):
        self.assertTrue(tangled_pr.page_names_branch("<div>e2e-pds/R4-real</div>", "e2e-pds/R4-real"))

    def test_absent_branch(self):
        self.assertFalse(tangled_pr.page_names_branch("<div>other</div>", "e2e-pds/R4-real"))

    def test_unreadable_page_is_not_a_match(self):
        self.assertFalse(tangled_pr.page_names_branch("", "b"))
        self.assertFalse(tangled_pr.page_names_branch(None, "b"))

    def test_empty_branch_never_matches(self):
        # guards against a caller passing "" and matching every page
        self.assertFalse(tangled_pr.page_names_branch("anything", ""))


class TestClassify(unittest.TestCase):
    """The regression that matters: a successful create with NO redirect header."""

    def test_new_pull_naming_our_branch_is_created(self):
        v, n = tangled_pr.classify({1, 2}, {1, 2, 3}, {3: True})
        self.assertEqual((v, n), ("created", 3))

    def test_created_even_though_the_header_said_nothing(self):
        # THE BUG: classify never sees the response at all, so a missing HX-Redirect
        # cannot turn a real create into a reported failure.
        v, n = tangled_pr.classify({396}, {396, 397}, {397: True})
        self.assertEqual(v, "created")

    def test_no_new_pull_is_a_real_failure(self):
        v, n = tangled_pr.classify({1, 2}, {1, 2}, {})
        self.assertEqual((v, n), ("none", None))

    def test_someone_elses_pull_is_not_ours(self):
        v, n = tangled_pr.classify({1}, {1, 2}, {2: False})
        self.assertEqual((v, n), ("unrelated", None))

    def test_picks_the_highest_of_several_of_ours(self):
        v, n = tangled_pr.classify({1}, {1, 2, 3}, {2: True, 3: True})
        self.assertEqual((v, n), ("created", 3))

    def test_a_pull_that_vanished_does_not_confuse_it(self):
        # after ⊂ before (someone closed+deleted one mid-run): no new pull, so no create
        v, n = tangled_pr.classify({1, 2}, {1}, {})
        self.assertEqual((v, n), ("none", None))


class TestExistingPullScan(unittest.TestCase):
    """The pre-check that makes a retry harmless instead of duplicating."""

    def _fetcher(self, pages):
        def _f(url, cookie, timeout=30):
            return pages.get(url.rsplit("/", 1)[-1], "")
        return _f

    def test_finds_an_existing_pull_for_the_branch(self):
        f = self._fetcher({"396": "branch: feat/x", "395": "branch: other"})
        got = tangled_pr.existing_pull_for("o", "r", "feat/x", "c", {395, 396}, _fetch=f)
        self.assertEqual(got, 396)

    def test_returns_none_when_the_branch_is_new(self):
        f = self._fetcher({"396": "other", "395": "other"})
        self.assertIsNone(tangled_pr.existing_pull_for("o", "r", "feat/x", "c", {395, 396}, _fetch=f))

    def test_prefers_the_newest_match(self):
        f = self._fetcher({"10": "feat/x", "20": "feat/x"})
        self.assertEqual(tangled_pr.existing_pull_for("o", "r", "feat/x", "c", {10, 20}, _fetch=f), 20)

    def test_scan_window_is_bounded(self):
        # a match older than the window is missed BY DESIGN; the post-check is the backstop.
        pages = {str(n): ("feat/x" if n == 1 else "other") for n in range(1, 30)}
        f = self._fetcher(pages)
        self.assertIsNone(
            tangled_pr.existing_pull_for("o", "r", "feat/x", "c", set(range(1, 30)), scan=5, _fetch=f))

    def test_unreadable_pages_do_not_claim_absence(self):
        # every fetch fails -> None, and main() warns rather than silently creating a duplicate
        f = lambda url, cookie, timeout=30: ""
        self.assertIsNone(tangled_pr.existing_pull_for("o", "r", "feat/x", "c", {1, 2}, _fetch=f))


class TestFetchIsFailSoft(unittest.TestCase):
    def test_network_error_returns_empty_not_raise(self):
        # a create must not die between POST and verification
        self.assertEqual(tangled_pr.fetch("http://127.0.0.1:1/nope", "c", timeout=1), "")


class TestCloseToolStateParsing(unittest.TestCase):
    """tangled_pr_close.state_of decides whether a pull is already closed."""

    def setUp(self):
        import tangled_pr_close
        self.mod = tangled_pr_close

    def test_badge_regex_reads_the_states(self):
        import re
        badges = lambda doc: set(re.findall(r'>\s*(Merged|Closed|Open)\s*<', doc))
        self.assertEqual(badges("<span> Closed </span>"), {"Closed"})
        self.assertEqual(badges("<b>Open</b><b>Merged</b>"), {"Open", "Merged"})
        self.assertEqual(badges("<p>closed</p>"), set())  # case-sensitive on purpose


class TestGatewayBackendValidation(unittest.TestCase):
    """gateway-domain refuses hostname backends: the gateway can store one and never remove it."""

    def setUp(self):
        import importlib.util
        p = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
                         "tools", "gateway-domain.py")
        if not os.path.exists(p):
            self.skipTest("gateway-domain.py not present")
        spec = importlib.util.spec_from_file_location("gwd", p)
        self.gwd = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(self.gwd)

    def test_accepts_ip_and_ip_port(self):
        self.assertTrue(self.gwd._BACKEND.match("100.74.89.63"))
        self.assertTrue(self.gwd._BACKEND.match("100.74.89.63:8443"))

    def test_rejects_hostnames(self):
        for bad in ("example.com", "host:443", "", "1.2.3", "1.2.3.4.5"):
            self.assertIsNone(self.gwd._BACKEND.match(bad), f"should reject {bad!r}")

    def test_allowed_tags_include_both_known_tags(self):
        self.assertIn("tag:notplants-test-server", self.gwd.ALLOWED_TAGS)


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