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

feat: add animal registry schema and models

Browse Source 
Add database tables for animal tracking:
- animal_registry: main snapshot table with all animal attributes
- live_animals_by_location: denormalized view for fast roster queries
- animal_aliases: merge tracking for when animals are discovered to be same

Includes Pydantic models and comprehensive tests for all constraints.

🤖 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-28 18:59:24 +00:00
parent d8259f4371
commit 739b7bfe32
4 changed files with 901 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
migrations/0003-animal-registry-schema.sql Normal file
View File

@@ -0,0 +1,58 @@
-- ABOUTME: Creates animal tracking tables for the registry system.
-- ABOUTME: Includes animal_registry, live_animals_by_location, and animal_aliases.
-- Main snapshot table for all animals (current state)
CREATE TABLE animal_registry (
animal_id TEXT PRIMARY KEY CHECK(length(animal_id) = 26),
species_code TEXT NOT NULL REFERENCES species(code),
identified INTEGER NOT NULL DEFAULT 0 CHECK(identified IN (0, 1)),
nickname TEXT,
sex TEXT NOT NULL CHECK(sex IN ('male', 'female', 'unknown')),
repro_status TEXT NOT NULL CHECK(repro_status IN ('intact', 'wether', 'spayed', 'unknown')),
life_stage TEXT NOT NULL CHECK(life_stage IN ('hatchling', 'juvenile', 'subadult', 'adult')),
status TEXT NOT NULL CHECK(status IN ('alive', 'dead', 'harvested', 'sold', 'merged_into')),
location_id TEXT NOT NULL REFERENCES locations(id),
origin TEXT NOT NULL CHECK(origin IN ('hatched', 'purchased', 'rescued', 'unknown')),
born_or_hatched_at INTEGER,
acquired_at INTEGER,
first_seen_utc INTEGER NOT NULL,
last_event_utc INTEGER NOT NULL
);
-- Unique nickname only for active animals (allows nulls, allows reuse for dead/merged)
CREATE UNIQUE INDEX idx_ar_nickname_active
ON animal_registry(nickname)
WHERE nickname IS NOT NULL
AND status NOT IN ('dead', 'harvested', 'sold', 'merged_into');
CREATE INDEX idx_ar_location ON animal_registry(location_id);
CREATE INDEX idx_ar_filter ON animal_registry(species_code, sex, life_stage, identified);
CREATE INDEX idx_ar_status ON animal_registry(status);
CREATE INDEX idx_ar_last_event ON animal_registry(last_event_utc);
-- Denormalized view for fast roster queries (only alive animals)
CREATE TABLE live_animals_by_location (
animal_id TEXT PRIMARY KEY CHECK(length(animal_id) = 26),
location_id TEXT NOT NULL REFERENCES locations(id),
species_code TEXT NOT NULL REFERENCES species(code),
identified INTEGER NOT NULL DEFAULT 0 CHECK(identified IN (0, 1)),
nickname TEXT,
sex TEXT NOT NULL CHECK(sex IN ('male', 'female', 'unknown')),
repro_status TEXT NOT NULL CHECK(repro_status IN ('intact', 'wether', 'spayed', 'unknown')),
life_stage TEXT NOT NULL CHECK(life_stage IN ('hatchling', 'juvenile', 'subadult', 'adult')),
first_seen_utc INTEGER NOT NULL,
last_move_utc INTEGER,
tags TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(tags))
);
CREATE INDEX idx_labl_location ON live_animals_by_location(location_id);
CREATE INDEX idx_labl_filter ON live_animals_by_location(location_id, species_code, sex, life_stage, identified);
-- Tracks when animals are discovered to be the same individual (merge)
CREATE TABLE animal_aliases (
alias_animal_id TEXT PRIMARY KEY CHECK(length(alias_animal_id) = 26),
survivor_animal_id TEXT NOT NULL CHECK(length(survivor_animal_id) = 26),
merged_at_utc INTEGER NOT NULL
);
CREATE INDEX idx_aa_survivor ON animal_aliases(survivor_animal_id);

61
src/animaltrack/models/animals.py Normal file
View File

@@ -0,0 +1,61 @@
# ABOUTME: Pydantic models for animal tracking tables.
# ABOUTME: Includes Animal, LiveAnimal, and AnimalAlias for registry system.
from pydantic import BaseModel, Field
from animaltrack.events.enums import AnimalStatus, LifeStage, Origin, ReproStatus, Sex
class Animal(BaseModel):
"""Full animal record from animal_registry table.
Represents the current state of an animal including all attributes,
location, and lifecycle status.
"""
animal_id: str = Field(..., min_length=26, max_length=26)
species_code: str
identified: bool = False
nickname: str | None = None
sex: Sex
repro_status: ReproStatus
life_stage: LifeStage
status: AnimalStatus
location_id: str = Field(..., min_length=26, max_length=26)
origin: Origin
born_or_hatched_at: int | None = None
acquired_at: int | None = None
first_seen_utc: int
last_event_utc: int
class LiveAnimal(BaseModel):
"""Animal record from live_animals_by_location table.
Denormalized view of active animals for fast roster queries.
Only includes animals with status='alive'.
"""
animal_id: str = Field(..., min_length=26, max_length=26)
location_id: str = Field(..., min_length=26, max_length=26)
species_code: str
identified: bool = False
nickname: str | None = None
sex: Sex
repro_status: ReproStatus
life_stage: LifeStage
first_seen_utc: int
last_move_utc: int | None = None
tags: list[str] = []
class AnimalAlias(BaseModel):
"""Merge tracking record from animal_aliases table.
When two animals are discovered to be the same individual,
one becomes an alias pointing to the survivor.
"""
alias_animal_id: str = Field(..., min_length=26, max_length=26)
survivor_animal_id: str = Field(..., min_length=26, max_length=26)
merged_at_utc: int

532
tests/test_migration_animal_registry.py Normal file
View File

@@ -0,0 +1,532 @@
# ABOUTME: Tests for the animal registry migration (0003-animal-registry-schema.sql).
# ABOUTME: Validates tables, constraints, and indexes for animal tracking.
import apsw
import pytest
# Helper to insert prerequisite data for FK tests
def _insert_species(db, code="duck"):
"""Insert a species for FK testing."""
db.execute(
"""
INSERT INTO species (code, name, active, created_at_utc, updated_at_utc)
VALUES (?, 'Test Species', 1, 1704067200000, 1704067200000)
""",
(code,),
)
def _insert_location(db, location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"):
"""Insert a location for FK testing."""
db.execute(
"""
INSERT INTO locations (id, name, active, created_at_utc, updated_at_utc)
VALUES (?, 'Test Location', 1, 1704067200000, 1704067200000)
""",
(location_id,),
)
class TestMigrationCreatesAllTables:
"""Tests that migration creates all animal tracking tables."""
def test_animal_registry_table_exists(self, migrated_db):
"""Migration creates animal_registry table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='animal_registry'"
).fetchone()
assert result is not None
assert result[0] == "animal_registry"
def test_live_animals_by_location_table_exists(self, migrated_db):
"""Migration creates live_animals_by_location table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='live_animals_by_location'"
).fetchone()
assert result is not None
assert result[0] == "live_animals_by_location"
def test_animal_aliases_table_exists(self, migrated_db):
"""Migration creates animal_aliases table."""
result = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='animal_aliases'"
).fetchone()
assert result is not None
assert result[0] == "animal_aliases"
class TestAnimalRegistryTable:
"""Tests for animal_registry table schema and constraints."""
def test_insert_valid_animal(self, migrated_db):
"""Can insert valid animal data."""
_insert_species(migrated_db)
_insert_location(migrated_db)
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, identified, nickname, sex, repro_status,
life_stage, status, location_id, origin, first_seen_utc, last_event_utc)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"01ARZ3NDEKTSV4RRFFQ69G5FAA", # animal_id
"duck", # species_code
1, # identified
"Daffy", # nickname
"male", # sex
"intact", # repro_status
"adult", # life_stage
"alive", # status
"01ARZ3NDEKTSV4RRFFQ69G5FAV", # location_id
"hatched", # origin
1704067200000, # first_seen_utc
1704067200000, # last_event_utc
),
)
result = migrated_db.execute(
"SELECT animal_id, nickname, sex, status FROM animal_registry"
).fetchone()
assert result == ("01ARZ3NDEKTSV4RRFFQ69G5FAA", "Daffy", "male", "alive")
def test_animal_id_length_check(self, migrated_db):
"""Animal ID must be exactly 26 characters."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('short', 'duck', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_sex_check_constraint(self, migrated_db):
"""Sex must be male, female, or unknown."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'invalid', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_repro_status_check_constraint(self, migrated_db):
"""Repro status must be intact, wether, spayed, or unknown."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'invalid', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_life_stage_check_constraint(self, migrated_db):
"""Life stage must be hatchling, juvenile, subadult, or adult."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'invalid', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_status_check_constraint(self, migrated_db):
"""Status must be alive, dead, harvested, sold, or merged_into."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'adult', 'invalid',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_origin_check_constraint(self, migrated_db):
"""Origin must be hatched, purchased, rescued, or unknown."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'invalid', 1704067200000, 1704067200000)
"""
)
def test_species_fk_constraint(self, migrated_db):
"""Species code must reference existing species."""
_insert_location(migrated_db)
# Do NOT insert species
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'nonexistent', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_location_fk_constraint(self, migrated_db):
"""Location ID must reference existing location."""
_insert_species(migrated_db)
# Do NOT insert location
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_identified_defaults_to_0(self, migrated_db):
"""Identified defaults to 0."""
_insert_species(migrated_db)
_insert_location(migrated_db)
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT identified FROM animal_registry WHERE animal_id='01ARZ3NDEKTSV4RRFFQ69G5FAA'"
).fetchone()
assert result[0] == 0
class TestUniqueNicknameIndex:
"""Tests for the unique nickname constraint on active animals."""
def test_null_nickname_allowed_multiple_times(self, migrated_db):
"""Multiple active animals can have null nickname."""
_insert_species(migrated_db)
_insert_location(migrated_db)
# First animal with null nickname
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
# Second animal with null nickname - should succeed
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAB', 'duck', 'female', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
count = migrated_db.execute(
"SELECT COUNT(*) FROM animal_registry WHERE nickname IS NULL"
).fetchone()[0]
assert count == 2
def test_duplicate_nickname_rejected_for_active_animals(self, migrated_db):
"""Two active animals cannot have the same nickname."""
_insert_species(migrated_db)
_insert_location(migrated_db)
# First animal with nickname
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, nickname, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'Daffy', 'male', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
# Second active animal with same nickname - should fail
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, nickname, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAB', 'duck', 'Daffy', 'female', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
def test_dead_animal_nickname_can_be_reused(self, migrated_db):
"""A dead animal's nickname can be reused by a new active animal."""
_insert_species(migrated_db)
_insert_location(migrated_db)
# First animal with nickname - dead
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, nickname, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'duck', 'Daffy', 'male', 'intact', 'adult', 'dead',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
# Second active animal with same nickname - should succeed
migrated_db.execute(
"""
INSERT INTO animal_registry
(animal_id, species_code, nickname, sex, repro_status, life_stage, status,
location_id, origin, first_seen_utc, last_event_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAB', 'duck', 'Daffy', 'female', 'intact', 'adult', 'alive',
'01ARZ3NDEKTSV4RRFFQ69G5FAV', 'hatched', 1704067200000, 1704067200000)
"""
)
count = migrated_db.execute(
"SELECT COUNT(*) FROM animal_registry WHERE nickname='Daffy'"
).fetchone()[0]
assert count == 2
class TestLiveAnimalsByLocationTable:
"""Tests for live_animals_by_location table schema and constraints."""
def test_insert_valid_live_animal(self, migrated_db):
"""Can insert valid live animal data."""
_insert_species(migrated_db)
_insert_location(migrated_db)
migrated_db.execute(
"""
INSERT INTO live_animals_by_location
(animal_id, location_id, species_code, identified, nickname, sex, repro_status,
life_stage, first_seen_utc, last_move_utc, tags)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
"01ARZ3NDEKTSV4RRFFQ69G5FAA", # animal_id
"01ARZ3NDEKTSV4RRFFQ69G5FAV", # location_id
"duck", # species_code
1, # identified
"Daffy", # nickname
"male", # sex
"intact", # repro_status
"adult", # life_stage
1704067200000, # first_seen_utc
1704067200000, # last_move_utc
'["friendly", "leader"]', # tags
),
)
result = migrated_db.execute(
"SELECT animal_id, nickname, tags FROM live_animals_by_location"
).fetchone()
assert result == ("01ARZ3NDEKTSV4RRFFQ69G5FAA", "Daffy", '["friendly", "leader"]')
def test_animal_id_length_check(self, migrated_db):
"""Animal ID must be exactly 26 characters."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO live_animals_by_location
(animal_id, location_id, species_code, sex, repro_status, life_stage, first_seen_utc)
VALUES ('short', '01ARZ3NDEKTSV4RRFFQ69G5FAV', 'duck', 'male', 'intact', 'adult', 1704067200000)
"""
)
def test_tags_must_be_valid_json(self, migrated_db):
"""Tags must be valid JSON."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO live_animals_by_location
(animal_id, location_id, species_code, sex, repro_status, life_stage, first_seen_utc, tags)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', '01ARZ3NDEKTSV4RRFFQ69G5FAV',
'duck', 'male', 'intact', 'adult', 1704067200000, 'not valid json')
"""
)
def test_tags_defaults_to_empty_array(self, migrated_db):
"""Tags defaults to empty JSON array."""
_insert_species(migrated_db)
_insert_location(migrated_db)
migrated_db.execute(
"""
INSERT INTO live_animals_by_location
(animal_id, location_id, species_code, sex, repro_status, life_stage, first_seen_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', '01ARZ3NDEKTSV4RRFFQ69G5FAV',
'duck', 'male', 'intact', 'adult', 1704067200000)
"""
)
result = migrated_db.execute(
"SELECT tags FROM live_animals_by_location WHERE animal_id='01ARZ3NDEKTSV4RRFFQ69G5FAA'"
).fetchone()
assert result[0] == "[]"
def test_sex_check_constraint(self, migrated_db):
"""Sex must be male, female, or unknown."""
_insert_species(migrated_db)
_insert_location(migrated_db)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO live_animals_by_location
(animal_id, location_id, species_code, sex, repro_status, life_stage, first_seen_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', '01ARZ3NDEKTSV4RRFFQ69G5FAV',
'duck', 'invalid', 'intact', 'adult', 1704067200000)
"""
)
class TestAnimalAliasesTable:
"""Tests for animal_aliases table schema and constraints."""
def test_insert_valid_alias(self, migrated_db):
"""Can insert valid alias data."""
migrated_db.execute(
"""
INSERT INTO animal_aliases (alias_animal_id, survivor_animal_id, merged_at_utc)
VALUES (?, ?, ?)
""",
(
"01ARZ3NDEKTSV4RRFFQ69G5FAA", # alias (merged away)
"01ARZ3NDEKTSV4RRFFQ69G5FAB", # survivor
1704067200000,
),
)
result = migrated_db.execute(
"SELECT alias_animal_id, survivor_animal_id FROM animal_aliases"
).fetchone()
assert result == ("01ARZ3NDEKTSV4RRFFQ69G5FAA", "01ARZ3NDEKTSV4RRFFQ69G5FAB")
def test_alias_animal_id_length_check(self, migrated_db):
"""Alias animal ID must be exactly 26 characters."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_aliases (alias_animal_id, survivor_animal_id, merged_at_utc)
VALUES ('short', '01ARZ3NDEKTSV4RRFFQ69G5FAB', 1704067200000)
"""
)
def test_survivor_animal_id_length_check(self, migrated_db):
"""Survivor animal ID must be exactly 26 characters."""
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_aliases (alias_animal_id, survivor_animal_id, merged_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', 'short', 1704067200000)
"""
)
def test_alias_is_primary_key(self, migrated_db):
"""Alias animal ID is primary key (duplicate rejected)."""
migrated_db.execute(
"""
INSERT INTO animal_aliases (alias_animal_id, survivor_animal_id, merged_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', '01ARZ3NDEKTSV4RRFFQ69G5FAB', 1704067200000)
"""
)
with pytest.raises(apsw.ConstraintError):
migrated_db.execute(
"""
INSERT INTO animal_aliases (alias_animal_id, survivor_animal_id, merged_at_utc)
VALUES ('01ARZ3NDEKTSV4RRFFQ69G5FAA', '01ARZ3NDEKTSV4RRFFQ69G5FAC', 1704067200000)
"""
)
class TestIndexes:
"""Tests that indexes are created correctly."""
def test_animal_registry_indexes_exist(self, migrated_db):
"""Animal registry table has required indexes."""
indexes = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='animal_registry'"
).fetchall()
index_names = {row[0] for row in indexes}
assert "idx_ar_nickname_active" in index_names
assert "idx_ar_location" in index_names
assert "idx_ar_filter" in index_names
assert "idx_ar_status" in index_names
assert "idx_ar_last_event" in index_names
def test_live_animals_by_location_indexes_exist(self, migrated_db):
"""Live animals by location table has required indexes."""
indexes = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='live_animals_by_location'"
).fetchall()
index_names = {row[0] for row in indexes}
assert "idx_labl_location" in index_names
assert "idx_labl_filter" in index_names
def test_animal_aliases_indexes_exist(self, migrated_db):
"""Animal aliases table has required indexes."""
indexes = migrated_db.execute(
"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='animal_aliases'"
).fetchall()
index_names = {row[0] for row in indexes}
assert "idx_aa_survivor" in index_names

250
tests/test_models_animals.py Normal file
View File

@@ -0,0 +1,250 @@
# ABOUTME: Tests for animal-related Pydantic models.
# ABOUTME: Validates Animal, LiveAnimal, and AnimalAlias models.
import pytest
from pydantic import ValidationError
from animaltrack.events.enums import AnimalStatus, LifeStage, Origin, ReproStatus, Sex
from animaltrack.models.animals import Animal, AnimalAlias, LiveAnimal
class TestAnimal:
"""Tests for the Animal model."""
def test_valid_animal(self):
"""Animal with all required fields validates successfully."""
animal = Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert animal.animal_id == "01ARZ3NDEKTSV4RRFFQ69G5FAA"
assert animal.species_code == "duck"
assert animal.sex == Sex.MALE
assert animal.status == AnimalStatus.ALIVE
def test_animal_id_must_be_26_chars(self):
"""Animal with ID not exactly 26 chars raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
Animal(
animal_id="short",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert "at least 26 characters" in str(exc_info.value)
def test_location_id_must_be_26_chars(self):
"""Animal with location_id not exactly 26 chars raises ValidationError."""
with pytest.raises(ValidationError) as exc_info:
Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="short",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert "at least 26 characters" in str(exc_info.value)
def test_identified_defaults_to_false(self):
"""Identified defaults to False."""
animal = Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert animal.identified is False
def test_nickname_is_optional(self):
"""Nickname is optional and defaults to None."""
animal = Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert animal.nickname is None
def test_born_or_hatched_at_is_optional(self):
"""born_or_hatched_at is optional."""
animal = Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
born_or_hatched_at=1704067200000,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
assert animal.born_or_hatched_at == 1704067200000
def test_invalid_sex_raises_error(self):
"""Invalid sex value raises ValidationError."""
with pytest.raises(ValidationError):
Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex="invalid",
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status=AnimalStatus.ALIVE,
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
def test_invalid_status_raises_error(self):
"""Invalid status value raises ValidationError."""
with pytest.raises(ValidationError):
Animal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
status="invalid",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
origin=Origin.HATCHED,
first_seen_utc=1704067200000,
last_event_utc=1704067200000,
)
class TestLiveAnimal:
"""Tests for the LiveAnimal model."""
def test_valid_live_animal(self):
"""LiveAnimal with all required fields validates successfully."""
live_animal = LiveAnimal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
first_seen_utc=1704067200000,
)
assert live_animal.animal_id == "01ARZ3NDEKTSV4RRFFQ69G5FAA"
assert live_animal.location_id == "01ARZ3NDEKTSV4RRFFQ69G5FAV"
def test_tags_defaults_to_empty_list(self):
"""Tags defaults to empty list."""
live_animal = LiveAnimal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
first_seen_utc=1704067200000,
)
assert live_animal.tags == []
def test_tags_can_be_set(self):
"""Tags can be set to a list of strings."""
live_animal = LiveAnimal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
first_seen_utc=1704067200000,
tags=["friendly", "leader"],
)
assert live_animal.tags == ["friendly", "leader"]
def test_last_move_utc_is_optional(self):
"""last_move_utc is optional."""
live_animal = LiveAnimal(
animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
first_seen_utc=1704067200000,
)
assert live_animal.last_move_utc is None
def test_animal_id_must_be_26_chars(self):
"""LiveAnimal with ID not exactly 26 chars raises ValidationError."""
with pytest.raises(ValidationError):
LiveAnimal(
animal_id="short",
location_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
species_code="duck",
sex=Sex.MALE,
repro_status=ReproStatus.INTACT,
life_stage=LifeStage.ADULT,
first_seen_utc=1704067200000,
)
class TestAnimalAlias:
"""Tests for the AnimalAlias model."""
def test_valid_alias(self):
"""AnimalAlias with all required fields validates successfully."""
alias = AnimalAlias(
alias_animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
survivor_animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAB",
merged_at_utc=1704067200000,
)
assert alias.alias_animal_id == "01ARZ3NDEKTSV4RRFFQ69G5FAA"
assert alias.survivor_animal_id == "01ARZ3NDEKTSV4RRFFQ69G5FAB"
assert alias.merged_at_utc == 1704067200000
def test_alias_animal_id_must_be_26_chars(self):
"""AnimalAlias with short alias_animal_id raises ValidationError."""
with pytest.raises(ValidationError):
AnimalAlias(
alias_animal_id="short",
survivor_animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAB",
merged_at_utc=1704067200000,
)
def test_survivor_animal_id_must_be_26_chars(self):
"""AnimalAlias with short survivor_animal_id raises ValidationError."""
with pytest.raises(ValidationError):
AnimalAlias(
alias_animal_id="01ARZ3NDEKTSV4RRFFQ69G5FAA",
survivor_animal_id="short",
merged_at_utc=1704067200000,
)