73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
"""Tests for database schema creation."""
|
|
|
|
from sqlalchemy import create_engine, inspect
|
|
|
|
from vigilar.storage.schema import metadata, pets, pet_sightings, wildlife_sightings, pet_training_images
|
|
|
|
|
|
def test_tables_created(tmp_path):
|
|
db_path = tmp_path / "test.db"
|
|
engine = create_engine(f"sqlite:///{db_path}", echo=False)
|
|
metadata.create_all(engine)
|
|
|
|
assert db_path.exists()
|
|
|
|
inspector = inspect(engine)
|
|
table_names = inspector.get_table_names()
|
|
|
|
expected = [
|
|
"cameras", "sensors", "sensor_states", "events", "recordings",
|
|
"system_events", "arm_state_log", "alert_log", "push_subscriptions",
|
|
]
|
|
for name in expected:
|
|
assert name in table_names, f"Missing table: {name}"
|
|
|
|
|
|
class TestPetTables:
|
|
def test_pets_table_exists(self, test_db):
|
|
with test_db.connect() as conn:
|
|
result = conn.execute(pets.insert().values(
|
|
id="pet-1", name="Angel", species="cat", breed="DSH",
|
|
color_description="black", training_count=0, created_at=1000.0,
|
|
))
|
|
row = conn.execute(pets.select().where(pets.c.id == "pet-1")).first()
|
|
assert row is not None
|
|
assert row.name == "Angel"
|
|
assert row.species == "cat"
|
|
|
|
def test_pet_sightings_table(self, test_db):
|
|
with test_db.begin() as conn:
|
|
conn.execute(pets.insert().values(
|
|
id="pet-1", name="Angel", species="cat", training_count=0, created_at=1000.0,
|
|
))
|
|
conn.execute(pet_sightings.insert().values(
|
|
ts=1000.0, pet_id="pet-1", species="cat", camera_id="kitchen",
|
|
confidence=0.92, labeled=True,
|
|
))
|
|
rows = conn.execute(pet_sightings.select()).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0].camera_id == "kitchen"
|
|
|
|
def test_wildlife_sightings_table(self, test_db):
|
|
with test_db.begin() as conn:
|
|
conn.execute(wildlife_sightings.insert().values(
|
|
ts=1000.0, species="bear", threat_level="PREDATOR",
|
|
camera_id="front", confidence=0.88,
|
|
))
|
|
rows = conn.execute(wildlife_sightings.select()).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0].threat_level == "PREDATOR"
|
|
|
|
def test_pet_training_images_table(self, test_db):
|
|
with test_db.begin() as conn:
|
|
conn.execute(pets.insert().values(
|
|
id="pet-1", name="Angel", species="cat", training_count=0, created_at=1000.0,
|
|
))
|
|
conn.execute(pet_training_images.insert().values(
|
|
pet_id="pet-1", image_path="/var/vigilar/pets/training/angel/001.jpg",
|
|
source="upload", created_at=1000.0,
|
|
))
|
|
rows = conn.execute(pet_training_images.select()).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0].source == "upload"
|