feat: add reference tables schema and Pydantic models

Implements Step 1.4: Creates migration for species, locations, products,
feed_types, and users tables with full CHECK constraints. Adds Pydantic
models with validation for all reference types including enums for
ProductUnit and UserRole.

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

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-27 19:18:08 +00:00
parent 1e2a0208e6
commit 2e28827683
5 changed files with 968 additions and 0 deletions

View File

@@ -0,0 +1,360 @@
# ABOUTME: Tests for the reference tables migration (0001-reference-tables.sql).
# ABOUTME: Validates that tables are created with correct schema and constraints.
import apsw
import pytest
from animaltrack.db import get_db
from animaltrack.migrations import run_migrations
@pytest.fixture
def migrated_db(tmp_path):
"""Create a database with migrations applied."""
db_path = str(tmp_path / "test.db")
migrations_dir = "migrations"
run_migrations(db_path, migrations_dir, verbose=False)
return get_db(db_path)
class TestMigrationCreatesAllTables:
"""Tests that migration creates all reference tables."""
def test_species_table_exists(self, migrated_db):
"""Migration creates species table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='species'"
).fetchone()
assert result is not None
assert result[0] == "species"
def test_locations_table_exists(self, migrated_db):
"""Migration creates locations table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='locations'"
).fetchone()
assert result is not None
assert result[0] == "locations"
def test_products_table_exists(self, migrated_db):
"""Migration creates products table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='products'"
).fetchone()
assert result is not None
assert result[0] == "products"
def test_feed_types_table_exists(self, migrated_db):
"""Migration creates feed_types table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='feed_types'"
).fetchone()
assert result is not None
assert result[0] == "feed_types"
def test_users_table_exists(self, migrated_db):
"""Migration creates users table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='users'"
).fetchone()
assert result is not None
assert result[0] == "users"
class TestSpeciesTable:
"""Tests for species table schema and constraints."""
def test_insert_valid_species(self, migrated_db):
"""Can insert valid species data."""
migrated_db.execute(
"""
INSERT INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES ('duck', 'Duck', 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT code, name, active FROM species WHERE code='duck'"
).fetchone()
assert result == ("duck", "Duck", 1)
def test_active_check_constraint(self, migrated_db):
"""Species active must be 0 or 1."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES ('invalid', 'Invalid', 2, 1704067200000, 1704067200000)
"""
)
def test_code_is_primary_key(self, migrated_db):
"""Species code is primary key (duplicate rejected)."""
migrated_db.execute(
"""
INSERT INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES ('goose', 'Goose', 1, 1704067200000, 1704067200000)
"""
)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES ('goose', 'Another Goose', 1, 1704067200000, 1704067200000)
"""
)
class TestLocationsTable:
"""Tests for locations table schema and constraints."""
def test_insert_valid_location(self, migrated_db):
"""Can insert valid location data with 26-char ULID."""
migrated_db.execute(
"""
INSERT INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAV', 'Strip 1', 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT id, name FROM locations WHERE name='Strip 1'"
).fetchone()
assert result == ("01ARZ3NDEKTSV4RRFFQ69G5FAV", "Strip 1")
def test_id_length_check_constraint(self, migrated_db):
"""Location ID must be exactly 26 characters."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES ('short', 'Bad Location', 1, 1704067200000, 1704067200000)
"""
)
def test_name_unique_constraint(self, migrated_db):
"""Location name must be unique."""
migrated_db.execute(
"""
INSERT INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAV', 'Strip 1', 1, 1704067200000, 1704067200000)
"""
)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAW', 'Strip 1', 1, 1704067200000, 1704067200000)
"""
)
class TestProductsTable:
"""Tests for products table schema and constraints."""
def test_insert_valid_product_piece(self, migrated_db):
"""Can insert valid product with piece unit."""
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES ('egg.duck', 'Duck Egg', 'piece', 1, 1, 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT code, unit FROM products WHERE code='egg.duck'"
).fetchone()
assert result == ("egg.duck", "piece")
def test_insert_valid_product_kg(self, migrated_db):
"""Can insert valid product with kg unit."""
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES ('meat', 'Meat', 'kg', 1, 1, 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT unit FROM products WHERE code='meat'").fetchone()
assert result[0] == "kg"
def test_unit_check_constraint(self, migrated_db):
"""Product unit must be 'piece' or 'kg'."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES ('bad', 'Bad', 'liter', 1, 1, 1, 1704067200000, 1704067200000)
"""
)
def test_collectable_check_constraint(self, migrated_db):
"""Product collectable must be 0 or 1."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES ('bad', 'Bad', 'piece', 5, 1, 1, 1704067200000, 1704067200000)
"""
)
def test_sellable_check_constraint(self, migrated_db):
"""Product sellable must be 0 or 1."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, active, created_at_utc, updated_at_utc)
VALUES ('bad', 'Bad', 'piece', 1, -1, 1, 1704067200000, 1704067200000)
"""
)
class TestFeedTypesTable:
"""Tests for feed_types table schema and constraints."""
def test_insert_valid_feed_type(self, migrated_db):
"""Can insert valid feed type with protein percentage."""
migrated_db.execute(
"""
INSERT INTO feed_types (code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc)
VALUES ('layer', 'Layer Feed', 20, 16.5, 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT code, default_bag_size_kg, protein_pct FROM feed_types WHERE code='layer'"
).fetchone()
assert result == ("layer", 20, 16.5)
def test_insert_feed_type_without_protein(self, migrated_db):
"""Can insert feed type with NULL protein percentage."""
migrated_db.execute(
"""
INSERT INTO feed_types (code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc)
VALUES ('starter', 'Starter Feed', 25, NULL, 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT protein_pct FROM feed_types WHERE code='starter'"
).fetchone()
assert result[0] is None
def test_bag_size_check_constraint(self, migrated_db):
"""Feed type default_bag_size_kg must be >= 1."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO feed_types (code, name, default_bag_size_kg, protein_pct, active, created_at_utc, updated_at_utc)
VALUES ('bad', 'Bad Feed', 0, NULL, 1, 1704067200000, 1704067200000)
"""
)
class TestUsersTable:
"""Tests for users table schema and constraints."""
def test_insert_valid_admin_user(self, migrated_db):
"""Can insert valid admin user."""
migrated_db.execute(
"""
INSERT INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES ('ppetru', 'admin', 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT username, role FROM users WHERE username='ppetru'"
).fetchone()
assert result == ("ppetru", "admin")
def test_insert_valid_recorder_user(self, migrated_db):
"""Can insert valid recorder user."""
migrated_db.execute(
"""
INSERT INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES ('ines', 'recorder', 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT role FROM users WHERE username='ines'").fetchone()
assert result[0] == "recorder"
def test_role_check_constraint(self, migrated_db):
"""User role must be 'admin' or 'recorder'."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES ('bad', 'superuser', 1, 1704067200000, 1704067200000)
"""
)
def test_username_is_primary_key(self, migrated_db):
"""Username is primary key (duplicate rejected)."""
migrated_db.execute(
"""
INSERT INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES ('guest', 'recorder', 1, 1704067200000, 1704067200000)
"""
)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO users (username, role, active, created_at_utc, updated_at_utc)
VALUES ('guest', 'admin', 1, 1704067200000, 1704067200000)
"""
)
class TestDefaultValues:
"""Tests for default values in tables."""
def test_species_active_defaults_to_1(self, migrated_db):
"""Species active defaults to 1 when not specified."""
migrated_db.execute(
"""
INSERT INTO species (code, name, created_at_utc, updated_at_utc)
VALUES ('sheep', 'Sheep', 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT active FROM species WHERE code='sheep'").fetchone()
assert result[0] == 1
def test_locations_active_defaults_to_1(self, migrated_db):
"""Locations active defaults to 1 when not specified."""
migrated_db.execute(
"""
INSERT INTO locations (id, name, created_at_utc, updated_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAV', 'Nursery 1', 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT active FROM locations WHERE name='Nursery 1'"
).fetchone()
assert result[0] == 1
def test_products_active_defaults_to_1(self, migrated_db):
"""Products active defaults to 1 when not specified."""
migrated_db.execute(
"""
INSERT INTO products (code, name, unit, collectable, sellable, created_at_utc, updated_at_utc)
VALUES ('offal', 'Offal', 'kg', 1, 1, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT active FROM products WHERE code='offal'").fetchone()
assert result[0] == 1
def test_feed_types_active_defaults_to_1(self, migrated_db):
"""Feed types active defaults to 1 when not specified."""
migrated_db.execute(
"""
INSERT INTO feed_types (code, name, default_bag_size_kg, created_at_utc, updated_at_utc)
VALUES ('grower', 'Grower Feed', 20, 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT active FROM feed_types WHERE code='grower'").fetchone()
assert result[0] == 1
def test_users_active_defaults_to_1(self, migrated_db):
"""Users active defaults to 1 when not specified."""
migrated_db.execute(
"""
INSERT INTO users (username, role, created_at_utc, updated_at_utc)
VALUES ('newuser', 'recorder', 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute("SELECT active FROM users WHERE username='newuser'").fetchone()
assert result[0] == 1

View File

@@ -0,0 +1,405 @@
# ABOUTME: Tests for reference table Pydantic models.
# ABOUTME: Validates Species, Location, Product, FeedType, and User models.
import pytest
from pydantic import ValidationError
from animaltrack.models.reference import (
FeedType,
Location,
Product,
ProductUnit,
Species,
User,
UserRole,
)
class TestSpecies:
"""Tests for the Species model."""
def test_valid_species(self):
"""Species with all required fields validates successfully."""
species = Species(
code="duck",
name="Duck",
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert species.code == "duck"
assert species.name == "Duck"
assert species.active is True
assert species.created_at_utc == 1704067200000
assert species.updated_at_utc == 1704067200000
def test_species_active_defaults_to_true(self):
"""Species active field defaults to True."""
species = Species(
code="goose",
name="Goose",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert species.active is True
def test_species_requires_code(self):
"""Species without code raises ValidationError."""
with pytest.raises(ValidationError):
Species(
name="Duck",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
def test_species_requires_name(self):
"""Species without name raises ValidationError."""
with pytest.raises(ValidationError):
Species(
code="duck",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
def test_species_requires_timestamps(self):
"""Species without timestamps raises ValidationError."""
with pytest.raises(ValidationError):
Species(code="duck", name="Duck")
class TestLocation:
"""Tests for the Location model."""
def test_valid_location(self):
"""Location with valid 26-char ULID validates successfully."""
location = Location(
id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
name="Strip 1",
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert location.id == "01ARZ3NDEKTSV4RRFFQ69G5FAV"
assert location.name == "Strip 1"
def test_location_id_must_be_26_chars(self):
"""Location with ID not exactly 26 chars raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
Location(
id="short",
name="Strip 1",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert "26 characters" in str(exc_info.value)
def test_location_id_too_long(self):
"""Location with ID longer than 26 chars raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
Location(
id="01ARZ3NDEKTSV4RRFFQ69G5FAVX", # 27 chars
name="Strip 1",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert "26 characters" in str(exc_info.value)
def test_location_active_defaults_to_true(self):
"""Location active field defaults to True."""
location = Location(
id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
name="Strip 1",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert location.active is True
def test_location_requires_name(self):
"""Location without name raises ValidationError."""
with pytest.raises(ValidationError):
Location(
id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
class TestProductUnit:
"""Tests for the ProductUnit enum."""
def test_piece_value(self):
"""ProductUnit.PIECE has value 'piece'."""
assert ProductUnit.PIECE.value == "piece"
def test_kg_value(self):
"""ProductUnit.KG has value 'kg'."""
assert ProductUnit.KG.value == "kg"
class TestProduct:
"""Tests for the Product model."""
def test_valid_product(self):
"""Product with all required fields validates successfully."""
product = Product(
code="egg.duck",
name="Duck Egg",
unit=ProductUnit.PIECE,
collectable=True,
sellable=True,
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert product.code == "egg.duck"
assert product.name == "Duck Egg"
assert product.unit == ProductUnit.PIECE
assert product.collectable is True
assert product.sellable is True
def test_product_unit_kg(self):
"""Product with kg unit validates successfully."""
product = Product(
code="meat",
name="Meat",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert product.unit == ProductUnit.KG
def test_product_unit_from_string(self):
"""Product accepts unit as string value."""
product = Product(
code="meat",
name="Meat",
unit="kg",
collectable=True,
sellable=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert product.unit == ProductUnit.KG
def test_product_invalid_unit(self):
"""Product with invalid unit raises ValidationError."""
with pytest.raises(ValidationError):
Product(
code="meat",
name="Meat",
unit="invalid",
collectable=True,
sellable=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
def test_product_active_defaults_to_true(self):
"""Product active field defaults to True."""
product = Product(
code="meat",
name="Meat",
unit=ProductUnit.KG,
collectable=True,
sellable=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert product.active is True
def test_product_requires_collectable(self):
"""Product without collectable raises ValidationError."""
with pytest.raises(ValidationError):
Product(
code="meat",
name="Meat",
unit=ProductUnit.KG,
sellable=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
class TestFeedType:
"""Tests for the FeedType model."""
def test_valid_feed_type(self):
"""FeedType with all required fields validates successfully."""
feed_type = FeedType(
code="layer",
name="Layer Feed",
default_bag_size_kg=20,
protein_pct=16.5,
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert feed_type.code == "layer"
assert feed_type.name == "Layer Feed"
assert feed_type.default_bag_size_kg == 20
assert feed_type.protein_pct == 16.5
def test_feed_type_protein_pct_optional(self):
"""FeedType protein_pct is optional (can be None)."""
feed_type = FeedType(
code="starter",
name="Starter Feed",
default_bag_size_kg=20,
protein_pct=None,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert feed_type.protein_pct is None
def test_feed_type_default_bag_size_must_be_positive(self):
"""FeedType default_bag_size_kg must be >= 1."""
with pytest.raises(ValidationError) as exc_info:
FeedType(
code="layer",
name="Layer Feed",
default_bag_size_kg=0,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert "greater than or equal to 1" in str(exc_info.value)
def test_feed_type_default_bag_size_negative(self):
"""FeedType negative default_bag_size_kg raises ValidationError."""
with pytest.raises(ValidationError):
FeedType(
code="layer",
name="Layer Feed",
default_bag_size_kg=-5,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
def test_feed_type_active_defaults_to_true(self):
"""FeedType active field defaults to True."""
feed_type = FeedType(
code="grower",
name="Grower Feed",
default_bag_size_kg=25,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert feed_type.active is True
class TestUserRole:
"""Tests for the UserRole enum."""
def test_admin_value(self):
"""UserRole.ADMIN has value 'admin'."""
assert UserRole.ADMIN.value == "admin"
def test_recorder_value(self):
"""UserRole.RECORDER has value 'recorder'."""
assert UserRole.RECORDER.value == "recorder"
class TestUser:
"""Tests for the User model."""
def test_valid_user_admin(self):
"""User with admin role validates successfully."""
user = User(
username="ppetru",
role=UserRole.ADMIN,
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert user.username == "ppetru"
assert user.role == UserRole.ADMIN
def test_valid_user_recorder(self):
"""User with recorder role validates successfully."""
user = User(
username="ines",
role=UserRole.RECORDER,
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert user.role == UserRole.RECORDER
def test_user_role_from_string(self):
"""User accepts role as string value."""
user = User(
username="guest",
role="recorder",
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert user.role == UserRole.RECORDER
def test_user_invalid_role(self):
"""User with invalid role raises ValidationError."""
with pytest.raises(ValidationError):
User(
username="bad",
role="superuser",
active=True,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
def test_user_active_defaults_to_true(self):
"""User active field defaults to True."""
user = User(
username="test",
role=UserRole.RECORDER,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
assert user.active is True
def test_user_requires_username(self):
"""User without username raises ValidationError."""
with pytest.raises(ValidationError):
User(
role=UserRole.ADMIN,
created_at_utc=1704067200000,
updated_at_utc=1704067200000,
)
class TestTimestampValidation:
"""Tests for timestamp validation across all models."""
def test_species_negative_timestamp_rejected(self):
"""Species with negative timestamp raises ValidationError."""
with pytest.raises(ValidationError):
Species(
code="duck",
name="Duck",
created_at_utc=-1,
updated_at_utc=1704067200000,
)
def test_location_negative_timestamp_rejected(self):
"""Location with negative timestamp raises ValidationError."""
with pytest.raises(ValidationError):
Location(
id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
name="Strip 1",
created_at_utc=1704067200000,
updated_at_utc=-1,
)
def test_product_zero_timestamp_allowed(self):
"""Product with zero timestamp (Unix epoch) is allowed."""
product = Product(
code="egg",
name="Egg",
unit=ProductUnit.PIECE,
collectable=True,
sellable=True,
created_at_utc=0,
updated_at_utc=0,
)
assert product.created_at_utc == 0