weblib-ci/test_sync_blocked_label.py

#!/usr/bin/env python3
"""Tests for `sync_blocked_label.py`.

Standard library only, like everything else here, and offline: the Gitea it
talks to is a `http.server` on localhost that keeps whatever state a test
hands it, so assertions read that state back rather than trusting an exit
code.

Run them with any python3:

    python3 test_sync_blocked_label.py

The point of most of them is the three-way split in `reconcile()` --
positively open, positively closed, or unresolvable -- because collapsing it
back to two is exactly how weblib-ci#11 happened. A cross-repo dependency the
Actions token cannot read comes back from Gitea as a zero-value stub
(`id: 0`, `state: ""`, `repository: null`, ...) rather than being omitted or
erroring, and the old code's `d.get("state") != "closed"` read that as "still
open" forever -- which first meant the label could never come off, and then
(weblib-archive#30, 2026-09-09) meant the reconciler *reasserted* the label
every 15 minutes over a human's hand-removal.
"""

import contextlib
import io
import json
import os
import threading
import unittest
from http.server import BaseHTTPRequestHandler, HTTPServer

import sync_blocked_label as sbl

HERE = os.path.dirname(os.path.abspath(__file__))


def resolved(number, state, repo="weblib/cfbypass"):
    """A dependency entry the token *could* read -- the shape a same-repo
    blocker, or a cross-repo one under a token that can see it, comes back
    as."""
    owner, name = repo.split("/")
    return {
        "id": 1000 + number, "number": number, "state": state,
        "repository": {"owner": owner, "name": name, "full_name": repo},
    }


def unresolvable():
    """The zero-value stub Gitea substitutes for an entry the token cannot
    read. Measured live against weblib-archive#68 and #30, 2026-09-09 --
    every field blank or zero, `state` included."""
    return {
        "id": 0, "url": "", "html_url": "", "number": 0, "user": None,
        "state": "", "repository": None, "labels": None, "milestone": None,
    }


class FakeGitea(BaseHTTPRequestHandler):
    """Just enough of the API to drive `reconcile()`.

    Class-level state, reset per test in `setUp`:
      - `labels`: {"repo"/"org": [label dicts]}
      - `issues`: [issue dicts] -- the open items in the one repo under test
      - `deps`: {issue number: [dependency entries]}
      - `label_calls`: every POST/PUT to a labels endpoint, in order
    """

    labels_repo = []
    labels_org = []
    issues = []
    deps = {}
    label_calls = []

    def _send(self, code, payload):
        raw = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

    def do_GET(self):
        cls = type(self)
        if self.path.startswith("/api/v1/repos/") and self.path.endswith(
                "/labels?limit=100"):
            return self._send(200, cls.labels_repo)
        if self.path.startswith("/api/v1/orgs/") and "/labels" in self.path:
            return self._send(200, cls.labels_org)
        if "/issues?state=open" in self.path:
            page = int(self.path.rsplit("page=", 1)[1])
            return self._send(200, cls.issues if page == 1 else [])
        if "/dependencies" in self.path:
            number = int(self.path.rsplit("/issues/", 1)[1].split("/")[0])
            return self._send(200, cls.deps.get(number, []))
        self._send(404, {"message": "no route: " + self.path})

    def _record_and_ack(self, method):
        length = int(self.headers.get("Content-Length") or 0)
        payload = json.loads(self.rfile.read(length) or b"{}")
        type(self).label_calls.append({
            "method": method, "path": self.path, "labels": payload["labels"],
        })
        self._send(200, [])

    def do_POST(self):
        self._record_and_ack("POST")

    def do_PUT(self):
        self._record_and_ack("PUT")

    def log_message(self, *a):          # keep the test output readable
        pass


BLOCKED_LABEL = {"id": 145, "name": "Status/Blocked"}
ON_HOLD_LABEL = {"id": 152, "name": "Status/On Hold"}


def issue(number, *label_dicts):
    return {"number": number, "labels": list(label_dicts)}


class ReconcileTestCase(unittest.TestCase):

    def setUp(self):
        FakeGitea.labels_repo = []
        FakeGitea.labels_org = [BLOCKED_LABEL]
        FakeGitea.issues = []
        FakeGitea.deps = {}
        FakeGitea.label_calls = []
        self.server = HTTPServer(("127.0.0.1", 0), FakeGitea)
        self.host = "http://127.0.0.1:%d" % self.server.server_address[1]
        self.thread = threading.Thread(
            target=self.server.serve_forever, kwargs={"poll_interval": 0.01},
            daemon=True)
        self.thread.start()
        self.addCleanup(self.server.server_close)
        self.addCleanup(self.server.shutdown)
        self.forge = sbl.Forge(self.host, "t0ken")

    def reconcile(self, dry_run=False):
        out = io.StringIO()
        with contextlib.redirect_stdout(out):
            changed = sbl.reconcile(self.forge, "weblib/weblib-archive",
                                    "Status/Blocked", dry_run)
        return changed, out.getvalue()

    def calls_for(self, number):
        needle = f"/issues/{number}/labels"
        return [c for c in FakeGitea.label_calls if c["path"].endswith(needle)]


class TestNoDependencyEdges(ReconcileTestCase):

    def test_untouched(self):
        """An issue with no dependency edges is never touched -- it may be
        blocked by hand for a reason the graph does not know about
        (weblib-archive#29/#30's original 'blocked on a decision' case)."""
        FakeGitea.issues = [issue(1, BLOCKED_LABEL)]
        FakeGitea.deps = {1: []}
        changed, _ = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(self.calls_for(1), [])


class TestSameRepoUnchanged(ReconcileTestCase):
    """Pins the pre-existing, already-correct behaviour so a fix aimed at the
    cross-repo case cannot quietly change the path that already worked --
    same-repo dependencies always resolve fully (no id:0 stub ever shows up
    for them), so `unresolved` is always empty here regardless of the fix."""

    def test_open_blocker_adds_the_label(self):
        FakeGitea.issues = [issue(5)]
        FakeGitea.deps = {5: [resolved(1, "open", "weblib/weblib-archive")]}
        changed, log = self.reconcile()
        self.assertEqual(changed, [(5, "add", "blocked by #1")])
        self.assertEqual(self.calls_for(5),
                         [{"method": "POST", "path": "/api/v1/repos/weblib/weblib-archive/issues/5/labels",
                           "labels": [145]}])
        self.assertIn("add    Status/Blocked on weblib/weblib-archive#5", log)

    def test_closed_blocker_removes_the_label(self):
        FakeGitea.issues = [issue(6, BLOCKED_LABEL)]
        FakeGitea.deps = {6: [resolved(2, "closed", "weblib/weblib-archive")]}
        changed, _ = self.reconcile()
        self.assertEqual(changed, [(6, "remove", "every blocker is closed")])
        self.assertEqual(self.calls_for(6),
                         [{"method": "PUT", "path": "/api/v1/repos/weblib/weblib-archive/issues/6/labels",
                           "labels": []}])

    def test_already_correct_is_a_no_op_either_direction(self):
        FakeGitea.issues = [
            issue(7, BLOCKED_LABEL),  # open blocker, already labelled
            issue(8),                 # closed blocker, already unlabelled
        ]
        FakeGitea.deps = {
            7: [resolved(1, "open", "weblib/weblib-archive")],
            8: [resolved(1, "closed", "weblib/weblib-archive")],
        }
        changed, _ = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(FakeGitea.label_calls, [])


class TestUnresolvableCrossRepoBlocker(ReconcileTestCase):
    """The fix. Reproduces weblib-archive#68/#30 exactly: a dependency entry
    that comes back as Gitea's zero-value stub because the Actions token
    cannot read the other repo."""

    def test_never_removes_a_label_it_cannot_confirm_is_clear(self):
        """The original bug: every blocker closed, cross-repo, label stuck
        forever."""
        FakeGitea.issues = [issue(30, BLOCKED_LABEL)]
        FakeGitea.deps = {30: [unresolvable(), unresolvable()]}
        changed, log = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(self.calls_for(30), [])
        self.assertIn("unknown weblib/weblib-archive#30", log)
        self.assertIn("2 blocker(s) unresolvable", log)
        self.assertIn("present", log)

    def test_never_reasserts_a_label_a_human_removed(self):
        """The sharper bug found live on 2026-09-09: same dependency graph
        as above, but a human had already taken the label off. The old code
        put it straight back on the next run; this must leave it off."""
        FakeGitea.issues = [issue(30)]                    # label removed by hand
        FakeGitea.deps = {30: [unresolvable(), unresolvable()]}
        changed, log = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(self.calls_for(30), [])
        self.assertIn("absent", log)

    def test_a_known_open_blocker_still_wins_over_an_unresolvable_one(self):
        """Positive evidence of an open blocker is decisive on its own --
        this is not a reason to *loosen* the guard, only to stop guessing
        where there is no evidence either way."""
        FakeGitea.issues = [issue(9)]
        FakeGitea.deps = {9: [resolved(1, "open", "weblib/weblib-archive"),
                              unresolvable()]}
        changed, _ = self.reconcile()
        self.assertEqual(changed, [(9, "add", "blocked by #1")])

    def test_an_unresolvable_blocker_blocks_removal_even_with_other_closed(self):
        """One confirmed-closed same-repo blocker plus one unresolvable
        cross-repo one must not remove the label -- the unresolvable one
        might still be open."""
        FakeGitea.issues = [issue(10, BLOCKED_LABEL)]
        FakeGitea.deps = {10: [resolved(1, "closed", "weblib/weblib-archive"),
                               unresolvable()]}
        changed, log = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(self.calls_for(10), [])
        self.assertIn("unknown weblib/weblib-archive#10", log)

    def test_is_resolved_keys_on_id_not_on_state(self):
        """bit, 2026-09-09: key on `id == 0`, not on the empty `state` --
        `state` is a symptom that happens to be visible today; a future
        Gitea could populate it while still stubbing the rest of the
        entry."""
        self.assertFalse(sbl.is_resolved(unresolvable()))
        self.assertFalse(sbl.is_resolved({"id": 0, "state": "closed"}))
        self.assertTrue(sbl.is_resolved({"id": 7, "state": ""}))


class TestHandsOff(ReconcileTestCase):

    def test_on_hold_is_never_touched_even_with_open_deps(self):
        FakeGitea.issues = [issue(11, ON_HOLD_LABEL)]
        FakeGitea.deps = {11: [resolved(1, "open", "weblib/weblib-archive")]}
        changed, log = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(self.calls_for(11), [])
        self.assertIn("hands off weblib/weblib-archive#11", log)


class TestNoLabelInRepo(ReconcileTestCase):

    def test_skips_everything_and_calls_nothing(self):
        FakeGitea.labels_repo = []
        FakeGitea.labels_org = []
        FakeGitea.issues = [issue(1)]
        changed, log = self.reconcile()
        self.assertEqual(changed, [])
        self.assertEqual(FakeGitea.label_calls, [])
        self.assertIn("skipped: no 'Status/Blocked' label in this repo", log)


class TestDryRun(ReconcileTestCase):

    def test_dry_run_reports_but_does_not_call(self):
        FakeGitea.issues = [issue(5)]
        FakeGitea.deps = {5: [resolved(1, "open", "weblib/weblib-archive")]}
        changed, log = self.reconcile(dry_run=True)
        self.assertEqual(changed, [(5, "add", "blocked by #1")])
        self.assertEqual(FakeGitea.label_calls, [])
        self.assertIn("would add", log)


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