Alo/animaltrack
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
1c836c6f7d65cb9df762014788ace5d3785e2d47
animaltrack/tests/test_migration_animal_registry.py
Petru Paler 739b7bfe32 feat: add animal registry schema and models 
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>
2025-12-28 18:59:24 +00:00

533 lines
22 KiB
Python
Raw Blame History

# 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
Reference in New Issue View Git Blame Copy Permalink