93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from flight_map.cli import main
|
|
from flight_map.geo import DEFAULT_BOUNDS
|
|
from flight_map.storage import (
|
|
Observation,
|
|
create_session,
|
|
database,
|
|
get_session,
|
|
initialize,
|
|
load_observations,
|
|
purge_session,
|
|
save_observations,
|
|
)
|
|
|
|
|
|
def seed(path: Path) -> tuple[str, str]:
|
|
with database(path) as connection:
|
|
initialize(connection)
|
|
sessions = [create_session(connection, DEFAULT_BOUNDS, 2000, 30, 100) for _ in range(2)]
|
|
for session in sessions:
|
|
save_observations(
|
|
connection,
|
|
session,
|
|
[
|
|
Observation("abc123", "TEST", 110, 4.82, 52.24, 900, 1000),
|
|
],
|
|
120,
|
|
)
|
|
return sessions[0].id, sessions[1].id
|
|
|
|
|
|
def test_purge_only_selected_session(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
|
path = tmp_path / "test.sqlite"
|
|
target, other = seed(path)
|
|
assert main(["purge", "--db", str(path), "--session", target, "--yes"]) == 0
|
|
assert f"Purged session {target} and 1 observation(s)." in capsys.readouterr().out
|
|
with database(path, readonly=True) as connection:
|
|
with pytest.raises(ValueError, match="not found"):
|
|
get_session(connection, target)
|
|
assert load_observations(connection, target, 100, 120) == []
|
|
assert get_session(connection, other).bounds == DEFAULT_BOUNDS
|
|
assert len(load_observations(connection, other, 100, 120)) == 1
|
|
assert connection.execute("PRAGMA foreign_key_check").fetchall() == []
|
|
|
|
|
|
def test_confirmation_and_unknown_session_preserve_data(tmp_path: Path) -> None:
|
|
path = tmp_path / "test.sqlite"
|
|
target, _ = seed(path)
|
|
before = path.read_bytes()
|
|
with pytest.raises(SystemExit):
|
|
main(["purge", "--db", str(path), "--session", target])
|
|
assert path.read_bytes() == before
|
|
assert main(["purge", "--db", str(path), "--session", "missing", "--yes"]) == 1
|
|
assert path.read_bytes() == before
|
|
|
|
|
|
def test_missing_database_is_not_created(tmp_path: Path) -> None:
|
|
path = tmp_path / "missing.sqlite"
|
|
assert main(["purge", "--db", str(path), "--session", "missing", "--yes"]) == 1
|
|
assert not path.exists()
|
|
|
|
|
|
def test_purge_rolls_back_observations_if_session_delete_fails(tmp_path: Path) -> None:
|
|
path = tmp_path / "test.sqlite"
|
|
target, _ = seed(path)
|
|
with database(path) as connection:
|
|
connection.execute("""
|
|
CREATE TRIGGER prevent_session_delete BEFORE DELETE ON sessions
|
|
BEGIN SELECT RAISE(ABORT, 'test failure'); END
|
|
""")
|
|
connection.commit()
|
|
with pytest.raises(sqlite3.IntegrityError, match="test failure"):
|
|
purge_session(connection, target)
|
|
assert get_session(connection, target).id == target
|
|
assert len(load_observations(connection, target, 100, 120)) == 1
|
|
connection.execute("DROP TRIGGER prevent_session_delete")
|
|
connection.commit()
|
|
assert purge_session(connection, target) == 1
|
|
|
|
|
|
def test_empty_session_can_be_purged(tmp_path: Path) -> None:
|
|
path = tmp_path / "test.sqlite"
|
|
with database(path) as connection:
|
|
initialize(connection)
|
|
session = create_session(connection, DEFAULT_BOUNDS, 2000, 30, 100)
|
|
assert purge_session(connection, session.id) == 0
|
|
with pytest.raises(ValueError, match="not found"):
|
|
get_session(connection, session.id)
|