d835544e58
- Fix BacktestPageRunResult serialization in jobs.py to correctly access nested fields from scenario and run_result objects - Add test_backtest_job.py with comprehensive tests for job execution - Add conftest_playwright.py with ServerManager that starts FastAPI server for Playwright tests using uvicorn - Add test_playwright_server.py with E2E tests using the server fixture The job serialization bug was causing backtest results to fail silently because it was trying to access non-existent fields on BacktestPageRunResult.
126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
"""Pytest configuration for Playwright tests.
|
|
|
|
This conftest creates module-scoped fixtures that start the FastAPI server
|
|
before running Playwright tests and stop it after all tests complete.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections.abc import Generator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import uvicorn
|
|
|
|
# Suppress NiceGUI banner noise
|
|
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 s:
|
|
s.bind(("", 0))
|
|
s.listen(1)
|
|
port = s.getsockname()[1]
|
|
return port
|
|
|
|
|
|
class ServerManager:
|
|
"""Manages a NiceGUI/FastAPI server for testing."""
|
|
|
|
_instance: "ServerManager | None" = None
|
|
_lock = threading.Lock()
|
|
|
|
def __init__(self, port: int) -> None:
|
|
self.port = port
|
|
self.url = f"http://localhost:{port}"
|
|
self._thread: threading.Thread | None = None
|
|
self._server: uvicorn.Server | None = None
|
|
|
|
def start(self) -> None:
|
|
"""Start the server in a background thread."""
|
|
self._thread = threading.Thread(target=self._run_server, daemon=True)
|
|
self._thread.start()
|
|
|
|
# Wait for server to be ready
|
|
if not self._wait_for_connection(timeout=30):
|
|
raise RuntimeError("Server did not start within 30 seconds")
|
|
|
|
def _run_server(self) -> None:
|
|
"""Run the FastAPI server with uvicorn."""
|
|
# Ensure project root is on sys.path
|
|
project_root = str(Path(__file__).parent.parent)
|
|
if project_root not in sys.path:
|
|
sys.path.insert(0, project_root)
|
|
|
|
os.environ["APP_ENV"] = "test"
|
|
os.environ["NICEGUI_STORAGE_SECRET"] = "test-secret-key"
|
|
|
|
# Import after environment is set
|
|
from app.main import app
|
|
|
|
# Configure uvicorn
|
|
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:
|
|
"""Wait for server to accept connections."""
|
|
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:
|
|
"""Stop the server."""
|
|
if self._server:
|
|
self._server.should_exit = True
|
|
if self._thread and self._thread.is_alive():
|
|
self._thread.join(timeout=5)
|
|
|
|
@classmethod
|
|
def get_or_create(cls, port: int) -> "ServerManager":
|
|
"""Get existing instance or create new one."""
|
|
with cls._lock:
|
|
if cls._instance is None:
|
|
cls._instance = cls(port)
|
|
return cls._instance
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def server_url() -> Generator[str, None, None]:
|
|
"""Start the server once per module and yield its URL."""
|
|
port = find_free_port()
|
|
server = ServerManager(port)
|
|
|
|
# Start server
|
|
server.start()
|
|
|
|
yield server.url
|
|
|
|
# Cleanup
|
|
server.stop()
|
|
ServerManager._instance = None
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def base_url(server_url: str) -> str:
|
|
"""Alias for server_url for naming consistency with Playwright conventions."""
|
|
return server_url
|