Alo/animaltrack
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: implement user defaults persistence (Step 9.3)

Browse Source 
Add user_defaults table and repository for persisting form defaults
across sessions. Feed and egg forms now load/save user preferences.

Changes:
- Add migration 0009-user-defaults.sql with table schema
- Add UserDefault model and UserDefaultsRepository
- Integrate defaults into feed route (location, feed_type, amount)
- Integrate defaults into egg route (location)
- Add repository unit tests and route integration tests

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Petru
2025-12-31 14:35:27 +00:00
parent d89c46ab51
commit 719d1e6ce7
9 changed files with 566 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

158
tests/test_repositories_user_defaults.py Normal file
View File

@@ -0,0 +1,158 @@
# ABOUTME: Tests for UserDefaultsRepository.
# ABOUTME: Validates CRUD operations for user form defaults.
import time
import pytest
from animaltrack.models.reference import UserDefault
from animaltrack.repositories.user_defaults import UserDefaultsRepository
@pytest.fixture
def now_utc():
"""Current time in milliseconds since epoch."""
return int(time.time() * 1000)
class TestUserDefaultsRepository:
"""Tests for UserDefaultsRepository."""
def test_get_returns_none_for_missing(self, seeded_db):
"""get returns None when no defaults exist."""
repo = UserDefaultsRepository(seeded_db)
result = repo.get("ppetru", "collect_egg")
assert result is None
def test_upsert_creates_new_record(self, seeded_db, now_utc):
"""upsert creates a new defaults record."""
repo = UserDefaultsRepository(seeded_db)
defaults = UserDefault(
username="ppetru",
action="collect_egg",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
updated_at_utc=now_utc,
)
repo.upsert(defaults)
result = repo.get("ppetru", "collect_egg")
assert result is not None
assert result.username == "ppetru"
assert result.action == "collect_egg"
assert result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5FAV"
def test_upsert_updates_existing_record(self, seeded_db, now_utc):
"""upsert updates an existing defaults record."""
repo = UserDefaultsRepository(seeded_db)
defaults = UserDefault(
username="ppetru",
action="feed_given",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
feed_type_code="layer",
amount_kg=20,
updated_at_utc=now_utc,
)
repo.upsert(defaults)
updated = UserDefault(
username="ppetru",
action="feed_given",
location_id="01ARZ3NDEKTSV4RRFFQ69G5ABC",
feed_type_code="grower",
amount_kg=25,
updated_at_utc=now_utc + 1000,
)
repo.upsert(updated)
result = repo.get("ppetru", "feed_given")
assert result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5ABC"
assert result.feed_type_code == "grower"
assert result.amount_kg == 25
def test_get_returns_all_fields(self, seeded_db, now_utc):
"""get returns all stored fields correctly."""
repo = UserDefaultsRepository(seeded_db)
defaults = UserDefault(
username="ppetru",
action="feed_given",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species="duck",
animal_filter="location:strip1",
feed_type_code="layer",
amount_kg=20,
bag_size_kg=25,
updated_at_utc=now_utc,
)
repo.upsert(defaults)
result = repo.get("ppetru", "feed_given")
assert result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5FAV"
assert result.species == "duck"
assert result.animal_filter == "location:strip1"
assert result.feed_type_code == "layer"
assert result.amount_kg == 20
assert result.bag_size_kg == 25
def test_different_actions_are_independent(self, seeded_db, now_utc):
"""Different actions for same user are stored independently."""
repo = UserDefaultsRepository(seeded_db)
egg_defaults = UserDefault(
username="ppetru",
action="collect_egg",
location_id="01ARZ3NDEKTSV4RRFFQ69G5EGG",
updated_at_utc=now_utc,
)
feed_defaults = UserDefault(
username="ppetru",
action="feed_given",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FED",
updated_at_utc=now_utc,
)
repo.upsert(egg_defaults)
repo.upsert(feed_defaults)
egg_result = repo.get("ppetru", "collect_egg")
feed_result = repo.get("ppetru", "feed_given")
assert egg_result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5EGG"
assert feed_result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5FED"
def test_different_users_are_independent(self, seeded_db, now_utc):
"""Different users have independent defaults."""
repo = UserDefaultsRepository(seeded_db)
ppetru_defaults = UserDefault(
username="ppetru",
action="collect_egg",
location_id="01ARZ3NDEKTSV4RRFFQ69G5PPP",
updated_at_utc=now_utc,
)
ines_defaults = UserDefault(
username="ines",
action="collect_egg",
location_id="01ARZ3NDEKTSV4RRFFQ69G5III",
updated_at_utc=now_utc,
)
repo.upsert(ppetru_defaults)
repo.upsert(ines_defaults)
ppetru_result = repo.get("ppetru", "collect_egg")
ines_result = repo.get("ines", "collect_egg")
assert ppetru_result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5PPP"
assert ines_result.location_id == "01ARZ3NDEKTSV4RRFFQ69G5III"
def test_null_fields_preserved(self, seeded_db, now_utc):
"""Null fields are stored and retrieved correctly."""
repo = UserDefaultsRepository(seeded_db)
defaults = UserDefault(
username="ppetru",
action="collect_egg",
location_id=None,
species=None,
updated_at_utc=now_utc,
)
repo.upsert(defaults)
result = repo.get("ppetru", "collect_egg")
assert result.location_id is None
assert result.species is None

223
tests/test_user_defaults_integration.py Normal file
View File

@@ -0,0 +1,223 @@
# ABOUTME: Integration tests for user defaults feature.
# ABOUTME: Verifies that form defaults are saved and loaded correctly.
import os
import time
import pytest
from starlette.testclient import TestClient
from animaltrack.events.payloads import FeedPurchasedPayload
from animaltrack.events.store import EventStore
from animaltrack.models.reference import UserDefault
from animaltrack.projections import ProjectionRegistry
from animaltrack.projections.feed import FeedInventoryProjection
from animaltrack.repositories.user_defaults import UserDefaultsRepository
from animaltrack.services.feed import FeedService
def make_test_settings(
csrf_secret: str = "test-secret",
trusted_proxy_ips: str = "127.0.0.1",
dev_mode: bool = False, # Disable dev_mode to test real auth
):
"""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 with real auth enabled."""
from animaltrack.web.app import create_app
settings = make_test_settings(trusted_proxy_ips="testclient", dev_mode=False)
app, rt = create_app(settings=settings, db=seeded_db)
return TestClient(app, raise_server_exceptions=True)
@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 feed_purchase_in_db(seeded_db):
"""Create a feed purchase so give_feed can work."""
event_store = EventStore(seeded_db)
registry = ProjectionRegistry()
registry.register(FeedInventoryProjection(seeded_db))
feed_service = FeedService(seeded_db, event_store, registry)
payload = FeedPurchasedPayload(
feed_type_code="layer",
bag_size_kg=20,
bags_count=5,
bag_price_cents=2400,
)
ts_utc = int(time.time() * 1000) - 86400000
feed_service.purchase_feed(payload, ts_utc, "ppetru")
return payload
def make_csrf_headers(csrf_token: str = "test-csrf-token"):
"""Make headers with CSRF token for POST requests."""
return {
"X-CSRF-Token": csrf_token,
"Origin": "http://testserver", # Match TestClient's default host
}
class TestFeedUserDefaults:
"""Tests for feed form user defaults."""
def test_defaults_saved_on_successful_give(
self, client, seeded_db, location_strip1_id, feed_purchase_in_db
):
"""Successful feed-given saves user defaults."""
csrf_token = "test-csrf-token"
response = client.post(
"/actions/feed-given",
data={
"location_id": location_strip1_id,
"feed_type_code": "layer",
"amount_kg": "15",
},
headers={
"X-Oidc-Username": "ppetru",
**make_csrf_headers(csrf_token),
},
cookies={"csrf_token": csrf_token},
)
assert response.status_code == 200
# Verify defaults were saved
defaults = UserDefaultsRepository(seeded_db).get("ppetru", "feed_given")
assert defaults is not None
assert defaults.location_id == location_strip1_id
assert defaults.feed_type_code == "layer"
assert defaults.amount_kg == 15
def test_defaults_loaded_on_feed_page(self, client, seeded_db, location_strip1_id):
"""GET /feed loads saved user defaults."""
# First set some defaults
now_utc = int(time.time() * 1000)
UserDefaultsRepository(seeded_db).upsert(
UserDefault(
username="ppetru",
action="feed_given",
location_id=location_strip1_id,
feed_type_code="grower",
amount_kg=25,
updated_at_utc=now_utc,
)
)
# Load the feed page
response = client.get(
"/feed",
headers={"X-Oidc-Username": "ppetru"},
)
assert response.status_code == 200
# Check that the form has pre-selected values
content = response.text
assert f'value="{location_strip1_id}"' in content or "selected" in content
assert "grower" in content
def test_no_defaults_for_unknown_user(
self, client, seeded_db, location_strip1_id, feed_purchase_in_db
):
"""Unknown users are rejected by auth middleware."""
csrf_token = "test-csrf-token"
response = client.post(
"/actions/feed-given",
data={
"location_id": location_strip1_id,
"feed_type_code": "layer",
"amount_kg": "10",
},
headers={
"X-Oidc-Username": "unknown_user",
**make_csrf_headers(csrf_token),
},
cookies={"csrf_token": csrf_token},
)
# Unknown user is rejected by auth middleware
assert response.status_code == 401
# Verify no defaults were saved
defaults = UserDefaultsRepository(seeded_db).get("unknown_user", "feed_given")
assert defaults is None
class TestEggUserDefaults:
"""Tests for egg form user defaults."""
def test_defaults_loaded_on_egg_page(self, client, seeded_db, location_strip1_id):
"""GET / loads saved user defaults for egg collection."""
# First set some defaults
now_utc = int(time.time() * 1000)
UserDefaultsRepository(seeded_db).upsert(
UserDefault(
username="ppetru",
action="collect_egg",
location_id=location_strip1_id,
updated_at_utc=now_utc,
)
)
# Load the egg page
response = client.get(
"/",
headers={"X-Oidc-Username": "ppetru"},
)
assert response.status_code == 200
# Check that the form has pre-selected location
content = response.text
assert location_strip1_id in content
def test_query_param_overrides_defaults(
self, client, seeded_db, location_strip1_id, location_strip2_id
):
"""Query param location_id overrides saved defaults."""
# Set defaults to Strip 1
now_utc = int(time.time() * 1000)
UserDefaultsRepository(seeded_db).upsert(
UserDefault(
username="ppetru",
action="collect_egg",
location_id=location_strip1_id,
updated_at_utc=now_utc,
)
)
# Load the egg page with Strip 2 in query params
response = client.get(
f"/?location_id={location_strip2_id}",
headers={"X-Oidc-Username": "ppetru"},
)
assert response.status_code == 200
# Query param should take precedence - Strip 2 should be selected
content = response.text
assert location_strip2_id in content