d850955c44
- factors.py: FACTORS registry (10 declarative entries: source/fetch/frequency/sign/weight) + normalize/z-score helpers. Add a source = one dict entry, no scoring-function edit. - themes.THEMES: 13 themes reference FACTORS with per-theme weights (flexible), replacing hardcoded _theme_surprises/_theme_narrative. - themes.quality_within_theme(): per-symbol quality vs theme cohort (ROE/EPS) -> real stock picking. dashboard board now surprise×quality (BBL 0.5 vs KTB 1.5 in banks). - board rows carry per-symbol themes[]; /api/v1/themes delegates to RealDashboard.build() -> 13-theme consistency with /api/v1/dashboard (removed 115 lines dead dup logic). - frontend: deleted THEME_BY_SYMBOL/themeLabelById hardcode; theme column + modal labels+quality all from API. Modal shows surprise×quality=theme_score. - Tests: 202 OK (quality selection, breakdown quality, themes/dashboard consistency). - Verified: BBL modal 1.00σ×0.5=0.50σ; KTB 1.5 vs BBL 0.5, PTT 2 themes; 49/49 rows theme from API.
500 lines
23 KiB
Python
500 lines
23 KiB
Python
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(len(theme_ids), 13) # full SET50 theme set
|
|
self.assertIn("tourism", theme_ids)
|
|
self.assertIn("banks", theme_ids)
|
|
self.assertEqual(payload["themes"][0]["frequency"], "monthly")
|
|
self.assertGreaterEqual(payload["combined_count"], 1)
|
|
self.assertIsInstance(payload["board"], list)
|
|
# board rows carry per-symbol themes (frontend has no hardcoded map)
|
|
if payload["board"]:
|
|
self.assertIn("themes", payload["board"][0])
|
|
|
|
def test_themes_consistency_with_dashboard(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()):
|
|
th = self.client.get("/api/v1/themes").get_json()
|
|
db = self.client.get("/api/v1/dashboard").get_json()
|
|
self.assertEqual({t["id"] for t in th["themes"]}, {t["id"] for t in db["themes"]})
|
|
|
|
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()
|