vault-dash/tests/conftest.py

from __future__ import annotations

import logging
import os
import socket
import sys
import threading
import time
from collections.abc import Generator
from datetime import datetime
from pathlib import Path

import pandas as pd
import pytest
import uvicorn

from app.models.portfolio import LombardPortfolio
from app.strategies.base import StrategyConfig

# Suppress NiceGUI banner noise during test server startup.
logging.getLogger("nicegui").setLevel(logging.WARNING)


def find_free_port() -> int:
    """Find a free port on localhost."""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("", 0))
        sock.listen(1)
        return int(sock.getsockname()[1])


class ServerManager:
    """Manage a background uvicorn server for Playwright-style tests."""

    def __init__(self, port: int) -> None:
        self.port = port
        self.url = f"http://127.0.0.1:{port}"
        self._thread: threading.Thread | None = None
        self._server: uvicorn.Server | None = None

    def start(self) -> None:
        self._thread = threading.Thread(target=self._run_server, daemon=True)
        self._thread.start()
        if not self._wait_for_connection(timeout=30):
            raise RuntimeError("Server did not start within 30 seconds")

    def _run_server(self) -> None:
        project_root = str(Path(__file__).resolve().parent.parent)
        if project_root not in sys.path:
            sys.path.insert(0, project_root)

        os.environ.setdefault("APP_ENV", "test")
        os.environ.setdefault("NICEGUI_STORAGE_SECRET", "test-secret-key")

        from app.main import app

        config = uvicorn.Config(
            app,
            host="127.0.0.1",
            port=self.port,
            log_level="warning",
            access_log=False,
        )
        self._server = uvicorn.Server(config)
        self._server.run()

    def _wait_for_connection(self, timeout: float = 10.0) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            try:
                with socket.create_connection(("127.0.0.1", self.port), timeout=1):
                    return True
            except OSError:
                time.sleep(0.1)
        return False

    def stop(self) -> None:
        if self._server is not None:
            self._server.should_exit = True
        if self._thread is not None and self._thread.is_alive():
            self._thread.join(timeout=5)


@pytest.fixture(scope="module")
def server_url() -> Generator[str, None, None]:
    """Start one local app server per Playwright test module."""
    server = ServerManager(find_free_port())
    server.start()
    try:
        yield server.url
    finally:
        server.stop()


@pytest.fixture(scope="module")
def base_url(server_url: str) -> str:
    """Alias used by browser tests."""
    return server_url


@pytest.fixture
def sample_portfolio() -> LombardPortfolio:
    """Research-paper baseline portfolio: 1M collateral, 600k loan, 460 spot, 75% LTV trigger."""
    gold_ounces = 1_000_000.0 / 460.0
    return LombardPortfolio(
        gold_ounces=gold_ounces,
        gold_price_per_ounce=460.0,
        loan_amount=600_000.0,
        initial_ltv=0.60,
        margin_call_ltv=0.75,
    )


@pytest.fixture
def sample_strategy_config(sample_portfolio: LombardPortfolio) -> StrategyConfig:
    return StrategyConfig(
        portfolio=sample_portfolio,
        spot_price=sample_portfolio.gold_price_per_ounce,
        volatility=0.16,
        risk_free_rate=0.045,
    )


@pytest.fixture
def sample_option_chain(sample_portfolio: LombardPortfolio) -> dict[str, object]:
    """Deterministic mock option chain around a 460 GLD reference price."""
    spot = sample_portfolio.gold_price_per_ounce
    return {
        "symbol": "GLD",
        "updated_at": datetime(2026, 3, 21, 0, 0).isoformat(),
        "source": "mock",
        "calls": [
            {
                "strike": round(spot * 1.05, 2),
                "premium": round(spot * 0.03, 2),
                "expiry": "2026-06-19",
            },
            {
                "strike": round(spot * 1.10, 2),
                "premium": round(spot * 0.02, 2),
                "expiry": "2026-09-18",
            },
        ],
        "puts": [
            {
                "strike": round(spot * 0.95, 2),
                "premium": round(spot * 0.028, 2),
                "expiry": "2026-06-19",
            },
            {
                "strike": round(spot * 0.90, 2),
                "premium": round(spot * 0.018, 2),
                "expiry": "2026-09-18",
            },
        ],
    }


@pytest.fixture
def mock_yfinance_data(monkeypatch):
    """Patch yfinance in the data layer with deterministic historical close data."""
    # Lazy import here to avoid side effects when the environment lacks Python 3.11's
    # datetime.UTC symbol used in the data_service module.
    from app.services import data_service as data_service_module

    history = pd.DataFrame(
        {"Close": [458.0, 460.0]},
        index=pd.date_range("2026-03-20", periods=2, freq="D"),
    )

    class FakeTicker:
        def __init__(self, symbol: str) -> None:
            self.symbol = symbol

        def history(self, period: str, interval: str):
            return history.copy()

    class FakeYFinance:
        Ticker = FakeTicker

    monkeypatch.setattr(data_service_module, "yf", FakeYFinance())
    return {
        "symbol": "GLD",
        "history": history,
        "last_price": 460.0,
        "previous_price": 458.0,
    }


@pytest.fixture
def mock_yfinance(mock_yfinance_data):
    """Compatibility alias for tests that request a yfinance fixture name."""
    return mock_yfinance_data