"""Tests for the database layer — CRUD, cascade, transactions.""" import pytest from sqlalchemy import event from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from gps_denied.db.models import Base from gps_denied.db.repository import FlightRepository @pytest.fixture async def session(): """Create an in-memory SQLite database for each test.""" engine = create_async_engine("sqlite+aiosqlite://", echo=False) @event.listens_for(engine.sync_engine, "connect") def _set_sqlite_pragma(dbapi_connection, connection_record): cursor = dbapi_connection.cursor() cursor.execute("PRAGMA foreign_keys=ON") cursor.close() async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) async with async_session() as s: yield s await engine.dispose() @pytest.fixture def repo(session: AsyncSession) -> FlightRepository: return FlightRepository(session) CAM = { "focal_length": 25.0, "sensor_width": 23.5, "sensor_height": 15.6, "resolution_width": 6252, "resolution_height": 4168, } # ── Flight CRUD ─────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_insert_and_get_flight(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="Test_Flight_001", description="Test", start_lat=48.275, start_lon=37.385, altitude=400, camera_params=CAM, ) await session.commit() loaded = await repo.get_flight(flight.id) assert loaded is not None assert loaded.name == "Test_Flight_001" assert loaded.altitude == 400 @pytest.mark.asyncio async def test_list_flights(repo: FlightRepository, session: AsyncSession): await repo.insert_flight( name="F1", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) await repo.insert_flight( name="F2", description="", start_lat=0, start_lon=0, altitude=200, camera_params=CAM ) await session.commit() flights = await repo.list_flights() assert len(flights) == 2 @pytest.mark.asyncio async def test_update_flight(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="Old", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) await session.commit() ok = await repo.update_flight(flight.id, name="New") await session.commit() assert ok is True reloaded = await repo.get_flight(flight.id) assert reloaded.name == "New" @pytest.mark.asyncio async def test_delete_flight_cascade(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="Del", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) fid = flight.id # Add related entities await repo.insert_waypoint(fid, lat=48.0, lon=37.0, confidence=0.9) await repo.save_frame_result(fid, frame_id=1, gps_lat=48.0, gps_lon=37.0) await repo.save_heading(fid, frame_id=1, heading=90.0) await repo.save_image_metadata(fid, frame_id=1, file_path="/img/1.jpg") await repo.save_chunk(fid, start_frame_id=1, frames=[1, 2, 3]) await repo.insert_geofence(fid, nw_lat=49.0, nw_lon=36.0, se_lat=47.0, se_lon=38.0) await session.commit() # Delete flight — should cascade ok = await repo.delete_flight(fid) await session.commit() assert ok is True assert await repo.get_flight(fid) is None assert await repo.get_waypoints(fid) == [] assert await repo.get_frame_results(fid) == [] assert await repo.get_heading_history(fid) == [] assert await repo.load_chunks(fid) == [] # ── Waypoints ───────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_waypoint_crud(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="WP", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) wp = await repo.insert_waypoint(flight.id, lat=48.1, lon=37.2, confidence=0.8) await session.commit() wps = await repo.get_waypoints(flight.id) assert len(wps) == 1 assert wps[0].lat == 48.1 ok = await repo.update_waypoint(flight.id, wp.id, lat=48.2, refined=True) await session.commit() assert ok is True # ── Flight State ────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_flight_state(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="State", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) await session.commit() state = await repo.load_flight_state(flight.id) assert state is not None assert state.status == "created" await repo.save_flight_state(flight.id, status="processing", frames_total=500) await session.commit() state = await repo.load_flight_state(flight.id) assert state.status == "processing" assert state.frames_total == 500 # ── Frame Results ───────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_frame_results(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="FR", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) await repo.save_frame_result( flight.id, frame_id=1, gps_lat=48.0, gps_lon=37.0, confidence=0.95 ) await repo.save_frame_result( flight.id, frame_id=2, gps_lat=48.001, gps_lon=37.001, confidence=0.90 ) await session.commit() results = await repo.get_frame_results(flight.id) assert len(results) == 2 assert results[0].frame_id == 1 # ── Heading History ─────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_heading_history(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="HD", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) for i in range(5): await repo.save_heading(flight.id, frame_id=i, heading=float(i * 30)) await session.commit() latest = await repo.get_latest_heading(flight.id) assert latest == 120.0 # last frame heading last3 = await repo.get_heading_history(flight.id, last_n=3) assert len(last3) == 3 # ── Chunks ──────────────────────────────────────────────────────────────── @pytest.mark.asyncio async def test_chunks(repo: FlightRepository, session: AsyncSession): flight = await repo.insert_flight( name="CK", description="", start_lat=0, start_lon=0, altitude=100, camera_params=CAM ) await repo.save_chunk( flight.id, chunk_id="chunk_001", start_frame_id=1, end_frame_id=10, frames=[1, 2, 3, 4, 5], has_anchor=True, anchor_lat=48.0, anchor_lon=37.0, ) await session.commit() chunks = await repo.load_chunks(flight.id) assert len(chunks) == 1 assert chunks[0].chunk_id == "chunk_001" ok = await repo.delete_chunk(flight.id, "chunk_001") await session.commit() assert ok is True assert await repo.load_chunks(flight.id) == []