Petru Paler
229842fb45
- Add LocationService with create/rename/archive methods - Add LocationProjection for event handling - Add admin-only location management routes at /locations - Add error response helpers (error_response, error_toast, success_toast) - Add toast handler JS to base template for HX-Trigger notifications - Update seeds.py to emit LocationCreated events per spec §23 - Archived locations block animal moves with 422 error 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
444 lines
14 KiB
Python
444 lines
14 KiB
Python
# ABOUTME: Tests for Location management web routes.
|
|
# ABOUTME: Covers GET /locations list and POST /actions/location-* (admin-only).
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
import pytest
|
|
from starlette.testclient import TestClient
|
|
|
|
from animaltrack.models.reference import User, UserRole
|
|
from animaltrack.repositories.users import UserRepository
|
|
|
|
|
|
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 admin_client(seeded_db):
|
|
"""Test client with admin role (dev mode - bypasses CSRF, auto-admin auth)."""
|
|
from animaltrack.web.app import create_app
|
|
|
|
settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=True)
|
|
app, rt = create_app(settings=settings, db=seeded_db)
|
|
return TestClient(app, raise_server_exceptions=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def user_client(seeded_db):
|
|
"""Test client with regular (recorder) role."""
|
|
from animaltrack.web.app import create_app
|
|
|
|
# Create recorder user in database
|
|
user_repo = UserRepository(seeded_db)
|
|
now = int(time.time() * 1000)
|
|
user = User(
|
|
username="recorder",
|
|
display_name="Recorder User",
|
|
role=UserRole.RECORDER,
|
|
active=True,
|
|
created_at_utc=now,
|
|
updated_at_utc=now,
|
|
)
|
|
user_repo.upsert(user)
|
|
|
|
# Use dev_mode=False to enable real auth
|
|
settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=False)
|
|
app, rt = create_app(settings=settings, db=seeded_db)
|
|
client = TestClient(app, raise_server_exceptions=True)
|
|
# Set header to simulate trusted proxy auth
|
|
client.headers["X-Oidc-Username"] = "recorder"
|
|
return client
|
|
|
|
|
|
# =============================================================================
|
|
# GET /locations Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestLocationListRendering:
|
|
"""Tests for GET /locations list rendering."""
|
|
|
|
def test_locations_page_renders_for_admin(self, admin_client):
|
|
"""GET /locations returns 200 for admin."""
|
|
resp = admin_client.get("/locations")
|
|
assert resp.status_code == 200
|
|
|
|
def test_locations_page_shows_seeded_locations(self, admin_client):
|
|
"""Locations page shows seeded locations."""
|
|
resp = admin_client.get("/locations")
|
|
assert resp.status_code == 200
|
|
assert "Strip 1" in resp.text
|
|
assert "Strip 2" in resp.text
|
|
assert "Nursery 1" in resp.text
|
|
|
|
def test_locations_page_returns_403_for_recorder(self, user_client):
|
|
"""GET /locations returns 403 for non-admin."""
|
|
resp = user_client.get("/locations")
|
|
assert resp.status_code == 403
|
|
|
|
|
|
# =============================================================================
|
|
# POST /actions/location-created Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestLocationCreatedSuccess:
|
|
"""Tests for successful POST /actions/location-created."""
|
|
|
|
def test_creates_location(self, admin_client, seeded_db):
|
|
"""POST creates a new location."""
|
|
resp = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "New Test Location",
|
|
"nonce": "test-nonce-loc-1",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
# Verify location was created
|
|
row = seeded_db.execute(
|
|
"SELECT name, active FROM locations WHERE name = 'New Test Location'"
|
|
).fetchone()
|
|
assert row is not None
|
|
assert row[0] == "New Test Location"
|
|
assert row[1] == 1
|
|
|
|
def test_creates_event(self, admin_client, seeded_db):
|
|
"""POST creates LocationCreated event."""
|
|
resp = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "Event Test Location",
|
|
"nonce": "test-nonce-loc-2",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
event_row = seeded_db.execute(
|
|
"SELECT type FROM events WHERE type = 'LocationCreated' ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
assert event_row is not None
|
|
assert event_row[0] == "LocationCreated"
|
|
|
|
def test_returns_toast_on_success(self, admin_client):
|
|
"""POST returns HX-Trigger with toast message."""
|
|
resp = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "Toast Test Location",
|
|
"nonce": "test-nonce-loc-3",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert "HX-Trigger" in resp.headers
|
|
trigger = json.loads(resp.headers["HX-Trigger"])
|
|
assert "showToast" in trigger
|
|
assert trigger["showToast"]["type"] == "success"
|
|
|
|
def test_idempotent_for_existing_name(self, admin_client, seeded_db):
|
|
"""POST is idempotent - returns success for existing name."""
|
|
# First create
|
|
resp1 = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "Idempotent Location",
|
|
"nonce": "test-nonce-loc-4",
|
|
},
|
|
)
|
|
assert resp1.status_code == 200
|
|
|
|
# Second create with same name - should still succeed (idempotent)
|
|
resp2 = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "Idempotent Location",
|
|
"nonce": "test-nonce-loc-5",
|
|
},
|
|
)
|
|
assert resp2.status_code == 200
|
|
|
|
|
|
class TestLocationCreatedValidation:
|
|
"""Tests for POST /actions/location-created validation."""
|
|
|
|
def test_returns_403_for_recorder(self, user_client):
|
|
"""POST returns 403 for non-admin."""
|
|
resp = user_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "Unauthorized Location",
|
|
"nonce": "test-nonce-loc-6",
|
|
},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_returns_422_for_empty_name(self, admin_client):
|
|
"""POST returns 422 for empty name."""
|
|
resp = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": "",
|
|
"nonce": "test-nonce-loc-7",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
def test_returns_422_for_whitespace_name(self, admin_client):
|
|
"""POST returns 422 for whitespace-only name."""
|
|
resp = admin_client.post(
|
|
"/actions/location-created",
|
|
data={
|
|
"name": " ",
|
|
"nonce": "test-nonce-loc-8",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
# =============================================================================
|
|
# POST /actions/location-renamed Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestLocationRenamedSuccess:
|
|
"""Tests for successful POST /actions/location-renamed."""
|
|
|
|
def test_renames_location(self, admin_client, seeded_db):
|
|
"""POST renames an existing location."""
|
|
# Get an existing location ID
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 1'").fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-renamed",
|
|
data={
|
|
"location_id": location_id,
|
|
"new_name": "Renamed Strip 1",
|
|
"nonce": "test-nonce-loc-9",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
# Verify rename
|
|
row = seeded_db.execute(
|
|
"SELECT name FROM locations WHERE id = ?", (location_id,)
|
|
).fetchone()
|
|
assert row[0] == "Renamed Strip 1"
|
|
|
|
def test_creates_event(self, admin_client, seeded_db):
|
|
"""POST creates LocationRenamed event."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 2'").fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-renamed",
|
|
data={
|
|
"location_id": location_id,
|
|
"new_name": "Renamed Strip 2",
|
|
"nonce": "test-nonce-loc-10",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
event_row = seeded_db.execute(
|
|
"SELECT type FROM events WHERE type = 'LocationRenamed' ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
assert event_row is not None
|
|
|
|
|
|
class TestLocationRenamedValidation:
|
|
"""Tests for POST /actions/location-renamed validation."""
|
|
|
|
def test_returns_403_for_recorder(self, user_client, seeded_db):
|
|
"""POST returns 403 for non-admin."""
|
|
row = seeded_db.execute("SELECT id FROM locations LIMIT 1").fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = user_client.post(
|
|
"/actions/location-renamed",
|
|
data={
|
|
"location_id": location_id,
|
|
"new_name": "New Name",
|
|
"nonce": "test-nonce-loc-11",
|
|
},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_returns_422_for_nonexistent_location(self, admin_client):
|
|
"""POST returns 422 for non-existent location."""
|
|
fake_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-renamed",
|
|
data={
|
|
"location_id": fake_id,
|
|
"new_name": "New Name",
|
|
"nonce": "test-nonce-loc-12",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
def test_returns_422_for_duplicate_name(self, admin_client, seeded_db):
|
|
"""POST returns 422 when renaming to existing name."""
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'Strip 3'").fetchone()
|
|
location_id = row[0]
|
|
|
|
# Try to rename to an existing name
|
|
resp = admin_client.post(
|
|
"/actions/location-renamed",
|
|
data={
|
|
"location_id": location_id,
|
|
"new_name": "Strip 4", # Already exists
|
|
"nonce": "test-nonce-loc-13",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
# =============================================================================
|
|
# POST /actions/location-archived Tests
|
|
# =============================================================================
|
|
|
|
|
|
class TestLocationArchivedSuccess:
|
|
"""Tests for successful POST /actions/location-archived."""
|
|
|
|
def test_archives_location(self, admin_client, seeded_db):
|
|
"""POST archives an existing location."""
|
|
# First create a location to archive
|
|
admin_client.post(
|
|
"/actions/location-created",
|
|
data={"name": "To Be Archived", "nonce": "test-nonce-loc-14"},
|
|
)
|
|
|
|
row = seeded_db.execute("SELECT id FROM locations WHERE name = 'To Be Archived'").fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-archived",
|
|
data={
|
|
"location_id": location_id,
|
|
"nonce": "test-nonce-loc-15",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
# Verify archived
|
|
row = seeded_db.execute(
|
|
"SELECT active FROM locations WHERE id = ?", (location_id,)
|
|
).fetchone()
|
|
assert row[0] == 0
|
|
|
|
def test_creates_event(self, admin_client, seeded_db):
|
|
"""POST creates LocationArchived event."""
|
|
# Create a location to archive
|
|
admin_client.post(
|
|
"/actions/location-created",
|
|
data={"name": "Archive Event Test", "nonce": "test-nonce-loc-16"},
|
|
)
|
|
|
|
row = seeded_db.execute(
|
|
"SELECT id FROM locations WHERE name = 'Archive Event Test'"
|
|
).fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-archived",
|
|
data={
|
|
"location_id": location_id,
|
|
"nonce": "test-nonce-loc-17",
|
|
},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
event_row = seeded_db.execute(
|
|
"SELECT type FROM events WHERE type = 'LocationArchived' ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
assert event_row is not None
|
|
|
|
|
|
class TestLocationArchivedValidation:
|
|
"""Tests for POST /actions/location-archived validation."""
|
|
|
|
def test_returns_403_for_recorder(self, user_client, seeded_db):
|
|
"""POST returns 403 for non-admin."""
|
|
row = seeded_db.execute("SELECT id FROM locations LIMIT 1").fetchone()
|
|
location_id = row[0]
|
|
|
|
resp = user_client.post(
|
|
"/actions/location-archived",
|
|
data={
|
|
"location_id": location_id,
|
|
"nonce": "test-nonce-loc-18",
|
|
},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_returns_422_for_nonexistent_location(self, admin_client):
|
|
"""POST returns 422 for non-existent location."""
|
|
fake_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
resp = admin_client.post(
|
|
"/actions/location-archived",
|
|
data={
|
|
"location_id": fake_id,
|
|
"nonce": "test-nonce-loc-19",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
def test_returns_422_for_already_archived(self, admin_client, seeded_db):
|
|
"""POST returns 422 when archiving already archived location."""
|
|
# Create and archive a location
|
|
admin_client.post(
|
|
"/actions/location-created",
|
|
data={"name": "Double Archive Test", "nonce": "test-nonce-loc-20"},
|
|
)
|
|
|
|
row = seeded_db.execute(
|
|
"SELECT id FROM locations WHERE name = 'Double Archive Test'"
|
|
).fetchone()
|
|
location_id = row[0]
|
|
|
|
# Archive once
|
|
admin_client.post(
|
|
"/actions/location-archived",
|
|
data={"location_id": location_id, "nonce": "test-nonce-loc-21"},
|
|
)
|
|
|
|
# Try to archive again
|
|
resp = admin_client.post(
|
|
"/actions/location-archived",
|
|
data={
|
|
"location_id": location_id,
|
|
"nonce": "test-nonce-loc-22",
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|