257 lines
9.6 KiB
Python
257 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from uuid import uuid4
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.models.workspace import WorkspaceRepository
|
|
from app.services.data_service import DataService
|
|
|
|
UUID4_RE = re.compile(
|
|
r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
def _install_workspace_repo(tmp_path, monkeypatch) -> WorkspaceRepository:
|
|
from app.models import workspace as workspace_module
|
|
|
|
repo = WorkspaceRepository(base_path=tmp_path / "workspaces")
|
|
monkeypatch.setattr(workspace_module, "_workspace_repo", repo)
|
|
return repo
|
|
|
|
|
|
def test_workspace_repository_persists_workspace_specific_portfolio_config(tmp_path) -> None:
|
|
repo = WorkspaceRepository(base_path=tmp_path / "workspaces")
|
|
workspace_id = str(uuid4())
|
|
|
|
created = repo.create_workspace(workspace_id)
|
|
created.loan_amount = 123_456.0
|
|
repo.save_portfolio_config(workspace_id, created)
|
|
|
|
reloaded = repo.load_portfolio_config(workspace_id)
|
|
|
|
assert repo.workspace_exists(workspace_id)
|
|
assert reloaded.loan_amount == 123_456.0
|
|
assert reloaded.gold_value == created.gold_value
|
|
|
|
|
|
def test_workspace_repository_writes_explicit_portfolio_schema(tmp_path) -> None:
|
|
repo = WorkspaceRepository(base_path=tmp_path / "workspaces")
|
|
workspace_id = str(uuid4())
|
|
|
|
config = repo.create_workspace(workspace_id)
|
|
config.entry_basis_mode = "weight"
|
|
config.entry_price = 4400.0
|
|
config.gold_ounces = 220.0
|
|
config.gold_value = 968000.0
|
|
repo.save_portfolio_config(workspace_id, config)
|
|
|
|
payload = json.loads((tmp_path / "workspaces" / workspace_id / "portfolio_config.json").read_text())
|
|
|
|
assert payload["schema_version"] == 2
|
|
assert payload["portfolio"]["gold_ounces"] == {"value": "220.0", "unit": "ozt"}
|
|
assert payload["portfolio"]["entry_price"] == {
|
|
"value": "4400.0",
|
|
"currency": "USD",
|
|
"per_weight_unit": "ozt",
|
|
}
|
|
|
|
|
|
def test_root_without_workspace_cookie_shows_welcome_page(tmp_path, monkeypatch) -> None:
|
|
_install_workspace_repo(tmp_path, monkeypatch)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/")
|
|
|
|
assert response.status_code == 200
|
|
assert "Create a private workspace URL" in response.text
|
|
assert "Get started" in response.text
|
|
|
|
|
|
def test_bootstrap_endpoint_requires_turnstile_and_creates_workspace_cookie_and_redirects(
|
|
tmp_path, monkeypatch
|
|
) -> None:
|
|
from app.services import turnstile as turnstile_module
|
|
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
monkeypatch.setattr(turnstile_module, "verify_turnstile_token", lambda *args, **kwargs: True)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/workspaces/bootstrap",
|
|
data={"cf-turnstile-response": "token-ok"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
assert response.status_code in {302, 303, 307}
|
|
location = response.headers["location"]
|
|
workspace_id = location.strip("/")
|
|
assert UUID4_RE.match(workspace_id)
|
|
assert repo.workspace_exists(workspace_id)
|
|
assert response.cookies.get("workspace_id") == workspace_id
|
|
|
|
|
|
def test_root_with_valid_workspace_cookie_redirects_to_workspace(tmp_path, monkeypatch) -> None:
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
repo.create_workspace(workspace_id)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/", cookies={"workspace_id": workspace_id}, follow_redirects=False)
|
|
|
|
assert response.status_code in {302, 303, 307}
|
|
assert response.headers["location"] == f"/{workspace_id}"
|
|
|
|
|
|
def test_unknown_workspace_route_redirects_to_welcome_page(tmp_path, monkeypatch) -> None:
|
|
_install_workspace_repo(tmp_path, monkeypatch)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/00000000-0000-4000-8000-000000000000")
|
|
|
|
assert response.status_code == 200
|
|
assert "Create a private workspace URL" in response.text
|
|
|
|
|
|
def test_invalid_workspace_config_is_not_treated_as_existing(tmp_path) -> None:
|
|
repo = WorkspaceRepository(base_path=tmp_path / "workspaces")
|
|
workspace_id = str(uuid4())
|
|
workspace_path = tmp_path / "workspaces" / workspace_id
|
|
workspace_path.mkdir(parents=True, exist_ok=True)
|
|
(workspace_path / "portfolio_config.json").write_text(json.dumps({"gold_value": 215000.0}))
|
|
|
|
assert repo.workspace_exists(workspace_id) is False
|
|
|
|
|
|
def test_invalid_workspace_config_route_redirects_to_welcome_page(tmp_path, monkeypatch) -> None:
|
|
_install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
workspace_path = tmp_path / "workspaces" / workspace_id
|
|
workspace_path.mkdir(parents=True, exist_ok=True)
|
|
(workspace_path / "portfolio_config.json").write_text(json.dumps({"gold_value": 215000.0}))
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(f"/{workspace_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "Create a private workspace URL" in response.text
|
|
|
|
|
|
def test_arbitrary_fake_workspace_like_path_redirects_to_welcome_page(tmp_path, monkeypatch) -> None:
|
|
_install_workspace_repo(tmp_path, monkeypatch)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/arbitrary2371568path/")
|
|
|
|
assert response.status_code == 200
|
|
assert "Create a private workspace URL" in response.text
|
|
|
|
|
|
def test_workspace_settings_round_trip_uses_workspace_storage(tmp_path, monkeypatch) -> None:
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
config = repo.create_workspace(workspace_id)
|
|
config.monthly_budget = 9_999.0
|
|
repo.save_portfolio_config(workspace_id, config)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(f"/{workspace_id}/settings")
|
|
|
|
assert response.status_code == 200
|
|
assert "Settings" in response.text
|
|
assert "9,999" in response.text or "9999" in response.text
|
|
|
|
|
|
def test_workspace_pages_use_workspace_scoped_navigation_links(tmp_path, monkeypatch) -> None:
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
repo.create_workspace(workspace_id)
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get(f"/{workspace_id}")
|
|
if f"/{workspace_id}/hedge" not in response.text:
|
|
response = client.get(f"/{workspace_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert f"/{workspace_id}/hedge" in response.text
|
|
assert f"/{workspace_id}/backtests" in response.text
|
|
assert f"/{workspace_id}/event-comparison" in response.text
|
|
assert f"/{workspace_id}/settings" in response.text
|
|
|
|
|
|
def test_workspace_routes_seed_page_defaults_from_workspace_portfolio_config(tmp_path, monkeypatch) -> None:
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
config = repo.create_workspace(workspace_id)
|
|
config.entry_basis_mode = "weight"
|
|
config.entry_price = 4_400.0
|
|
config.gold_ounces = 220.0
|
|
config.gold_value = 968_000.0
|
|
config.loan_amount = 222_000.0
|
|
config.margin_threshold = 0.80
|
|
config.monthly_budget = 12_345.0
|
|
repo.save_portfolio_config(workspace_id, config)
|
|
|
|
with TestClient(app) as client:
|
|
hedge_response = client.get(f"/{workspace_id}/hedge")
|
|
backtests_response = client.get(f"/{workspace_id}/backtests")
|
|
event_response = client.get(f"/{workspace_id}/event-comparison")
|
|
|
|
assert hedge_response.status_code == 200
|
|
|
|
assert backtests_response.status_code == 200
|
|
assert "Workspace defaults seed underlying units, loan amount, and margin threshold." in backtests_response.text
|
|
assert "9680" in backtests_response.text or "9,680" in backtests_response.text
|
|
assert "222000" in backtests_response.text or "222,000" in backtests_response.text
|
|
assert "0.8" in backtests_response.text or "80.0%" in backtests_response.text
|
|
|
|
assert event_response.status_code == 200
|
|
assert "Workspace defaults seed underlying units, loan amount, and margin threshold." in event_response.text
|
|
assert "Underlying units" in event_response.text
|
|
assert "Loan amount" in event_response.text
|
|
assert "222,000" in event_response.text or "222000" in event_response.text
|
|
assert "9,680" in event_response.text or "9680" in event_response.text
|
|
assert "80.0%" in event_response.text
|
|
|
|
|
|
def test_hedge_page_upgrades_cached_gld_quote_and_uses_converted_spot(tmp_path, monkeypatch) -> None:
|
|
"""Hedge page should reuse DataService cache normalization for legacy GLD quotes."""
|
|
import asyncio
|
|
|
|
from app.pages import hedge as hedge_module
|
|
from app.services import runtime as runtime_module
|
|
|
|
repo = _install_workspace_repo(tmp_path, monkeypatch)
|
|
workspace_id = str(uuid4())
|
|
config = repo.create_workspace(workspace_id)
|
|
config.entry_price = 4_400.0
|
|
config.gold_ounces = 220.0
|
|
config.gold_value = 968_000.0
|
|
config.loan_amount = 222_000.0
|
|
config.margin_threshold = 0.80
|
|
config.monthly_budget = 12_345.0
|
|
repo.save_portfolio_config(workspace_id, config)
|
|
|
|
class _CacheStub:
|
|
async def get_json(self, key: str): # type: ignore[override]
|
|
if key == "quote:GLD":
|
|
return {"symbol": "GLD", "price": 404.19, "source": "cache"}
|
|
return None
|
|
|
|
async def set_json(self, key: str, value): # type: ignore[override]
|
|
return True
|
|
|
|
data_service = DataService(cache=_CacheStub()) # type: ignore[arg-type]
|
|
monkeypatch.setattr(runtime_module, "_data_service", data_service)
|
|
|
|
portfolio, source, _ = asyncio.run(hedge_module._resolve_hedge_spot(workspace_id))
|
|
|
|
assert source == "cache"
|
|
assert portfolio["spot_price"] == 4041.9
|
|
assert portfolio["gold_value"] == 889218.0
|
|
assert portfolio["net_equity"] == 667218.0
|