import contextlib import io import json import tempfile import unittest from pathlib import Path from scripts import run_readonly_live_analysis_smoke class ReadonlyLiveAnalysisSmokeScriptTests(unittest.TestCase): def test_main_writes_passed_json_artifact_without_trade_operations(self) -> None: output = io.StringIO() with tempfile.TemporaryDirectory() as tmpdir: artifact = Path(tmpdir) / "readonly-smoke.json" opener = _FakeOpener( { "/health": {"app": "ai-ict", "database_configured": True}, "/trading/runtime-status": {"scheduler_state": "running", "current_blockers": []}, "/trading/launch-readiness?instrument_id=1&timeframe=1m": { "live_analysis_assistant_ready": True, "execution_assist_ready": False, "real_live_ready": False, "live_analysis_assistant_summary": { "ready_for_readonly_live_analysis": True, "must_not_trade": True, }, }, "/trading/review-packet?instrument_id=1": { "status": "blocked", "action_recommendation": "observe_only", "decision_assistant_summary": { "trader_mode_summary": { "status_label": "Observe Only", "primary_conclusion": "Observe only.", "focus_quality_explanation": "No single setup is clean enough.", "plain_language_reasons": ["No clean focus."], "recovery_actions": [{"label": "Continue Review"}], "keyboard_shortcuts": [{"key": "Space", "action": "pause_or_resume_auto_refresh"}], "profile_template": {"default_timeframe": "1m", "default_queue_instruments": [{"symbol": "BTC"}]}, "multi_instrument_queue_template": {"symbols": ["BTC", "ETH", "SOL"]}, "chart_context_bridge": {"status": "ready"}, "post_market_review_draft": {"status": "draft_ready"}, "ops_duty_panel": {"status": "use_today"}, "freshness_summary": {"status": "fresh"}, "r2_evidence_window": {"status": "collecting_evidence"}, "primary_question_trade_action": "not_a_buy_sell_instruction", "execution_allowed": False, "real_live_ready": False, } }, }, "/trading/shadow-paper-validation?instrument_id=1&timeframe=1m": { "status": "collecting_paper_samples", "execution_boundary": "real_live_still_blocked", }, "/trading/research/weekly-review?instrument_id=1": { "entry_count": 0, "next_week_actions": ["prepare_shadow_paper_weekly_review"], }, "/trading/maintenance-summary?instrument_id=1&alert_limit=20&visual_limit=20&incident_limit=20": { "status": "healthy", "blockers": [], }, "/trading/unattended-harness?instrument_id=1": {"status": "passed"}, "/candles?instrument_id=1&timeframe=1m&limit=3": {"items": []}, "/trading/backtests?instrument_id=1&limit=1": {"items": [], "latest": None}, "/trading/candidates?instrument_id=1&limit=5": {"items": []}, "/static/app.js": ( "Read-only Live Analysis No order submission Not a buy/sell instruction " "Execution remains disabled ready_for_readonly_live_analysis must_not_trade" ), "/trader-mode": ( "Trader Mode Now Wait ICT Fast Read Market Last Updated Review Packet Updated " "Fast Queue Chart Bridge Duty Panel Profile Sessions Setups Recovery No Trade " "Need More Evidence After-Close Draft R2 Evidence Window" ), "/static/trader_mode.js": ( "/trading/review-packet /trading/shadow-paper/observe /trading/operator-notes " "/trading/research/weekly-review /trading/maintenance-summary /trading/unattended-harness /candles /trading/backtests " "Do Not Use Review Now Observe Only TRADER_SIGNATURE_KEY TRADER_PROFILE_STORE_KEY ETH London SOL " "DEFAULT_QUEUE_INSTRUMENTS post_market_review_draft chart_context_bridge ops_duty_panel " "aria-keyshortcuts pause_or_resume_auto_refresh Historical Edge buildHistoricalEdgeSummary" ), } ) original_urlopen = run_readonly_live_analysis_smoke._urlopen_no_proxy run_readonly_live_analysis_smoke._urlopen_no_proxy = opener try: with contextlib.redirect_stdout(output): exit_code = run_readonly_live_analysis_smoke.main( argv=[ "--base-url", "http://example.test", "--output-json", "--artifact-path", str(artifact), ], env={}, ) finally: run_readonly_live_analysis_smoke._urlopen_no_proxy = original_urlopen artifact_payload = json.loads(artifact.read_text(encoding="utf-8")) payload = json.loads(output.getvalue()) self.assertEqual(exit_code, 0) self.assertEqual(payload["status"], "passed") self.assertTrue(payload["ready_for_readonly_live_analysis"]) self.assertTrue(payload["must_not_trade"]) self.assertFalse(payload["submit_cancel_or_live_execution_called"]) self.assertFalse(payload["trade_permission_required"]) self.assertEqual(artifact_payload["status"], "passed") def test_run_reports_blocker_when_launch_boundary_allows_execution(self) -> None: payload = run_readonly_live_analysis_smoke.run_readonly_live_analysis_smoke( base_url="http://example.test", attempts=1, sleep_seconds=0, opener=_FakeOpener( { "/health": {"app": "ai-ict"}, "/trading/runtime-status": {"scheduler_state": "running"}, "/trading/launch-readiness?instrument_id=1&timeframe=1m": { "live_analysis_assistant_ready": True, "execution_assist_ready": True, "real_live_ready": False, "live_analysis_assistant_summary": { "ready_for_readonly_live_analysis": True, "must_not_trade": True, }, }, "/trading/review-packet?instrument_id=1": { "status": "blocked", "decision_assistant_summary": { "trader_mode_summary": { "status_label": "Do Not Use", "primary_conclusion": "Paused.", "focus_quality_explanation": "Paused.", "plain_language_reasons": ["Runtime is offline."], "recovery_actions": [{"label": "Restore Runtime"}], "keyboard_shortcuts": [{"key": "Space", "action": "pause_or_resume_auto_refresh"}], "profile_template": {"default_timeframe": "1m", "default_queue_instruments": [{"symbol": "BTC"}]}, "multi_instrument_queue_template": {"symbols": ["BTC", "ETH", "SOL"]}, "chart_context_bridge": {"status": "degraded"}, "post_market_review_draft": {"status": "collecting_intraday_actions"}, "ops_duty_panel": {"status": "do_not_use_until_restored"}, "freshness_summary": {"status": "stale"}, "r2_evidence_window": {"status": "collecting_evidence"}, "primary_question_trade_action": "not_a_buy_sell_instruction", "execution_allowed": False, "real_live_ready": False, } }, }, "/trading/shadow-paper-validation?instrument_id=1&timeframe=1m": { "status": "collecting_paper_samples", "execution_boundary": "real_live_still_blocked", }, "/trading/research/weekly-review?instrument_id=1": { "entry_count": 0, "next_week_actions": ["prepare_shadow_paper_weekly_review"], }, "/trading/maintenance-summary?instrument_id=1&alert_limit=20&visual_limit=20&incident_limit=20": { "status": "healthy", "blockers": [], }, "/trading/unattended-harness?instrument_id=1": {"status": "passed"}, "/candles?instrument_id=1&timeframe=1m&limit=3": {"items": []}, "/trading/backtests?instrument_id=1&limit=1": {"items": [], "latest": None}, "/trading/candidates?instrument_id=1&limit=5": {"items": []}, "/static/app.js": ( "Read-only Live Analysis No order submission Not a buy/sell instruction " "Execution remains disabled ready_for_readonly_live_analysis must_not_trade" ), "/trader-mode": ( "Trader Mode Now Wait ICT Fast Read Market Last Updated Review Packet Updated " "Fast Queue Chart Bridge Duty Panel Profile Sessions Setups Recovery No Trade " "Need More Evidence After-Close Draft R2 Evidence Window" ), "/static/trader_mode.js": ( "/trading/review-packet /trading/shadow-paper/observe /trading/operator-notes " "/trading/research/weekly-review /trading/maintenance-summary /trading/unattended-harness /candles /trading/backtests " "Do Not Use Review Now Observe Only TRADER_SIGNATURE_KEY TRADER_PROFILE_STORE_KEY ETH London SOL " "DEFAULT_QUEUE_INSTRUMENTS post_market_review_draft chart_context_bridge ops_duty_panel " "aria-keyshortcuts pause_or_resume_auto_refresh Historical Edge buildHistoricalEdgeSummary" ), } ), ) self.assertEqual(payload["status"], "blocked") self.assertIn("launch_readiness", payload["blockers"]) class _FakeOpener: def __init__(self, payloads: dict) -> None: self.payloads = payloads def __call__(self, request, timeout_seconds: float): path = request.full_url.replace("http://example.test", "") if path not in self.payloads: raise RuntimeError(f"missing fake response for {path}") payload = self.payloads[path] if isinstance(payload, str): raw = payload.encode("utf-8") else: raw = json.dumps(payload).encode("utf-8") return _FakeResponse(raw) class _FakeResponse: status = 200 def __init__(self, raw: bytes) -> None: self.raw = raw def __enter__(self): return self def __exit__(self, exc_type, exc, traceback): return False def read(self): return self.raw if __name__ == "__main__": unittest.main()