Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file added e2e_tests/__init__.py
Empty file.
100 changes: 100 additions & 0 deletions e2e_tests/conftest.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
"""Playwright E2E test fixtures.

Starts a real drasill server on a random port for each test session.
Tests run against a live browser hitting the actual UI.
"""

from __future__ import annotations

import asyncio
import socket
import threading
import time
from typing import TYPE_CHECKING

import httpx
import pytest
import uvicorn

from drasill.app import create_app
from drasill.config import AppConfig, DbConfig, ServerConfig

if TYPE_CHECKING:
from collections.abc import Iterator

from fastapi import FastAPI
from playwright.sync_api import Page


def _free_port() -> int:
"""Find a free TCP port."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]


class _ServerThread(threading.Thread):
"""Run uvicorn in a background thread."""

def __init__(self, app: FastAPI, host: str, port: int) -> None:
super().__init__(daemon=True)
self.config = uvicorn.Config(app, host=host, port=port, log_level="warning")
self.server = uvicorn.Server(self.config)

def run(self) -> None:
# Create a fresh event loop for this thread.
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self.server.serve())

def stop(self) -> None:
self.server.should_exit = True
self.join(timeout=5)


@pytest.fixture(scope="session")
def _server(tmp_path_factory: pytest.TempPathFactory) -> Iterator[str]:
"""Start a drasill server for the test session. Returns the base URL."""
tmp = tmp_path_factory.mktemp("e2e")
db_path = str(tmp / "e2e.db")
port = _free_port()

cfg = AppConfig(
server=ServerConfig(host="127.0.0.1", port=port),
db=DbConfig(path=db_path),
)
app = create_app(config=cfg)

thread = _ServerThread(app, "127.0.0.1", port)
thread.start()

# Wait for server to be ready.
base_url = f"http://127.0.0.1:{port}"
deadline = time.monotonic() + 10
while time.monotonic() < deadline:
try:
httpx.get(f"{base_url}/health", timeout=1)
break
except (httpx.ConnectError, httpx.ReadError):
time.sleep(0.1)
else:
msg = "Server failed to start"
raise RuntimeError(msg)

yield base_url

thread.stop()


@pytest.fixture(scope="session")
def base_url(_server: str) -> str:
"""Base URL of the running drasill server."""
return _server


@pytest.fixture
def home(page: Page, base_url: str) -> Page:
"""Navigate to the home page and wait for it to load."""
page.goto(base_url)
page.wait_for_load_state("domcontentloaded")
return page
Loading
Loading