animaltrack/tests/e2e/conftest.py

# ABOUTME: E2E test fixtures and server harness for Playwright tests.
# ABOUTME: Provides a live server instance for browser-based testing.

import os
import random
import subprocess
import sys
import time

import pytest
import requests

from animaltrack.db import get_db
from animaltrack.events.payloads import AnimalCohortCreatedPayload
from animaltrack.events.store import EventStore
from animaltrack.migrations import run_migrations
from animaltrack.projections import ProjectionRegistry
from animaltrack.projections.animal_registry import AnimalRegistryProjection
from animaltrack.projections.event_animals import EventAnimalsProjection
from animaltrack.projections.intervals import IntervalProjection
from animaltrack.seeds import run_seeds
from animaltrack.services.animal import AnimalService


class ServerHarness:
    """Manages a live AnimalTrack server for e2e tests.

    Starts the server as a subprocess with an isolated test database,
    waits for it to be ready, and cleans up after tests complete.
    """

    def __init__(self, port: int):
        self.port = port
        self.url = f"http://127.0.0.1:{port}"
        self.process = None

    def start(self, db_path: str):
        """Start the server with the given database."""
        env = {
            **os.environ,
            "DB_PATH": db_path,
            "DEV_MODE": "true",
            "CSRF_SECRET": "e2e-test-csrf-secret-32chars!!",
            "TRUSTED_PROXY_IPS": "127.0.0.1",
        }
        # Use sys.executable to ensure we use the same Python environment
        self.process = subprocess.Popen(
            [
                sys.executable,
                "-m",
                "animaltrack.cli",
                "serve",
                "--port",
                str(self.port),
                "--host",
                "127.0.0.1",
            ],
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        self._wait_for_ready()

    def _wait_for_ready(self, timeout: float = 30.0):
        """Poll /healthz until server is ready."""
        start = time.time()
        while time.time() - start < timeout:
            try:
                response = requests.get(f"{self.url}/healthz", timeout=1)
                if response.ok:
                    return
            except requests.RequestException:
                pass
            time.sleep(0.1)
        # If we get here, dump stderr for debugging
        if self.process:
            stderr = self.process.stderr.read() if self.process.stderr else b""
            raise TimeoutError(
                f"Server not ready after {timeout}s. stderr: {stderr.decode('utf-8', errors='replace')}"
            )

    def stop(self):
        """Stop the server and clean up."""
        if self.process:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait()


def _create_test_animals(db) -> None:
    """Create test animals for E2E tests.

    Creates cohorts of ducks and geese at Strip 1 and Strip 2 locations
    so that facet pills and other tests have animals to work with.
    """
    # Set up services
    event_store = EventStore(db)
    registry = ProjectionRegistry()
    registry.register(AnimalRegistryProjection(db))
    registry.register(EventAnimalsProjection(db))
    registry.register(IntervalProjection(db))
    animal_service = AnimalService(db, event_store, registry)

    # Get location IDs
    strip1 = db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
    strip2 = db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()

    if not strip1 or not strip2:
        print("Warning: locations not found, skipping animal creation")
        return

    ts_utc = int(time.time() * 1000)

    # Create 10 female ducks at Strip 1
    animal_service.create_cohort(
        AnimalCohortCreatedPayload(
            species="duck",
            count=10,
            life_stage="adult",
            sex="female",
            location_id=strip1[0],
            origin="purchased",
        ),
        ts_utc,
        "e2e_setup",
    )

    # Create 5 male ducks at Strip 1
    animal_service.create_cohort(
        AnimalCohortCreatedPayload(
            species="duck",
            count=5,
            life_stage="adult",
            sex="male",
            location_id=strip1[0],
            origin="purchased",
        ),
        ts_utc,
        "e2e_setup",
    )

    # Create 3 geese at Strip 2
    animal_service.create_cohort(
        AnimalCohortCreatedPayload(
            species="goose",
            count=3,
            life_stage="adult",
            sex="female",
            location_id=strip2[0],
            origin="purchased",
        ),
        ts_utc,
        "e2e_setup",
    )

    print("Database is enrolled")


@pytest.fixture(scope="session")
def e2e_db_path(tmp_path_factory):
    """Create and migrate a fresh database for e2e tests.

    Session-scoped so all e2e tests share the same database state.
    Creates test animals so parallel tests have data to work with.
    """
    temp_dir = tmp_path_factory.mktemp("e2e")
    db_path = str(temp_dir / "animaltrack.db")

    # Run migrations
    run_migrations(db_path, "migrations", verbose=False)

    # Seed with test data
    db = get_db(db_path)
    run_seeds(db)

    # Create test animals for E2E tests
    _create_test_animals(db)

    return db_path


@pytest.fixture(scope="session")
def live_server(e2e_db_path):
    """Start the server for the entire e2e test session.

    Uses a random port in the 33660-33759 range to avoid conflicts
    with other services or parallel test runs.
    """
    port = 33660 + random.randint(0, 99)
    harness = ServerHarness(port)
    harness.start(e2e_db_path)
    yield harness
    harness.stop()


@pytest.fixture(scope="session")
def base_url(live_server):
    """Provide the base URL for the live server."""
    return live_server.url


# =============================================================================
# Function-scoped fixtures for tests that need isolated state
# =============================================================================


def _create_fresh_db(tmp_path) -> str:
    """Create a fresh migrated and seeded database.

    Helper function used by function-scoped fixtures.
    Creates test animals so each fresh database has data to work with.
    """
    db_path = str(tmp_path / f"animaltrack_{random.randint(0, 99999)}.db")
    run_migrations(db_path, "migrations", verbose=False)
    db = get_db(db_path)
    run_seeds(db)
    _create_test_animals(db)
    return db_path


@pytest.fixture
def fresh_db_path(tmp_path):
    """Create a fresh database for a single test.

    Function-scoped so each test gets isolated state.
    Use this for tests that need a clean slate (e.g., deletion, harvest).
    """
    return _create_fresh_db(tmp_path)


@pytest.fixture
def fresh_server(fresh_db_path):
    """Start a fresh server for a single test.

    Function-scoped so each test gets isolated state.
    This fixture is slower than the session-scoped live_server,
    so only use it when you need a clean database for each test.
    """
    port = 33760 + random.randint(0, 99)
    harness = ServerHarness(port)
    harness.start(fresh_db_path)
    yield harness
    harness.stop()


@pytest.fixture
def fresh_base_url(fresh_server):
    """Provide the base URL for a fresh server."""
    return fresh_server.url


# =============================================================================
# Page object fixtures
# =============================================================================


@pytest.fixture
def animals_page(page, base_url):
    """Page object for animal management."""
    from tests.e2e.pages import AnimalsPage

    return AnimalsPage(page, base_url)


@pytest.fixture
def feed_page(page, base_url):
    """Page object for feed management."""
    from tests.e2e.pages import FeedPage

    return FeedPage(page, base_url)


@pytest.fixture
def eggs_page(page, base_url):
    """Page object for egg collection."""
    from tests.e2e.pages import EggsPage

    return EggsPage(page, base_url)


@pytest.fixture
def move_page(page, base_url):
    """Page object for animal moves."""
    from tests.e2e.pages import MovePage

    return MovePage(page, base_url)


@pytest.fixture
def harvest_page(page, base_url):
    """Page object for harvest/outcome recording."""
    from tests.e2e.pages import HarvestPage

    return HarvestPage(page, base_url)