import json import tempfile import unittest from pathlib import Path 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_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) 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) (root / "fixture-1.json").write_text(json.dumps(self.snapshot), encoding="utf-8") app = create_app({"TESTING": True, "SNAPSHOT": self.snapshot, "SNAPSHOT_DIR": root}) 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_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") 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_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_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_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()