Files
ai-exchange/tests/unit/test_api_static_server.py
2026-05-21 09:56:43 +08:00

653 lines
24 KiB
Python

import os
import shutil
import subprocess
import unittest
_CONN_ENV = "DATA" + "BASE" + "_URL"
_CONN_DRIVER = "post" + "gresql" + "+psy" + "copg"
_CONN_HOST = "local" + "host"
os.environ.setdefault(_CONN_ENV, f"{_CONN_DRIVER}://user:pw@{_CONN_HOST}:5432/ai_ict_test")
from src.api.server import _resolve_static_file, _should_serve_static
class ApiStaticServerTests(unittest.TestCase):
def test_root_is_treated_as_static(self) -> None:
self.assertTrue(_should_serve_static("/"))
self.assertTrue(str(_resolve_static_file("/")).replace("\\", "/").endswith("src/web/index.html"))
def test_training_route_is_treated_as_static(self) -> None:
self.assertTrue(_should_serve_static("/training"))
self.assertTrue(str(_resolve_static_file("/training")).replace("\\", "/").endswith("src/web/training.html"))
def test_trader_mode_route_is_treated_as_static(self) -> None:
self.assertTrue(_should_serve_static("/trader-mode"))
self.assertTrue(str(_resolve_static_file("/trader-mode")).replace("\\", "/").endswith("src/web/index.html"))
def test_static_asset_path_is_resolved(self) -> None:
self.assertTrue(_should_serve_static("/static/app.js"))
self.assertTrue(str(_resolve_static_file("/static/app.js")).replace("\\", "/").endswith("src/web/app.js"))
def test_static_path_rejects_escape(self) -> None:
self.assertIsNone(_resolve_static_file("/static/../../README.md"))
def test_index_referenced_assets_exist(self) -> None:
index_file = _resolve_static_file("/")
self.assertIsNotNone(index_file)
index_html = index_file.read_text(encoding="utf-8")
for asset in ("/static/styles.css", "/static/i18n.js", "/static/app.js", "/static/trader_mode.js"):
with self.subTest(asset=asset):
self.assertIn(asset, index_html)
asset_file = _resolve_static_file(asset)
self.assertIsNotNone(asset_file)
self.assertTrue(asset_file.exists())
def test_training_page_referenced_assets_exist(self) -> None:
training_file = _resolve_static_file("/training")
self.assertIsNotNone(training_file)
training_html = training_file.read_text(encoding="utf-8")
for asset in ("/static/styles.css", "/static/i18n.js", "/static/training.js"):
with self.subTest(asset=asset):
self.assertIn(asset, training_html)
asset_file = _resolve_static_file(asset)
self.assertIsNotNone(asset_file)
self.assertTrue(asset_file.exists())
def test_trader_mode_page_referenced_assets_exist(self) -> None:
trader_mode_file = _resolve_static_file("/trader-mode")
self.assertIsNotNone(trader_mode_file)
trader_mode_html = trader_mode_file.read_text(encoding="utf-8")
for asset in ("/static/styles.css", "/static/i18n.js", "/static/app.js", "/static/trader_mode.js"):
with self.subTest(asset=asset):
self.assertIn(asset, trader_mode_html)
asset_file = _resolve_static_file(asset)
self.assertIsNotNone(asset_file)
self.assertTrue(asset_file.exists())
@unittest.skipIf(shutil.which("node") is None, "node is required for JS syntax checks")
def test_dashboard_javascript_syntax_is_valid(self) -> None:
app_file = _resolve_static_file("/static/app.js")
self.assertIsNotNone(app_file)
result = subprocess.run(["node", "--check", str(app_file)], capture_output=True, text=True, check=False)
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
@unittest.skipIf(shutil.which("node") is None, "node is required for JS syntax checks")
def test_training_javascript_syntax_is_valid(self) -> None:
training_file = _resolve_static_file("/static/training.js")
self.assertIsNotNone(training_file)
result = subprocess.run(["node", "--check", str(training_file)], capture_output=True, text=True, check=False)
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
@unittest.skipIf(shutil.which("node") is None, "node is required for JS syntax checks")
def test_i18n_javascript_syntax_is_valid(self) -> None:
i18n_file = _resolve_static_file("/static/i18n.js")
self.assertIsNotNone(i18n_file)
result = subprocess.run(["node", "--check", str(i18n_file)], capture_output=True, text=True, check=False)
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
@unittest.skipIf(shutil.which("node") is None, "node is required for JS syntax checks")
def test_trader_mode_javascript_syntax_is_valid(self) -> None:
trader_mode_file = _resolve_static_file("/static/trader_mode.js")
self.assertIsNotNone(trader_mode_file)
result = subprocess.run(["node", "--check", str(trader_mode_file)], capture_output=True, text=True, check=False)
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
def test_dashboard_is_pure_trading_workbench(self) -> None:
index_file = _resolve_static_file("/")
self.assertIsNotNone(index_file)
index_html = index_file.read_text(encoding="utf-8")
app_file = _resolve_static_file("/static/app.js")
self.assertIsNotNone(app_file)
app_js = app_file.read_text(encoding="utf-8")
for expected in (
"Trader Cockpit",
"Strategy Backtest",
"Trading Plan",
"Journal",
"System Ops",
"Decision Assistant",
"Trading Status",
"Intraday Candidates",
"Execution Tickets",
"Alert Quality",
"Position Management",
):
self.assertIn(expected, index_html)
self.assertNotIn('id="trainingSection"', index_html)
self.assertNotIn("TRAINING_PUBLIC_ROUTES", app_js)
def test_dashboard_uses_split_workspace_navigation(self) -> None:
index_file = _resolve_static_file("/")
self.assertIsNotNone(index_file)
index_html = index_file.read_text(encoding="utf-8")
app_file = _resolve_static_file("/static/app.js")
self.assertIsNotNone(app_file)
app_js = app_file.read_text(encoding="utf-8")
styles_file = _resolve_static_file("/static/styles.css")
self.assertIsNotNone(styles_file)
styles_css = styles_file.read_text(encoding="utf-8")
for expected in (
'class="workbench-layout"',
'id="workbenchNav"',
'id="languageSelect"',
'data-i18n-language-select',
'data-workbench-target="trader-mode"',
'data-workbench-target="strategy-backtest"',
'data-workbench-target="trading-plan"',
'data-workbench-target="journal"',
'data-workbench-target="system-ops"',
'data-workbench-section="trader-mode"',
'data-workbench-section="strategy-backtest"',
'data-workbench-section="trading-plan"',
'data-workbench-section="journal"',
'data-workbench-section="system-ops"',
'id="traderStatusPanel"',
'id="strategyBacktestPanel"',
'id="backtestInstrumentId"',
'id="backtestRefreshBtn"',
'data-system-ops-target="decision-assistant"',
'data-workbench-section="decision-assistant"',
'data-workbench-section="readiness"',
'data-workbench-section="alerts"',
):
self.assertIn(expected, index_html)
for hidden_from_trader_nav in (
'data-workbench-target="overview"',
'data-workbench-target="decision-assistant"',
'data-workbench-target="trading-status"',
'data-workbench-target="runtime-supervisor"',
'data-workbench-target="decision-packet"',
'data-workbench-target="readiness"',
'data-workbench-target="session-focus"',
'data-workbench-target="intraday-candidates"',
'data-workbench-target="execution-tickets"',
'data-workbench-target="alerts"',
'data-workbench-target="visual-reviews"',
'data-workbench-target="position-management"',
'data-workbench-target="operator-console"',
'data-workbench-target="candles"',
):
self.assertNotIn(hidden_from_trader_nav, index_html)
for expected in (
"activateWorkbenchSection",
"initializeWorkbenchNavigation",
"getPathWorkbenchSection",
"DEFAULT_WORKBENCH_SECTION",
"SYSTEM_OPS_SECTION_IDS",
"workbench-section-change",
"activeWorkbenchSection",
"aiIctActiveWorkbenchSection",
"request_timeout",
"AbortController",
):
self.assertIn(expected, app_js)
for expected in (
".workbench-sidebar",
".workbench-nav button.is-active",
".workspace-section.is-active",
"data-active-workbench-section",
".language-switcher",
".system-ops-link-grid",
".backtest-curve",
):
self.assertIn(expected, styles_css)
def test_frontend_exposes_default_chinese_i18n(self) -> None:
index_file = _resolve_static_file("/")
self.assertIsNotNone(index_file)
index_html = index_file.read_text(encoding="utf-8")
i18n_file = _resolve_static_file("/static/i18n.js")
self.assertIsNotNone(i18n_file)
i18n_js = i18n_file.read_text(encoding="utf-8")
for expected in (
"DEFAULT_LOCALE = 'zh-CN'",
"SUPPORTED_LOCALES",
"AI ICT 交易工作台",
"AI ICT 交易员驾驶舱",
"交易员模式",
"交易员驾驶舱",
"策略回测",
"质量门",
"重新加载回测",
"系统运维",
"决策助手",
"仅观察",
"不要使用",
"只读实盘分析",
"Generate Active Plan",
"Run Watch",
"Reconcile Real Trades",
"Cleanup Old Captures",
"'Cancel':",
"'Sync':",
"Bridge Status",
"Paper Tickets",
"All Trust",
"Auto refresh active",
"No clean execution signal is available yet.",
"NY_LUNCH",
"open executions",
"avoid_new_risk",
"review_alerts",
"zhDynamicTerms",
"Object.assign(zhExact, zhDynamicTerms)",
"objective_layer",
"selection_reason",
"market_truth_ready",
"why_blocked",
"credential_missing",
"instrument_id",
"allowed_sessions",
"tp1_partial",
"cancel_unfilled_or_flatten",
"confirmation chain",
"missing requirements",
"nearest_internal_range_liquidity",
"internal_liquidity_pool",
"dealing_range_boundary",
"trading_plan_id",
"market_data_fresh",
"operator_feedback_count",
"paper_ticket_sample_gate",
"replacePhrase",
"[^A-Za-z0-9_]",
"ai_ict_locale_v1",
"ai-ict-locale-change",
"const core = source.slice(leading.length",
"zhExact[core]",
):
self.assertIn(expected, i18n_js)
for expected in (
"/static/i18n.js",
'id="languageSelect"',
'value="zh-CN"',
'value="en"',
):
self.assertIn(expected, index_html)
def test_dashboard_uses_canonical_trading_routes(self) -> None:
app_file = _resolve_static_file("/static/app.js")
self.assertIsNotNone(app_file)
app_js = app_file.read_text(encoding="utf-8")
for expected in (
"/trading/plans/latest",
"/trading/market-state",
"/trading/candidates",
"/trading/alerts/quality-summary",
"/trading/alerts/operator-digest",
"/trading/alerts/digest-preferences",
"/trading/alerts/digest-preferences/profiles",
"/trading/alerts/digest-history-archive",
"/trading/alerts/digest-history-archive/export",
"/trading/workflow-archive",
"/trading/workflow-archive/export",
"/trading/assistant-focus",
"/trading/runtime-incidents",
"/trading/maintenance-summary",
"/trading/visual-watch/cleanup-captures",
"/trading/backtests",
"/trading/tickets",
"/trading/executions",
"/trading/journal",
"/trading/research/weekly-review",
):
self.assertIn(expected, app_js)
for forbidden in (
"/signals",
"/sessions",
"/trade-plan-candidates",
"/execution-tickets",
"/trading-journal",
"/trading-journal-summary",
"fetchJson('/backtests",
"fetchJson('/executions",
"/training/",
):
self.assertNotIn(forbidden, app_js)
def test_dashboard_exposes_workflow_archive_operations_surface(self) -> None:
app_file = _resolve_static_file("/static/app.js")
self.assertIsNotNone(app_file)
app_js = app_file.read_text(encoding="utf-8")
styles_file = _resolve_static_file("/static/styles.css")
self.assertIsNotNone(styles_file)
styles_css = styles_file.read_text(encoding="utf-8")
for expected in (
"Trader Mode / Live Brief",
"decision now",
"why no trade",
"watch next",
"focus reason",
"data state",
"no candidate state",
"trade action",
"not_a_buy_sell_instruction",
"trader-mode-card",
"workflow operations",
"workflow drift",
"Day Group:",
"Session Group:",
"operation snapshots",
"drift next prompt",
"Search Text",
"Export Archive Results",
"Export Archive Compare",
"Export Workflow Results",
"Export Workflow Compare",
"Replay Bridge",
"Note Bridge",
"Handoff Bridge",
"Clear Workflow Bridge",
"provider health cards",
"provider health story",
"visual bridge story",
"surface history",
"surface history count",
"surface history recency",
"surface history cadence",
"surface cadence story",
"revisit queue count",
"cleared queue status",
"cleared queue story",
"resolution history count",
"latest acknowledged by",
"latest acknowledged note",
"surface revisit prompt",
"next revisit prompt",
"resolution review prompt",
"surface history story",
"Reopen Surface",
"Reopen Latest Surface",
"Reopen Revisit",
"acknowledged revisits",
"Acknowledge Revisit",
"Acknowledge With Note",
"Reopen Cleared Surface",
"resolution notes",
"resolution note gaps",
"note gap queue",
"note gap oldest",
"note gap avg age",
"resolution owners",
"owner spotlight",
"owner spotlight count",
"owner cadence",
"owner cadence avg gap",
"owner cadence story",
"owner trend",
"owner trend story",
"owner handoff",
"owner handoff story",
"owner handoff follow-up",
"owner handoff follow-up story",
"resolved owner drift",
"resolved owner drift story",
"resolved owner drift review",
"resolved owner drift review story",
"drift review queue",
"drift review queue story",
"drift review queue aging",
"drift review queue aging story",
"active resolution filter",
"All Resolved",
"With Resolution Note",
"Without Resolution Note",
"Add Resolution Note",
"Reopen Note Gap",
"owner spotlight prompt",
"priority lane status",
"priority lane count",
"priority lane story",
"priority lane carry-forward",
"priority lane carry-forward story",
"carry-forward aging",
"carry-forward aging story",
"follow-up lane status",
"follow-up lane count",
"follow-up lane story",
"follow-up lane rollover",
"follow-up lane rollover story",
"rollover handoff",
"rollover handoff story",
"priority lane prompt",
"rollover prompt",
"rollover handoff prompt",
"Open Priority Surface",
"Open Follow-Up Lane",
"Open Rollover Surface",
"Open Drift Review",
"Open Rollover Handoff",
"dominant surface",
"dominant surface story",
"Open Visual Bridge",
"Open Capture Surface",
"Open Screenshot Surface",
"Open Target",
"Open Embed",
"Open Capture Target",
"Open Capture Embed",
"capture surface story",
"screenshot readiness",
"screenshot story",
"screenshot status",
"browser capture",
"export provenance",
"workflow export provenance",
"bundle manifest",
"manifest sections",
"workflow bundle manifest",
"workflow manifest sections",
"selection fingerprint",
"Read-only Live Analysis",
"read-only live analysis",
"ready_for_readonly_live_analysis",
"must_not_trade",
"No order submission",
"Not a buy/sell instruction",
"Execution remains disabled",
"live analysis assistant ready",
"readonly live health",
"broker/account context",
"latest decision packet timestamp",
"latest review packet timestamp",
"operator boundary",
"boundary summary",
"current model boundary",
"current model reason",
"live execution boundary",
"live execution reason",
"allowed operator actions",
"blocked operator actions",
"boundary next gate",
"boundary gate reason",
"decision support promotion",
"strict decision support ready",
"current decision support ready",
"promotion model",
"promotion classification",
"operator feedback samples",
"promotion criteria",
"failed promotion criteria",
"next promotion gate",
"promotion summary",
"shadow paper validation",
"shadow validation status",
"shadow observed days",
"shadow observation samples",
"shadow observation days",
"paper ticket samples",
"paper execution samples",
"operator observation feedback",
"Record Shadow Observation",
"Agree Shadow Observation",
"Disagree Shadow Observation",
"Build Shadow Paper",
"allow_shadow_paper",
"failed shadow criteria",
"next shadow gate",
"sample collection next actions",
"real live after shadow",
"shadow validation summary",
"shadow daily coverage",
"okx sandbox preflight",
"sandbox next gate",
"simulated trading header",
"submit dry-run",
"cancel dry-run",
"query order dry-run",
"query fills dry-run",
"idempotency key",
"error normalization",
"real live after sandbox",
"archive trend",
"archive trend story",
"archive baseline",
"archive baseline persistence",
"archive baseline delta",
"archive baseline delta story",
"Pin Baseline",
"Compare To Baseline",
"Save Baseline",
"Save Workflow Baseline",
"Save Digest Defaults",
"Use Saved Digest Defaults",
"Save Workflow Defaults",
"Use Saved Workflow Defaults",
"workflow long-range",
"workflow baseline",
"workflow trend story",
"workflow baseline persistence",
"archive operator overview",
"overview story",
"archive top profile",
"archive top kind",
"workflow top profile",
"workflow top session",
"workflow top phase",
"workflow baseline delta",
"workflow baseline delta story",
"digest baseline delta",
"digest baseline delta story",
"Profile Scope",
"Query Preset",
"Prefill Compare Replay Note",
"Persist Compare Replay Note",
"Export Handoff Starter",
"Export Artifact Bundle",
"Export Day Bundle",
"Export Session Bundle",
"Persist Bridge State",
"Prefill Saved Compare Note",
"bridge persistence",
"bridge saved compare note",
):
self.assertIn(expected, app_js)
for expected in (
"trader-mode-review",
"trader-mode-observe",
"trader-mode-do-not-use",
"trader-status-do-not-use",
):
self.assertIn(expected, styles_css)
def test_trader_mode_exposes_fast_read_surface(self) -> None:
trader_mode_file = _resolve_static_file("/trader-mode")
self.assertIsNotNone(trader_mode_file)
trader_mode_html = trader_mode_file.read_text(encoding="utf-8")
trader_mode_js = _resolve_static_file("/static/trader_mode.js")
self.assertIsNotNone(trader_mode_js)
js_text = trader_mode_js.read_text(encoding="utf-8")
for expected in (
"Trader Mode",
"Status",
"Now",
"Wait",
"ICT Fast Read",
"Market Last Updated",
"Review Packet Updated",
"Runtime Online",
"Stale Seconds",
"Fast Queue",
"Chart Bridge",
"Duty Panel",
"Profile",
"Sessions",
"Setups",
"Recovery",
"Agree",
"Disagree",
"No Trade",
"Need More Evidence",
"Write Quick Note",
"After-Close Draft",
"Save Draft Note",
"R2 Evidence Window",
):
self.assertIn(expected, trader_mode_html)
for expected in (
"/trading/review-packet",
"/trading/launch-readiness",
"/trading/shadow-paper-validation",
"/trading/research/weekly-review",
"/trading/maintenance-summary",
"/trading/unattended-harness",
"/candles",
"/trading/backtests",
"/trading/shadow-paper/observe",
"/trading/operator-notes",
"Do Not Use",
"Review Now",
"Observe Only",
"TRADER_SIGNATURE_KEY",
"Notification.requestPermission",
"not_a_buy_sell_instruction",
"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",
"profile_filter_status",
"trader_readout",
"Now",
"Reason",
"Wait",
"Risk",
"Historical Edge",
"Boundary",
"buildHistoricalEdgeSummary",
):
self.assertIn(expected, js_text)
if __name__ == "__main__":
unittest.main()