249 lines
7.3 KiB
Python
249 lines
7.3 KiB
Python
from pathlib import Path
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.dependencies import get_app_settings, get_homeassistant_client
|
|
from scripts.poo_db_adopt import (
|
|
EXPECTED_USER_VERSION,
|
|
POO_BASELINE_REVISION,
|
|
PooDatabaseAdoptionError,
|
|
adopt_or_initialize_poo_db,
|
|
)
|
|
|
|
|
|
class _FakeHomeAssistantClient:
|
|
def __init__(self) -> None:
|
|
self.sensor_calls: list[dict] = []
|
|
self.webhook_calls: list[dict] = []
|
|
|
|
def publish_sensor(self, *, entity_id: str, state: str, attributes: dict | None = None) -> None:
|
|
self.sensor_calls.append(
|
|
{"entity_id": entity_id, "state": state, "attributes": attributes or {}}
|
|
)
|
|
|
|
def trigger_webhook(self, *, webhook_id: str, body) -> None:
|
|
self.webhook_calls.append({"webhook_id": webhook_id, "body": body})
|
|
|
|
|
|
@pytest.fixture
|
|
def poo_client_with_overrides(poo_client):
|
|
client, engine = poo_client
|
|
fake_ha = _FakeHomeAssistantClient()
|
|
settings = Settings(
|
|
poo_webhook_id="poo-hook",
|
|
poo_sensor_entity_name="sensor.test_poo_status",
|
|
poo_sensor_friendly_name="Poo Status",
|
|
)
|
|
client.app.dependency_overrides[get_homeassistant_client] = lambda: fake_ha
|
|
client.app.dependency_overrides[get_app_settings] = lambda: settings
|
|
try:
|
|
yield client, engine, fake_ha
|
|
finally:
|
|
client.app.dependency_overrides.clear()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_poo_record_endpoint_writes_row_and_notifies_homeassistant(
|
|
poo_client_with_overrides,
|
|
) -> None:
|
|
client, engine, fake_ha = poo_client_with_overrides
|
|
|
|
response = client.post(
|
|
"/poo/record",
|
|
json={
|
|
"status": "done",
|
|
"latitude": "1.23",
|
|
"longitude": "4.56",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT status, latitude, longitude FROM poo_records "
|
|
"ORDER BY timestamp DESC LIMIT 1"
|
|
)
|
|
).one()
|
|
|
|
assert row.status == "done"
|
|
assert row.latitude == pytest.approx(1.23)
|
|
assert row.longitude == pytest.approx(4.56)
|
|
assert len(fake_ha.sensor_calls) == 1
|
|
assert fake_ha.sensor_calls[0]["entity_id"] == "sensor.test_poo_status"
|
|
assert fake_ha.sensor_calls[0]["state"] == "done"
|
|
assert fake_ha.sensor_calls[0]["attributes"]["friendly_name"] == "Poo Status"
|
|
assert len(fake_ha.webhook_calls) == 1
|
|
assert fake_ha.webhook_calls[0] == {
|
|
"webhook_id": "poo-hook",
|
|
"body": {"status": "done"},
|
|
}
|
|
|
|
|
|
def test_poo_latest_endpoint_publishes_latest_status(poo_client_with_overrides) -> None:
|
|
client, engine, fake_ha = poo_client_with_overrides
|
|
|
|
with engine.begin() as conn:
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO poo_records (timestamp, status, latitude, longitude) "
|
|
"VALUES (:timestamp, :status, :latitude, :longitude)"
|
|
),
|
|
{
|
|
"timestamp": "2026-04-20T10:05Z",
|
|
"status": "urgent",
|
|
"latitude": 3.21,
|
|
"longitude": 6.54,
|
|
},
|
|
)
|
|
|
|
response = client.get("/poo/latest")
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
assert len(fake_ha.sensor_calls) == 1
|
|
assert fake_ha.sensor_calls[0]["state"] == "urgent"
|
|
assert fake_ha.sensor_calls[0]["attributes"]["last_poo"]
|
|
|
|
|
|
def test_poo_record_endpoint_rejects_unknown_fields(poo_client_with_overrides) -> None:
|
|
client, _, _ = poo_client_with_overrides
|
|
|
|
response = client.post(
|
|
"/poo/record",
|
|
json={
|
|
"status": "done",
|
|
"latitude": "1.23",
|
|
"longitude": "4.56",
|
|
"extra": "nope",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
|
|
|
|
def test_poo_record_endpoint_rejects_invalid_latitude(poo_client_with_overrides) -> None:
|
|
client, _, _ = poo_client_with_overrides
|
|
|
|
response = client.post(
|
|
"/poo/record",
|
|
json={
|
|
"status": "done",
|
|
"latitude": "oops",
|
|
"longitude": "4.56",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
|
|
|
|
def test_poo_latest_endpoint_returns_ok_when_no_record_exists(poo_client_with_overrides) -> None:
|
|
client, _, _ = poo_client_with_overrides
|
|
|
|
response = client.get("/poo/latest")
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
|
|
|
|
def test_poo_db_adoption_initializes_new_db(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "new_poo.db"
|
|
|
|
result = adopt_or_initialize_poo_db(f"sqlite:///{database_path}")
|
|
|
|
assert result == "initialized"
|
|
assert database_path.exists()
|
|
|
|
conn = sqlite3.connect(database_path)
|
|
try:
|
|
revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
|
|
poo_table = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'poo_records'"
|
|
).fetchone()
|
|
finally:
|
|
conn.close()
|
|
|
|
assert revision == POO_BASELINE_REVISION
|
|
assert poo_table is not None
|
|
|
|
|
|
def test_poo_db_adoption_validates_and_stamps_legacy_db(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "legacy_poo.db"
|
|
conn = sqlite3.connect(database_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE poo_records (
|
|
timestamp TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
latitude REAL NOT NULL,
|
|
longitude REAL NOT NULL,
|
|
PRIMARY KEY (timestamp)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
result = adopt_or_initialize_poo_db(f"sqlite:///{database_path}")
|
|
|
|
assert result == "adopted"
|
|
|
|
conn = sqlite3.connect(database_path)
|
|
try:
|
|
revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
|
|
finally:
|
|
conn.close()
|
|
|
|
assert revision == POO_BASELINE_REVISION
|
|
|
|
|
|
def test_poo_db_adoption_fails_closed_on_schema_mismatch(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "bad_poo_schema.db"
|
|
conn = sqlite3.connect(database_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE poo_records (
|
|
timestamp TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
latitude REAL NOT NULL,
|
|
PRIMARY KEY (timestamp)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute(f"PRAGMA user_version = {EXPECTED_USER_VERSION}")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
with pytest.raises(PooDatabaseAdoptionError, match="schema does not match"):
|
|
adopt_or_initialize_poo_db(f"sqlite:///{database_path}")
|
|
|
|
|
|
def test_poo_db_adoption_fails_closed_on_user_version_mismatch(tmp_path: Path) -> None:
|
|
database_path = tmp_path / "bad_poo_user_version.db"
|
|
conn = sqlite3.connect(database_path)
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE poo_records (
|
|
timestamp TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
latitude REAL NOT NULL,
|
|
longitude REAL NOT NULL,
|
|
PRIMARY KEY (timestamp)
|
|
)
|
|
"""
|
|
)
|
|
conn.execute("PRAGMA user_version = 999")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
with pytest.raises(PooDatabaseAdoptionError, match="Expected PRAGMA user_version"):
|
|
adopt_or_initialize_poo_db(f"sqlite:///{database_path}")
|