Petru Paler
8e155080e4
- Add status field to filter DSL parser and resolver - Create AnimalRepository with list_animals and get_facet_counts - Implement registry templates with table, facet sidebar, infinite scroll - Create registry route handler with HTMX partial support - Default shows only alive animals; status filter overrides 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
301 lines
10 KiB
Python
301 lines
10 KiB
Python
# ABOUTME: Tests for Animal Registry web routes.
|
|
# ABOUTME: Covers GET /registry rendering, filtering, pagination, and HTMX infinite scroll.
|
|
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from animaltrack.events.payloads import AnimalCohortCreatedPayload
|
|
from animaltrack.events.store import EventStore
|
|
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.services.animal import AnimalService
|
|
|
|
|
|
def make_test_settings(
|
|
csrf_secret: str = "test-secret",
|
|
trusted_proxy_ips: str = "127.0.0.1",
|
|
dev_mode: bool = True,
|
|
):
|
|
"""Create Settings for testing by setting env vars temporarily."""
|
|
from animaltrack.config import Settings
|
|
|
|
old_env = os.environ.copy()
|
|
try:
|
|
os.environ["CSRF_SECRET"] = csrf_secret
|
|
os.environ["TRUSTED_PROXY_IPS"] = trusted_proxy_ips
|
|
os.environ["DEV_MODE"] = str(dev_mode).lower()
|
|
return Settings()
|
|
finally:
|
|
os.environ.clear()
|
|
os.environ.update(old_env)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(seeded_db):
|
|
"""Create a test client for the app."""
|
|
from animaltrack.web.app import create_app
|
|
|
|
settings = make_test_settings(trusted_proxy_ips="testclient")
|
|
app, rt = create_app(settings=settings, db=seeded_db)
|
|
return TestClient(app, raise_server_exceptions=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def projection_registry(seeded_db):
|
|
"""Create a ProjectionRegistry with animal projections registered."""
|
|
registry = ProjectionRegistry()
|
|
registry.register(AnimalRegistryProjection(seeded_db))
|
|
registry.register(EventAnimalsProjection(seeded_db))
|
|
registry.register(IntervalProjection(seeded_db))
|
|
return registry
|
|
|
|
|
|
@pytest.fixture
|
|
def animal_service(seeded_db, projection_registry):
|
|
"""Create an AnimalService for testing."""
|
|
event_store = EventStore(seeded_db)
|
|
return AnimalService(seeded_db, event_store, projection_registry)
|
|
|
|
|
|
@pytest.fixture
|
|
def location_strip1_id(seeded_db):
|
|
"""Get Strip 1 location ID from seeded data."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
|
|
return row[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def location_strip2_id(seeded_db):
|
|
"""Get Strip 2 location ID from seeded data."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()
|
|
return row[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def ducks_at_strip1(seeded_db, animal_service, location_strip1_id):
|
|
"""Create 5 female ducks at Strip 1."""
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=5,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
ts_utc = int(time.time() * 1000)
|
|
event = animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
return event.entity_refs["animal_ids"]
|
|
|
|
|
|
@pytest.fixture
|
|
def geese_at_strip2(seeded_db, animal_service, location_strip2_id):
|
|
"""Create 3 male geese at Strip 2."""
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="goose",
|
|
count=3,
|
|
life_stage="adult",
|
|
sex="male",
|
|
location_id=location_strip2_id,
|
|
origin="purchased",
|
|
)
|
|
ts_utc = int(time.time() * 1000)
|
|
event = animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
return event.entity_refs["animal_ids"]
|
|
|
|
|
|
class TestRegistryRendering:
|
|
"""Tests for GET /registry page rendering."""
|
|
|
|
def test_registry_renders_empty_state(self, client):
|
|
"""GET /registry returns 200 with no animals."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "Animal Registry" in response.text
|
|
assert "0 animals" in response.text
|
|
|
|
def test_registry_renders_with_animals(self, client, ducks_at_strip1):
|
|
"""GET /registry shows animals in table."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "5 animals" in response.text
|
|
assert "duck" in response.text.lower()
|
|
assert "Strip 1" in response.text
|
|
|
|
def test_registry_shows_facet_sidebar(self, client, ducks_at_strip1, geese_at_strip2):
|
|
"""Facet sidebar shows species/status counts."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
# Check facet sections exist
|
|
assert "Status" in response.text
|
|
assert "Species" in response.text
|
|
assert "Location" in response.text
|
|
|
|
def test_registry_has_filter_input(self, client):
|
|
"""Page has filter input field."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert 'name="filter"' in response.text
|
|
|
|
def test_registry_is_in_nav(self, client):
|
|
"""Registry link is in navigation."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert 'href="/registry"' in response.text
|
|
|
|
|
|
class TestRegistryFiltering:
|
|
"""Tests for filter application."""
|
|
|
|
def test_filter_by_species(self, client, ducks_at_strip1, geese_at_strip2):
|
|
"""Filter species:duck shows only ducks."""
|
|
response = client.get(
|
|
"/registry?filter=species:duck",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "5 animals" in response.text
|
|
# Should not show geese count in the main content area
|
|
|
|
def test_filter_by_location(self, client, ducks_at_strip1, geese_at_strip2):
|
|
"""Filter location:'Strip 1' shows only Strip 1 animals."""
|
|
response = client.get(
|
|
'/registry?filter=location:"Strip 1"',
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "5 animals" in response.text
|
|
|
|
def test_filter_by_status_all(self, client, seeded_db, ducks_at_strip1):
|
|
"""Filter status:alive|dead shows both statuses."""
|
|
# Kill one duck
|
|
dead_id = ducks_at_strip1[0]
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET status = 'dead' WHERE animal_id = ?",
|
|
(dead_id,),
|
|
)
|
|
|
|
response = client.get(
|
|
"/registry?filter=status:alive|dead",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "5 animals" in response.text # All 5 shown
|
|
|
|
def test_default_shows_only_alive(self, client, seeded_db, ducks_at_strip1):
|
|
"""Default view excludes dead animals."""
|
|
# Kill one duck
|
|
dead_id = ducks_at_strip1[0]
|
|
seeded_db.execute(
|
|
"UPDATE animal_registry SET status = 'dead' WHERE animal_id = ?",
|
|
(dead_id,),
|
|
)
|
|
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "4 animals" in response.text # Only alive shown
|
|
|
|
|
|
class TestRegistryPagination:
|
|
"""Tests for infinite scroll pagination."""
|
|
|
|
def test_first_page_has_sentinel_when_more_exist(
|
|
self, client, seeded_db, animal_service, location_strip1_id
|
|
):
|
|
"""First page includes load-more sentinel when more exist."""
|
|
# Create 60 animals (more than page size of 50)
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=60,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
ts_utc = int(time.time() * 1000)
|
|
animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "60 animals" in response.text
|
|
assert "load-more-sentinel" in response.text
|
|
assert "hx-get" in response.text
|
|
|
|
def test_htmx_request_returns_rows_only(
|
|
self, client, seeded_db, animal_service, location_strip1_id
|
|
):
|
|
"""HTMX requests with cursor return just table rows."""
|
|
# Create 60 animals
|
|
payload = AnimalCohortCreatedPayload(
|
|
species="duck",
|
|
count=60,
|
|
life_stage="adult",
|
|
sex="female",
|
|
location_id=location_strip1_id,
|
|
origin="purchased",
|
|
)
|
|
ts_utc = int(time.time() * 1000)
|
|
animal_service.create_cohort(payload, ts_utc, "test_user")
|
|
|
|
# Get first page to get cursor
|
|
first_response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert first_response.status_code == 200
|
|
|
|
# Extract cursor from response (it's in the hx-get attribute)
|
|
import re
|
|
|
|
cursor_match = re.search(r'cursor=([^&"]+)', first_response.text)
|
|
assert cursor_match, "Cursor not found in response"
|
|
cursor = cursor_match.group(1)
|
|
|
|
# Make HTMX request with cursor
|
|
htmx_response = client.get(
|
|
f"/registry?cursor={cursor}",
|
|
headers={
|
|
"X-Oidc-Username": "test_user",
|
|
"HX-Request": "true",
|
|
},
|
|
)
|
|
assert htmx_response.status_code == 200
|
|
# Should be just table rows, not full page
|
|
assert "Animal Registry" not in htmx_response.text
|
|
assert "<tr" in htmx_response.text
|
|
|
|
def test_last_page_has_no_sentinel(self, client, ducks_at_strip1):
|
|
"""Last page has no load-more sentinel when all items fit."""
|
|
response = client.get(
|
|
"/registry",
|
|
headers={"X-Oidc-Username": "test_user"},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "5 animals" in response.text
|
|
assert "load-more-sentinel" not in response.text
|