#!/usr/bin/env python3 """Unit tests for agentctl visibility helpers (table/brief/claim-archive).""" import datetime as dt import pathlib import sys import unittest SCRIPTS = pathlib.Path(__file__).resolve().parents[1] sys.path.insert(0, str(SCRIPTS)) sys.dont_write_bytecode = True import agentctl_visibility as vis # noqa: E402 SAMPLE_CURRENT = """# 当前焦点 ## 焦点 1. something ## 并行认领 | owner | task | branch | worktree | mutex | status | |---|---|---|---|---|---| | Cursor | keep-active-2026-09-24 | feature/keep | .worktrees/keep | m1 | in_progress | | Cursor | recent-done-2026-09-23 | feature/recent | .worktrees/recent | m2 | done | | Codex | old-done-2026-09-10 | feature/old | .worktrees/old | m3 | done | | Cursor | undated-done | feature/undated | .worktrees/u | m4 | done | ## 集成槽位 | holder | phase | |---|---| | - | idle | """ class FormatTableTests(unittest.TestCase): def test_format_table_aligns_columns(self): text = vis.format_table(['a', 'bb'], [['1', '22'], ['333', '4']]) lines = text.splitlines() self.assertEqual(len(lines), 4) self.assertIn('a', lines[0]) self.assertTrue(lines[1].startswith('-')) def test_render_status_table_active_only(self): payload = { 'main': '/repo', 'epoch': 'abc-epoch', 'version': 1, 'tasks': [ {'id': 't1', 'owner': 'A', 'state': 'active', 'worktree': '/repo/.worktrees/t1', 'branch': 'feature/t1'}, {'id': 't2', 'owner': 'B', 'state': 'done', 'worktree': '/repo/.worktrees/t2', 'branch': 'feature/t2'}, ], 'runs': [], 'queue': [{'id': 'q1', 'task': 't1', 'state': 'pending', 'priority': 1, 'head_sha': 'deadbeef', 'created': 0}], 'policy': {'loaded': 'ok', 'policy_revision': 'r1'}, } text = vis.render_status_table(payload, now=100.0) self.assertIn('active=1', text) self.assertIn('t1', text) self.assertNotIn('t2', text.split('## runs')[0]) # closed not in active table self.assertIn('queue depth=1', text) class BriefTests(unittest.TestCase): def test_build_brief_markdown(self): text = vis.build_brief( status={ 'tasks': [{'id': 't1', 'owner': 'A', 'state': 'active', 'worktree': '/x/.worktrees/t1'}], 'runs': [], 'queue': [], }, docs_drift={'regressions': []}, cleanup={'worktrees': [{'path': 'a', 'cleanup_allowed': False}]}, catalog_drift={'regressions': []}, daemon={'status': 'ok', 'manager_lock': {'held': True}, 'queue': {'pending': 0}}, now=1_700_000_000, ) self.assertTrue(text.startswith('# agentctl brief')) self.assertIn('daemon: **ok**', text) self.assertIn('`t1`', text) self.assertIn('cleanup backlog: 1', text) class ClaimArchiveTests(unittest.TestCase): def test_parse_current_claim_rows(self): rows = vis.parse_current_claim_rows(SAMPLE_CURRENT) self.assertEqual(len(rows), 4) self.assertEqual(rows[0]['task'], 'keep-active-2026-09-24') self.assertEqual(rows[2]['status'], 'done') def test_parse_task_date(self): self.assertEqual(vis.parse_task_date('foo-2026-09-10'), dt.date(2026, 9, 10)) self.assertEqual(vis.parse_task_date('2026-09-10-bar'), dt.date(2026, 9, 10)) self.assertIsNone(vis.parse_task_date('no-date-here')) def test_select_archive_candidates_respects_retain_window(self): rows = vis.parse_current_claim_rows(SAMPLE_CURRENT) selected = vis.select_archive_candidates( rows, retain_days=7, today=dt.date(2026, 9, 24), include_undated=False) eligible = [r for r in selected if r['eligible']] self.assertEqual([r['task'] for r in eligible], ['old-done-2026-09-10']) skipped_reasons = {r['task']: r['reason'] for r in selected if not r['eligible']} self.assertEqual(skipped_reasons['recent-done-2026-09-23'], 'within_retain_window') self.assertEqual(skipped_reasons['undated-done'], 'undated_skipped') def test_select_include_undated(self): rows = vis.parse_current_claim_rows(SAMPLE_CURRENT) selected = vis.select_archive_candidates( rows, retain_days=0, today=dt.date(2026, 9, 24), include_undated=True) eligible_ids = {r['task'] for r in selected if r['eligible']} self.assertIn('undated-done', eligible_ids) self.assertIn('old-done-2026-09-10', eligible_ids) self.assertIn('recent-done-2026-09-23', eligible_ids) self.assertNotIn('keep-active-2026-09-24', eligible_ids) def test_build_archived_current_removes_rows(self): out = vis.build_archived_current(SAMPLE_CURRENT, ['old-done-2026-09-10', 'undated-done']) self.assertNotIn('old-done-2026-09-10', out) self.assertNotIn('undated-done', out) self.assertIn('keep-active-2026-09-24', out) self.assertIn('## 集成槽位', out) self.assertIn('## 焦点', out) def test_build_archive_section(self): rows = [{'owner': 'A', 'task': 't', 'branch': 'b', 'worktree': 'w', 'mutex': 'm', 'status': 'done'}] section = vis.build_archive_section(rows, '2026-09-24') self.assertIn('## 并行认领归档 2026-09-24', section) self.assertIn('| A | t | b | w | m | done |', section) class QueueClaimCleanupRenderTests(unittest.TestCase): def test_render_queue_and_claims(self): q = vis.render_queue_table({'items': [ {'id': '1', 'task': 't', 'state': 'pending', 'priority': 0, 'head_sha': 'abc', 'expected_dev': 'def', 'created': 0}, ]}, now=10) self.assertIn('queue list count=1', q) c = vis.render_claim_table({'claims': [ {'id': 'c1', 'task': 't', 'resource': 'claim:x', 'state': 'held', 'updated': 0}, ]}, now=10) self.assertIn('claim list count=1', c) def test_render_cleanup(self): text = vis.render_cleanup_table({'worktrees': [ {'path': '/repo/.worktrees/a', 'cleanup_allowed': True, 'cleanup_owner': 'X'}, {'path': '/repo/.worktrees/b', 'status': 'blocked', 'gates': [ {'name': 'dev_ancestor', 'status': 'fail'}, ]}, ]}) self.assertIn('allowed=1', text) self.assertIn('blocked/pending=1', text) self.assertIn('dev_ancestor', text) if __name__ == '__main__': unittest.main()