126 lines
5.6 KiB
Python
126 lines
5.6 KiB
Python
import unittest
|
|
|
|
from tracker import CompletedCommand, RecordingRunner, RetryPolicy, Tracker # type: ignore[reportMissingImports]
|
|
|
|
|
|
class TrackerOperationTests(unittest.TestCase):
|
|
def test_add_label_ensures_missing_label_before_assignment(self):
|
|
runner = RecordingRunner(
|
|
[
|
|
CompletedCommand("[]"),
|
|
CompletedCommand('{"name":"ready","color":"ededed"}'),
|
|
CompletedCommand('{"number":7,"labels":[{"name":"ready"}]}'),
|
|
]
|
|
)
|
|
result = Tracker(provider="github", runner=runner).add_label("issue", 7, "ready")
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual([call[0][:3] for call in runner.calls], [["gh", "label", "list"], ["gh", "label", "create"], ["gh", "issue", "edit"]])
|
|
self.assertEqual(result["details"]["ensured"]["created"], True)
|
|
|
|
def test_existing_label_is_not_recreated(self):
|
|
runner = RecordingRunner(
|
|
[
|
|
CompletedCommand('[{"name":"ready","color":"ff0000"}]'),
|
|
CompletedCommand('{"number":7,"labels":[{"name":"ready"}]}'),
|
|
]
|
|
)
|
|
result = Tracker(provider="gitea", runner=runner).add_label("issue", 7, "ready")
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(len(runner.calls), 2)
|
|
self.assertEqual(result["details"]["ensured"]["created"], False)
|
|
|
|
def test_transient_failure_is_retried_and_normalized(self):
|
|
runner = RecordingRunner(
|
|
[
|
|
CompletedCommand("", "connection reset", 1),
|
|
CompletedCommand('{"number":4,"iid":4,"title":"Fix","description":"body","state":"opened","labels":[]}'),
|
|
]
|
|
)
|
|
result = Tracker(provider="gitlab", runner=runner, retry=RetryPolicy(attempts=2)).get_issue(4)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(len(runner.calls), 2)
|
|
self.assertEqual(result["result"]["number"], 4)
|
|
self.assertEqual(result["result"]["body"], "body")
|
|
|
|
def test_external_pr_filter_normalizes_github_associations(self):
|
|
runner = RecordingRunner([
|
|
CompletedCommand('[{"number":1,"title":"inside","authorAssociation":"MEMBER"},{"number":2,"title":"outside","authorAssociation":"NONE"}]')
|
|
])
|
|
result = Tracker(provider="github", runner=runner).list_external_prs()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual([item["number"] for item in result["result"]], [2])
|
|
|
|
def test_external_filter_reports_missing_membership_metadata(self):
|
|
runner = RecordingRunner([CompletedCommand('[{"number":1,"title":"unknown"}]')])
|
|
result = Tracker(provider="gitea", runner=runner).list_external_prs()
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(result["error"]["code"], "unsupported_capability")
|
|
|
|
def test_pr_get_can_include_diff(self):
|
|
runner = RecordingRunner([
|
|
CompletedCommand('{"number":3,"title":"Change","state":"open"}'),
|
|
CompletedCommand("diff --git a/a b/a"),
|
|
])
|
|
result = Tracker(provider="gitlab", runner=runner).get_pr(3, diff=True)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["result"]["diff"], "diff --git a/a b/a")
|
|
|
|
def test_child_creation_links_native_and_updates_map_order(self):
|
|
runner = RecordingRunner([
|
|
CompletedCommand("[]"),
|
|
CompletedCommand("{}"),
|
|
CompletedCommand('{"number":7,"title":"Research"}'),
|
|
CompletedCommand("{}"),
|
|
CompletedCommand('{"number":9,"body":"Notes"}'),
|
|
CompletedCommand("{}"),
|
|
])
|
|
result = Tracker(provider="github", runner=runner).create_child(9, "Research", wayfinder_type="research")
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["result"]["number"], 7)
|
|
self.assertEqual(result["details"]["relationship"], "native")
|
|
self.assertTrue(result["details"]["map_updated"])
|
|
self.assertTrue(any("sub_issues" in part for part in runner.calls[3][0]))
|
|
|
|
def test_resolve_reports_completed_steps_after_partial_failure(self):
|
|
runner = RecordingRunner([
|
|
CompletedCommand("{}"),
|
|
CompletedCommand("", "connection reset", 1),
|
|
])
|
|
result = Tracker(provider="gitea", runner=runner, retry=RetryPolicy(attempts=1)).resolve("issue", 7, "answer")
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(result["error"]["code"], "partial_failure")
|
|
self.assertEqual(result["error"]["details"]["completed"], ["comment"])
|
|
|
|
def test_frontier_uses_map_task_order_and_filters_closed_or_claimed(self):
|
|
runner = RecordingRunner([
|
|
CompletedCommand('{"number":9,"body":"- [ ] #2 second\\n- [ ] #1 first","state":"open"}'),
|
|
CompletedCommand('{"number":2,"title":"second","state":"open","assignees":[]}'),
|
|
CompletedCommand('{"number":1,"title":"first","state":"closed","assignees":[]}'),
|
|
])
|
|
result = Tracker(provider="github", runner=runner).frontier(9)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual([item["number"] for item in result["result"]], [2])
|
|
self.assertTrue(result["details"]["deterministic"])
|
|
|
|
def test_authentication_failure_is_not_retryable(self):
|
|
runner = RecordingRunner([CompletedCommand("", "not logged in", 1)])
|
|
result = Tracker(provider="github", runner=runner).get_issue(4)
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertEqual(result["error"]["code"], "auth_required")
|
|
self.assertFalse(result["error"]["retryable"])
|
|
self.assertEqual(len(runner.calls), 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|