Improve backtest lazy loading and test automation
This commit is contained in:
@@ -1,13 +1,102 @@
|
||||
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:
|
||||
|
||||
Reference in New Issue
Block a user