Codex
89 lines
4.0 KiB
Python
89 lines
4.0 KiB
Python
import unittest
|
|
|
|
from src.services.operator.workflow_archive_projection import build_workflow_archive_operations_view
|
|
|
|
|
|
class WorkflowArchiveProjectionTests(unittest.TestCase):
|
|
def test_builds_daily_session_and_drift_views(self) -> None:
|
|
items = [
|
|
{
|
|
"snapshot_id": 3,
|
|
"created_ts": "2026-05-14T09:30:00+00:00",
|
|
"current_session": "LONDON",
|
|
"next_phase": "intraday",
|
|
"next_artifact_kind": "intraday_digest",
|
|
"artifact_count": 4,
|
|
"ready_for_handoff_count": 2,
|
|
"needs_update_count": 1,
|
|
"pending_capture_count": 1,
|
|
"workflow_body": "London plan ready.",
|
|
"next_artifact_prompt": "Check intraday follow-through.",
|
|
"phase_briefs": [
|
|
{"phase": "pre_market", "archive_body": "Macro and session prep done."},
|
|
{"phase": "intraday", "archive_body": "Watch the first liquidity sweep."},
|
|
],
|
|
},
|
|
{
|
|
"snapshot_id": 2,
|
|
"created_ts": "2026-05-13T15:45:00+00:00",
|
|
"current_session": "NY_AM",
|
|
"next_phase": "post_trade",
|
|
"next_artifact_kind": "post_trade_recap",
|
|
"artifact_count": 2,
|
|
"ready_for_handoff_count": 1,
|
|
"needs_update_count": 1,
|
|
"pending_capture_count": 0,
|
|
"workflow_body": "NY recap pending.",
|
|
"next_artifact_prompt": "Write post-trade recap.",
|
|
"phase_briefs": [
|
|
{"phase": "pre_market", "archive_body": "Prep was lighter."},
|
|
{"phase": "intraday", "archive_body": "NY sweep already happened."},
|
|
],
|
|
},
|
|
{
|
|
"snapshot_id": 1,
|
|
"created_ts": "2026-05-13T12:15:00+00:00",
|
|
"current_session": "LONDON",
|
|
"next_phase": "intraday",
|
|
"next_artifact_kind": "intraday_digest",
|
|
"artifact_count": 3,
|
|
"ready_for_handoff_count": 1,
|
|
"needs_update_count": 1,
|
|
"pending_capture_count": 1,
|
|
"workflow_body": "Earlier London handoff.",
|
|
"next_artifact_prompt": "Re-check the London setup.",
|
|
"phase_briefs": [
|
|
{"phase": "pre_market", "archive_body": "Prep baseline."},
|
|
{"phase": "intraday", "archive_body": "Initial sweep watch."},
|
|
],
|
|
},
|
|
]
|
|
|
|
result = build_workflow_archive_operations_view(items)
|
|
|
|
self.assertEqual(result["daily_group_count"], 2)
|
|
self.assertEqual(result["session_group_count"], 2)
|
|
self.assertEqual(result["operations_summary"]["status"], "active")
|
|
self.assertEqual(result["operations_summary"]["latest_group_date"], "2026-05-14")
|
|
self.assertEqual(result["operations_summary"]["most_active_session"], "LONDON")
|
|
self.assertEqual(result["multi_day_drift_review"]["status"], "active")
|
|
self.assertEqual(result["multi_day_drift_review"]["drift_status"], "drift_detected")
|
|
self.assertEqual(result["bridge_summary"]["status"], "active")
|
|
self.assertEqual(result["bridge_summary"]["latest_snapshot_id"], 3)
|
|
self.assertGreaterEqual(result["multi_day_drift_review"]["changed_phase_count"], 1)
|
|
self.assertTrue(result["daily_groups"][0]["phase_names"])
|
|
self.assertEqual(result["session_groups"][0]["session_code"], "LONDON")
|
|
|
|
def test_handles_empty_archive(self) -> None:
|
|
result = build_workflow_archive_operations_view([])
|
|
|
|
self.assertEqual(result["daily_group_count"], 0)
|
|
self.assertEqual(result["session_group_count"], 0)
|
|
self.assertEqual(result["operations_summary"]["status"], "idle")
|
|
self.assertEqual(result["multi_day_drift_review"]["status"], "insufficient_history")
|
|
self.assertEqual(result["bridge_summary"]["status"], "idle")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|