weichen
0d1840da96
Add human-readable --format table, one-page brief, claim archive via docs apply CAS, and four MCP readonly tools. Includes P1 console design doc and engineering-tool test evidence.
160 lines
6.4 KiB
Python
160 lines
6.4 KiB
Python
#!/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()
|