tliu93
3d3c2bcc57
Collapse the three data layers into one. app/db.py now exposes a single Base, a cached engine bound to app_database_url with SQLite WAL enabled, and get_engine/get_session_local/reset_db_caches/get_db_session. Delete app/auth_db.py, app/poo_db.py and app/models/base.py. All models (auth, config, public_ip, location, poo) inherit the one Base and register on a single metadata. Dependencies converge to a single get_db; all routes use it. Also update the alembic env.py files (app/location/poo) and tests that imported the removed modules so the suite stays green, and drop the obsolete test_legacy_style_location_db test whose flow (app reading a separate location DB) no longer exists. Location/poo Alembic chains, adopt scripts and adoption tests remain for M1-T04; config fields remain for M1-T05. pytest 109 passed; ruff clean (pre-existing only); WAL verified; single Base.metadata holds all seven tables.
286 lines
8.3 KiB
Python
286 lines
8.3 KiB
Python
from sqlalchemy import text
|
|
|
|
from app.config import Settings, get_settings
|
|
from app.dependencies import get_app_settings, get_homeassistant_client
|
|
from app.main import create_app
|
|
|
|
|
|
class _FakeHomeAssistantClient:
|
|
def __init__(self) -> None:
|
|
self.sensor_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 test_homeassistant_publish_records_location(location_client) -> None:
|
|
client, engine = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "record",
|
|
"content": "{'person': 'tianyu', 'latitude': '1.23', 'longitude': '4.56'}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT person, latitude, longitude, altitude "
|
|
"FROM location ORDER BY datetime DESC LIMIT 1"
|
|
)
|
|
).one()
|
|
|
|
assert row.person == "tianyu"
|
|
assert row.latitude == 1.23
|
|
assert row.longitude == 4.56
|
|
assert row.altitude == 0.0
|
|
|
|
|
|
def test_homeassistant_publish_records_location_with_altitude(location_client) -> None:
|
|
client, engine = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "record",
|
|
"content": (
|
|
"{'person': 'tianyu-alt', 'latitude': '1.23', "
|
|
"'longitude': '4.56', 'altitude': '7.89'}"
|
|
),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
|
|
with engine.connect() as conn:
|
|
row = conn.execute(
|
|
text(
|
|
"SELECT person, latitude, longitude, altitude "
|
|
"FROM location ORDER BY datetime DESC LIMIT 1"
|
|
)
|
|
).one()
|
|
|
|
assert row.person == "tianyu-alt"
|
|
assert row.latitude == 1.23
|
|
assert row.longitude == 4.56
|
|
assert row.altitude == 7.89
|
|
|
|
|
|
def test_homeassistant_publish_rejects_invalid_envelope(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "record",
|
|
"content": "{}",
|
|
"extra": "not-allowed",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
assert "extra" not in response.text
|
|
|
|
|
|
def test_homeassistant_publish_rejects_invalid_json_body(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
content='{"target": "location_recorder", "action": "record", "content": ',
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
|
|
|
|
def test_homeassistant_publish_rejects_missing_content(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "record",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
assert "content" not in response.text
|
|
|
|
|
|
def test_homeassistant_publish_returns_internal_error_for_unconfigured_ticktick(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "ticktick",
|
|
"action": "create_action_task",
|
|
"content": "{'action': 'take out trash', 'due_hour': 6}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert response.text == "internal server error"
|
|
|
|
|
|
def test_homeassistant_publish_rejects_invalid_ticktick_content(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "ticktick",
|
|
"action": "create_action_task",
|
|
"content": "{}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
|
|
|
|
def test_homeassistant_publish_poo_get_latest_publishes_latest_status(
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
) -> None:
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
|
|
app_url = auth_database["app_url"]
|
|
engine = create_engine(app_url, connect_args={"check_same_thread": False})
|
|
|
|
fake_ha = _FakeHomeAssistantClient()
|
|
settings = Settings(
|
|
poo_sensor_entity_name="sensor.test_poo_status",
|
|
poo_sensor_friendly_name="Poo Status",
|
|
)
|
|
|
|
test_app = create_app()
|
|
test_app.dependency_overrides[get_homeassistant_client] = lambda: fake_ha
|
|
test_app.dependency_overrides[get_app_settings] = lambda: settings
|
|
|
|
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": "done",
|
|
"latitude": 1.23,
|
|
"longitude": 4.56,
|
|
},
|
|
)
|
|
|
|
try:
|
|
with TestClient(test_app) as client:
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "poo_recorder",
|
|
"action": "get_latest",
|
|
"content": "",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.text == ""
|
|
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 fake_ha.sensor_calls[0]["attributes"]["last_poo"]
|
|
finally:
|
|
test_app.dependency_overrides.clear()
|
|
get_settings.cache_clear()
|
|
engine.dispose()
|
|
|
|
|
|
def test_homeassistant_publish_returns_internal_error_for_unknown_poo_action(
|
|
ready_location_database,
|
|
ready_poo_database,
|
|
auth_database,
|
|
) -> None:
|
|
from fastapi.testclient import TestClient
|
|
|
|
fake_ha = _FakeHomeAssistantClient()
|
|
settings = Settings(
|
|
poo_sensor_entity_name="sensor.test_poo_status",
|
|
poo_sensor_friendly_name="Poo Status",
|
|
)
|
|
|
|
test_app = create_app()
|
|
test_app.dependency_overrides[get_homeassistant_client] = lambda: fake_ha
|
|
test_app.dependency_overrides[get_app_settings] = lambda: settings
|
|
|
|
try:
|
|
with TestClient(test_app) as client:
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "poo_recorder",
|
|
"action": "unknown_action",
|
|
"content": "",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert response.text == "internal server error"
|
|
assert fake_ha.sensor_calls == []
|
|
finally:
|
|
test_app.dependency_overrides.clear()
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_homeassistant_publish_returns_not_implemented_for_unknown_location_action(
|
|
location_client,
|
|
) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "unknown_action",
|
|
"content": "{}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 500
|
|
assert response.text == "internal server error"
|
|
|
|
|
|
def test_homeassistant_publish_rejects_invalid_location_content(location_client) -> None:
|
|
client, _ = location_client
|
|
|
|
response = client.post(
|
|
"/homeassistant/publish",
|
|
json={
|
|
"target": "location_recorder",
|
|
"action": "record",
|
|
"content": "{'person': 'tianyu', 'latitude': 'bad-lat', 'longitude': '4.56'}",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.text == "bad request"
|
|
assert "bad-lat" not in response.text
|