feat(bridge): add assign:agent and agent:agent label direct routing

2026-10-09 21:52:58 +00:00
parent 69555f809c
commit 698798df2a
2 changed files with 300 additions and 0 deletions
@@ -0,0 +1,94 @@
+import os
+import sys
+import tempfile
+import unittest
+import json
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.join(REPO_ROOT, "bin"))
+
+import importlib.util
+spec = importlib.util.spec_from_file_location("box_gitea_bridge", os.path.join(REPO_ROOT, "bin", "box-gitea-bridge.py"))
+bgb = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(bgb)
+
+class TestBoxGiteaBridge(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        bgb.TASKS_DIR = os.path.join(self.tmpdir.name, "fleet", "tasks")
+        os.makedirs(os.path.join(bgb.TASKS_DIR, "pending"), exist_ok=True)
+        os.makedirs(os.path.join(bgb.TASKS_DIR, "claimed"), exist_ok=True)
+        os.makedirs(os.path.join(bgb.TASKS_DIR, "done"), exist_ok=True)
+
+    def tearDown(self):
+        self.tmpdir.cleanup()
+
+    def test_slugify(self):
+        self.assertEqual(bgb.slugify("Hello World! 123"), "hello-world-123")
+        self.assertEqual(bgb.slugify("Fix: Gitea & Box Integration"), "fix-gitea-box-integration")
+
+    def test_label_gating(self):
+        # Unlabeled issue -> skipped
+        issue_unlabeled = {"number": 208, "title": "Untagged discussion", "labels": []}
+        path, status = bgb.create_task_from_issue(issue_unlabeled)
+        self.assertIsNone(path)
+        self.assertEqual(status, "skipped_label_gate")
+
+        # Irrelevant label -> skipped
+        issue_wontfix = {"number": 208, "title": "Wontfix bug", "labels": [{"name": "wontfix"}]}
+        path, status = bgb.create_task_from_issue(issue_wontfix)
+        self.assertIsNone(path)
+        self.assertEqual(status, "skipped_label_gate")
+
+        # Labeled 'task' -> created
+        issue_task = {"number": 208, "title": "Build Gitea Bridge", "labels": [{"name": "task"}], "body": "Implement bridge"}
+        path, status = bgb.create_task_from_issue(issue_task)
+        self.assertIsNotNone(path)
+        self.assertEqual(status, "created")
+        self.assertTrue(os.path.exists(path))
+        self.assertIn("208-build-gitea-bridge.md", path)
+
+    def test_assigned_issue_claims_directly(self):
+        issue_assigned = {
+            "number": 209,
+            "title": "OPM Recovery Task",
+            "labels": [{"name": "ready"}],
+            "body": "Run recovery",
+            "assignee": {"username": "opm"}
+        }
+        path, status = bgb.create_task_from_issue(issue_assigned)
+        self.assertIsNotNone(path)
+        self.assertEqual(status, "created")
+        self.assertIn("claimed", path)
+        self.assertTrue(path.endswith(".opm"))
+
+    def test_label_based_routing(self):
+        issue_label_assigned = {
+            "number": 211,
+            "title": "Direct Labeled Task",
+            "labels": [{"name": "task"}, {"name": "assign:opm"}],
+            "body": "Direct routing via label"
+        }
+        path, status = bgb.create_task_from_issue(issue_label_assigned)
+        self.assertIsNotNone(path)
+        self.assertEqual(status, "created")
+        self.assertIn("claimed", path)
+        self.assertTrue(path.endswith(".opm"))
+
+    def test_close_task_moves_to_done(self):
+        issue = {"number": 210, "title": "Close test", "labels": [{"name": "ready"}]}
+        created_path, _ = bgb.create_task_from_issue(issue)
+        self.assertTrue(os.path.exists(created_path))
+
+        done_path = bgb.close_task_for_issue(210, "Verified fixed")
+        self.assertIsNotNone(done_path)
+        self.assertFalse(os.path.exists(created_path))
+        self.assertTrue(os.path.exists(done_path))
+        self.assertIn("done", done_path)
+
+        with open(done_path) as f:
+            content = f.read()
+        self.assertIn("Verified fixed", content)
+
+if __name__ == "__main__":
+    unittest.main()