#!/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)