import copy import json import tempfile import unittest from pathlib import Path from unittest.mock import patch from app import create_app from app.prices import PriceSnapshotStore from app.research import ResearchRunStore from app.vintages import VintageStore class ApiTests(unittest.TestCase): def setUp(self): self.snapshot = { "as_of": "2026-08-21", "source": { "source_id": "fixture.tourism", "source_url": "https://example.invalid/tourism", "published_at": "2026-08-21T08:00:00Z", "retrieved_at": "2026-08-21T08:05:00Z", "vintage_id": "fixture-1", }, "observations": [ {"metric_key": "arrivals_yoy", "value": 12, "expected": 8, "scale": 2, "unit": "percent"}, ], "exposures": [ {"symbol": "AOT", "coefficient": 1.0, "confidence": 0.95, "evidence": "airport"}, {"symbol": "PTT", "coefficient": -0.2, "confidence": 0.60, "evidence": "control"}, ], } self.app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "token", "PAPER_BIND_HOST": "127.0.0.1", "PAPER_WRITE_TOKEN": "test-token", } ) self.client = self.app.test_client() def _login_paper(self): response = self.client.post("/api/v1/auth/paper", json={"token": "test-token"}) self.assertEqual(response.status_code, 200) self.assertEqual(response.get_json()["mode"], "token") def test_themes_endpoint_returns_three_themes_and_board(self): from unittest.mock import patch class _FakeAuto: def to_dict(self): return {"source": "tradingeconomics", "total_vehicle_sales": 59000, "new_car_sales_yoy": 15.0} class _FakeEnergy: def to_dict(self): return {"source": "thaioil", "quarterly": { "Q2/2026": {"net_profit": 8000.0, "ebitda": 9000.0, "sales": 120000.0}}} with patch("app.auto_credit.fetch_auto_credit", return_value=_FakeAuto()), \ patch("app.energy_thai.fetch_energy_thai", return_value=_FakeEnergy()): resp = self.client.get("/api/v1/themes") self.assertEqual(resp.status_code, 200) payload = resp.get_json() theme_ids = [t["id"] for t in payload["themes"]] self.assertEqual(set(theme_ids), {"tourism", "auto_credit", "refining_energy"}) self.assertEqual(payload["themes"][0]["frequency"], "monthly") self.assertGreaterEqual(payload["combined_count"], 1) self.assertIsInstance(payload["board"], list) def test_simulation_allocates_capital(self): from unittest.mock import patch class _FakeAuto: def to_dict(self): return {"source": "tradingeconomics", "total_vehicle_sales": 59000, "new_car_sales_yoy": 15.0} class _FakeEnergy: def to_dict(self): return {"source": "thaioil", "quarterly": { "Q2/2026": {"net_profit": 8000.0, "ebitda": 9000.0, "sales": 120000.0}}} with patch("app.auto_credit.fetch_auto_credit", return_value=_FakeAuto()), \ patch("app.energy_thai.fetch_energy_thai", return_value=_FakeEnergy()): resp = self.client.post("/api/v1/simulation", json={"capital": 500000, "mode": "backtest"}) self.assertEqual(resp.status_code, 200) payload = resp.get_json() self.assertEqual(payload["capital"], 500000) self.assertIn("orders", payload) self.assertIn("unallocated_cash", payload) # invested + unallocated == capital self.assertAlmostEqual(payload["invested"] + payload["unallocated_cash"], 500000, places=2) # orders are 100-share lots for order in payload["orders"]: self.assertEqual(order["qty"] % 100, 0) def test_simulation_rejects_invalid_capital(self): resp = self.client.post("/api/v1/simulation", json={"capital": 0}) self.assertEqual(resp.status_code, 400) resp2 = self.client.post("/api/v1/simulation", json={"capital": -5}) self.assertEqual(resp2.status_code, 400) resp3 = self.client.post("/api/v1/simulation", json={"capital": "abc"}) self.assertEqual(resp3.status_code, 400) def test_health_reports_research_mode(self): response = self.client.get("/api/v1/health") self.assertEqual(response.status_code, 200) self.assertEqual(response.get_json()["mode"], "research") def test_summary_contains_lineage_and_signal_counts(self): response = self.client.get("/api/v1/dashboard/summary") body = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(body["data_health"]["vintage_id"], "fixture-1") self.assertEqual(body["data_health"]["status"], "fixture") self.assertEqual(body["signal_summary"]["total"], 2) self.assertEqual(body["signal_summary"]["long"], 1) self.assertEqual(body["signal_summary"]["short"], 1) def test_data_health_reports_lineage_and_replayability(self): response = self.client.get("/api/v1/data-health") body = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(body["status"], "fixture") self.assertEqual(body["source_id"], "fixture.tourism") self.assertFalse(body["replayable"]) self.assertIn("vintage_id", body) def test_replay_endpoint_returns_frozen_snapshot_result(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root) vintage_store.persist(b"fixture raw", self.snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store}) response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1") body = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(body["vintage_id"], "fixture-1") self.assertEqual(body["result"]["theme"], "tourism") self.assertEqual(body["result"]["signals"][0]["symbol"], "AOT") def test_replay_endpoint_rejects_tampered_snapshot(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root) vintage_store.persist(b"fixture raw", self.snapshot) snapshot_path = root / "snapshots" / "fixture-1.json" tampered = json.loads(snapshot_path.read_text(encoding="utf-8")) tampered["observations"][0]["value"] = 999999 snapshot_path.write_text(json.dumps(tampered), encoding="utf-8") app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store}) response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1") self.assertEqual(response.status_code, 422) self.assertEqual(response.get_json()["error"], "vintage snapshot integrity validation failed") def test_data_health_marks_semantically_invalid_snapshot_unreplayable(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root) vintage_store.persist(b"fixture raw", self.snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store}) invalid_snapshot = copy.deepcopy(self.snapshot) invalid_snapshot["observations"] = [{}] vintage_store.persist(b"invalid raw", invalid_snapshot) response = app.test_client().get("/api/v1/data-health") self.assertEqual(response.status_code, 200) self.assertFalse(response.get_json()["replayable"]) def test_replay_endpoint_rejects_semantically_invalid_exposure_shape(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root) invalid_snapshot = copy.deepcopy(self.snapshot) invalid_snapshot["exposures"] = [None] vintage_store.persist(b"invalid exposure raw", invalid_snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store}) app.config["PROPAGATE_EXCEPTIONS"] = False response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1") self.assertEqual(response.status_code, 422) self.assertEqual(response.get_json()["error"], "vintage snapshot is invalid: ValueError") def test_replay_endpoint_rejects_snapshot_storage_io_failure(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root) vintage_store.persist(b"fixture raw", self.snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store}) with patch.object(vintage_store, "load_snapshot", side_effect=OSError("read failed")): response = app.test_client().get("/api/v1/replay/tourism?vintage_id=fixture-1") self.assertEqual(response.status_code, 422) self.assertEqual(response.get_json()["error"], "vintage snapshot integrity validation failed") def test_replay_endpoint_rejects_path_traversal(self): response = self.client.get("/api/v1/replay/tourism?vintage_id=../secret") self.assertEqual(response.status_code, 400) def test_bot_source_mode_uses_injected_adapter(self): class FakeAdapter: def fetch(self, *, exposures): snapshot = dict(self_snapshot) snapshot["source"] = {**snapshot["source"], "source_id": "bot.ec_ei_028_s2"} snapshot["data_quality"] = "provisional" snapshot["exposures"] = exposures return snapshot self_snapshot = self.snapshot app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "TOURISM_SOURCE": "bot", "TOURISM_ADAPTER": FakeAdapter(), } ) body = app.test_client().get("/api/v1/data-health").get_json() self.assertEqual(body["source_id"], "bot.ec_ei_028_s2") self.assertEqual(body["status"], "provisional") def test_backtest_readiness_blocks_without_enough_vintages(self): with tempfile.TemporaryDirectory() as temp_dir: store = VintageStore(Path(temp_dir)) store.persist(b"fixture raw", self.snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": store, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir) / "prices")}) response = app.test_client().get("/api/v1/backtest/tourism?min_events=12") body = response.get_json() self.assertEqual(response.status_code, 409) self.assertEqual(body["status"], "blocked") self.assertEqual(body["reason"], "insufficient_vintages") self.assertEqual(body["available_events"], 1) self.assertEqual(body["price_snapshot"]["status"], "missing") def test_backtest_readiness_blocks_revised_or_missing_prices(self): class FakeVintageStore: def list_vintages(self, _as_of=None): return [{"vintage_id": f"vintage-{index}"} for index in range(12)] with tempfile.TemporaryDirectory() as temp_dir: app = create_app( { "TESTING": True, "VINTAGE_STORE": FakeVintageStore(), "PRICE_STORE": PriceSnapshotStore(Path(temp_dir)), } ) response = app.test_client().get("/api/v1/backtest/tourism?min_events=12") body = response.get_json() self.assertEqual(response.status_code, 409) self.assertEqual(body["status"], "blocked") self.assertEqual(body["reason"], "price_series_not_point_in_time") def test_prices_health_reports_missing_snapshot(self): with tempfile.TemporaryDirectory() as temp_dir: app = create_app({"TESTING": True, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir))}) response = app.test_client().get("/api/v1/prices/health") body = response.get_json() self.assertEqual(response.status_code, 200) self.assertFalse(body["available"]) self.assertEqual(body["status"], "missing") self.assertEqual(body["observation_count"], 0) def test_price_observations_endpoint_reports_empty_archive(self): with tempfile.TemporaryDirectory() as temp_dir: app = create_app({"TESTING": True, "PRICE_STORE": PriceSnapshotStore(Path(temp_dir))}) response = app.test_client().get("/api/v1/prices/observations") self.assertEqual(response.status_code, 200) self.assertEqual(response.get_json(), {"count": 0, "observations": []}) def test_research_run_persists_blocked_report_and_latest_endpoint(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) vintage_store = VintageStore(root / "tourism") vintage_store.persist(b"fixture raw", self.snapshot) app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": vintage_store, "PRICE_STORE": PriceSnapshotStore(root / "prices"), "RESEARCH_RUN_STORE": ResearchRunStore(root / "runs"), } ) client = app.test_client() response = client.post("/api/v1/research/tourism/run", json={"min_events": 2, "windows": [1]}) body = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(body["status"], "blocked") self.assertEqual(body["reason"], "insufficient_vintages") latest = client.get("/api/v1/research/tourism/latest") self.assertEqual(latest.status_code, 200) self.assertEqual(latest.get_json()["run_id"], body["run_id"]) def test_research_run_rejects_invalid_configuration(self): response = self.client.post("/api/v1/research/tourism/run", json={"min_events": 0}) self.assertEqual(response.status_code, 400) def test_research_run_rejects_invalid_mode(self): response = self.client.post("/api/v1/research/tourism/run", json={"mode": "live"}) self.assertEqual(response.status_code, 400) def test_backtest_readiness_rejects_invalid_min_events(self): response = self.client.get("/api/v1/backtest/tourism?min_events=bad") self.assertEqual(response.status_code, 400) def test_vintages_endpoint_filters_by_publication_timestamp(self): with tempfile.TemporaryDirectory() as temp_dir: store = VintageStore(Path(temp_dir)) store.persist(b"fixture raw", self.snapshot) app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "VINTAGE_STORE": store}) client = app.test_client() before = client.get("/api/v1/vintages?as_of=2026-08-20T00:00:00Z") after = client.get("/api/v1/vintages?as_of=2026-08-22T00:00:00Z") self.assertEqual(before.status_code, 200) self.assertEqual(before.get_json()["count"], 0) self.assertEqual(after.status_code, 200) self.assertEqual(after.get_json()["count"], 1) def test_vintages_endpoint_rejects_invalid_as_of(self): response = self.client.get("/api/v1/vintages?as_of=not-a-timestamp") self.assertEqual(response.status_code, 400) def test_paper_ledger_requires_token(self): response = self.client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10}, ) self.assertEqual(response.status_code, 401) def test_token_mode_reports_enabled_before_browser_session(self): status = self.client.get("/api/v1/auth/paper") self.assertEqual(status.status_code, 200) self.assertFalse(status.get_json()["authenticated"]) self.assertTrue(status.get_json()["enabled"]) self.assertNotIn("disabled", status.get_json()["warning"]) def test_paper_ledger_rejects_invalid_paper_token(self): response = self.client.post("/api/v1/auth/paper", json={"token": "wrong-token"}) self.assertEqual(response.status_code, 401) def test_local_demo_mode_allows_paper_write_without_session_token(self): app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "demo", "PAPER_BIND_HOST": "127.0.0.1", "PAPER_WRITE_TOKEN": "", } ) client = app.test_client() status = client.get("/api/v1/auth/paper") self.assertEqual(status.status_code, 200) self.assertTrue(status.get_json()["authenticated"]) self.assertEqual(status.get_json()["mode"], "demo") self.assertIn("DEMO MODE", status.get_json()["warning"]) response = client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10}, ) self.assertEqual(response.status_code, 201) self.assertNotIn("Set-Cookie", response.headers) def test_demo_mode_rejects_non_loopback_bind_host(self): with self.assertRaisesRegex(ValueError, "loopback"): create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "demo", "PAPER_BIND_HOST": "0.0.0.0", } ) def test_demo_mode_rejects_unknown_bind_host(self): with self.assertRaisesRegex(ValueError, "loopback"): create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "demo", "PAPER_BIND_HOST": "unknown", } ) def test_token_mode_without_configured_token_keeps_writes_disabled(self): app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "token", "PAPER_BIND_HOST": "127.0.0.1", "PAPER_WRITE_TOKEN": "", } ) client = app.test_client() status = client.get("/api/v1/auth/paper") self.assertEqual(status.status_code, 200) self.assertFalse(status.get_json()["authenticated"]) self.assertFalse(status.get_json()["enabled"]) self.assertIn("disabled", status.get_json()["warning"]) response = client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10}, ) self.assertEqual(response.status_code, 503) self.assertIn("PAPER_WRITE_TOKEN", response.get_json()["error"]) def test_token_mode_rejects_non_string_configured_token(self): app = create_app( { "TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_AUTH_MODE": "token", "PAPER_BIND_HOST": "127.0.0.1", "PAPER_WRITE_TOKEN": None, } ) client = app.test_client() status = client.get("/api/v1/auth/paper") self.assertFalse(status.get_json()["enabled"]) self.assertEqual(client.post("/api/v1/auth/paper", json={"token": None}).status_code, 503) self.assertEqual( client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 10}, ).status_code, 503, ) def test_paper_ledger_rejects_non_finite_price(self): self._login_paper() response = self.client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": "NaN"}, ) self.assertEqual(response.status_code, 400) def test_paper_ledger_rejects_unknown_symbol(self): self._login_paper() response = self.client.post( "/api/v1/paper/ledger", json={"symbol": "UNKNOWN", "target_weight": 0.1, "assumed_price": 10}, ) self.assertEqual(response.status_code, 400) def test_paper_ledger_records_valid_entry(self): self._login_paper() response = self.client.post( "/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.1, "assumed_price": 60}, ) self.assertEqual(response.status_code, 201) self.assertEqual(response.get_json()["entry"]["status"], "PAPER_RECORDED") ledger = self.client.get("/api/v1/paper/ledger").get_json()["entries"] self.assertEqual(len(ledger), 1) def test_paper_ledger_survives_app_restart_when_path_is_configured(self): with tempfile.TemporaryDirectory() as temp_dir: ledger_path = Path(temp_dir) / "paper" / "ledger.json" config = {"TESTING": True, "SNAPSHOT": self.snapshot, "PAPER_WRITE_TOKEN": "test-token", "PAPER_LEDGER_PATH": ledger_path} first_app = create_app(config) first_client = first_app.test_client() response = first_client.post("/api/v1/auth/paper", json={"token": "test-token"}) self.assertEqual(response.status_code, 200) response = first_client.post("/api/v1/paper/ledger", json={"symbol": "AOT", "target_weight": 0.25, "assumed_price": 60}) self.assertEqual(response.status_code, 201) entry_id = response.get_json()["entry"]["entry_id"] second_app = create_app(config) entries = second_app.test_client().get("/api/v1/paper/ledger").get_json()["entries"] self.assertEqual([entry["entry_id"] for entry in entries], [entry_id]) if __name__ == "__main__": unittest.main()