tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Restore legacy poo inbound dispatch
pytest / test (push) Successful in 43s
Details
docker-image / build-and-push (push) Successful in 3m38s
Details

Browse Source
This commit is contained in:
Tianyu Liu
2026-04-20 23:33:57 +02:00
parent b9e7f51d51
commit 35aee79d93
3 changed files with 227 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/routes/homeassistant.py
+32 -4
View File
@@ -6,7 +6,19 @@ from fastapi.responses import PlainTextResponse, Response
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app.dependencies import get_db, get_ticktick_client
from app.config import Settings
from app.dependencies import (
get_app_settings,
get_db,
get_homeassistant_client,
get_poo_db,
get_ticktick_client,
)
from app.integrations.homeassistant import (
HomeAssistantClient,
HomeAssistantConfigError,
HomeAssistantRequestError,
)
from app.integrations.ticktick import TickTickClient, TickTickConfigError, TickTickRequestError
from app.schemas.homeassistant import HomeAssistantPublishEnvelope
from app.services.homeassistant_inbound import (
@@ -24,13 +36,23 @@ INTERNAL_SERVER_ERROR_MESSAGE = "internal server error"
async def publish_from_homeassistant(
request: Request,
db: Session = Depends(get_db),
poo_db: Session = Depends(get_poo_db),
settings: Settings = Depends(get_app_settings),
homeassistant_client: HomeAssistantClient = Depends(get_homeassistant_client),
ticktick_client: TickTickClient = Depends(get_ticktick_client),
) -> Response:
try:
raw_payload = await request.body()
data = json.loads(raw_payload)
envelope = HomeAssistantPublishEnvelope.model_validate(data)
handle_homeassistant_message(db, envelope, ticktick_client)
handle_homeassistant_message(
db,
envelope,
ticktick_client=ticktick_client,
poo_session=poo_db,
settings=settings,
homeassistant_client=homeassistant_client,
)
except json.JSONDecodeError as exc:
logger.warning("Rejected Home Assistant publish request due to invalid JSON: %s", exc)
return PlainTextResponse(BAD_REQUEST_MESSAGE, status_code=status.HTTP_400_BAD_REQUEST)
@@ -45,8 +67,14 @@ async def publish_from_homeassistant(
INTERNAL_SERVER_ERROR_MESSAGE,
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
except (TickTickConfigError, TickTickRequestError, RuntimeError) as exc:
logger.warning("Home Assistant publish request failed during TickTick handling: %s", exc)
except (
TickTickConfigError,
TickTickRequestError,
HomeAssistantConfigError,
HomeAssistantRequestError,
RuntimeError,
) as exc:
logger.warning("Home Assistant publish request failed during integration handling: %s", exc)
return PlainTextResponse(
INTERNAL_SERVER_ERROR_MESSAGE,
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
app/services/homeassistant_inbound.py
+37
View File
@@ -4,11 +4,14 @@ import json
from datetime import UTC, datetime, time, timedelta
from sqlalchemy.orm import Session
from app.config import Settings
from app.integrations.homeassistant import HomeAssistantClient
from app.integrations.ticktick import TICKTICK_DATETIME_FORMAT, TickTickClient, TickTickTask
from app.schemas.homeassistant import HomeAssistantPublishEnvelope
from app.schemas.location import LocationRecordRequest
from app.schemas.ticktick import TickTickActionTaskRequest
from app.services.location import record_location
from app.services.poo import publish_latest_poo_status
class UnsupportedHomeAssistantMessage(RuntimeError):
@@ -19,11 +22,23 @@ def handle_homeassistant_message(
session: Session,
envelope: HomeAssistantPublishEnvelope,
ticktick_client: TickTickClient | None = None,
poo_session: Session | None = None,
settings: Settings | None = None,
homeassistant_client: HomeAssistantClient | None = None,
) -> None:
if envelope.target == "location_recorder":
_handle_location_message(session, envelope)
return
if envelope.target == "poo_recorder":
_handle_poo_message(
envelope,
poo_session=poo_session,
settings=settings,
homeassistant_client=homeassistant_client,
)
return
if envelope.target == "ticktick":
_handle_ticktick_message(envelope, ticktick_client)
return
@@ -44,6 +59,28 @@ def _handle_location_message(session: Session, envelope: HomeAssistantPublishEnv
record_location(session, payload)
def _handle_poo_message(
envelope: HomeAssistantPublishEnvelope,
*,
poo_session: Session | None,
settings: Settings | None,
homeassistant_client: HomeAssistantClient | None,
) -> None:
if envelope.action != "get_latest":
raise UnsupportedHomeAssistantMessage(
f"Unsupported Home Assistant target/action: {envelope.target}/{envelope.action}"
)
if poo_session is None or settings is None or homeassistant_client is None:
raise RuntimeError("Poo recorder integration is unavailable")
publish_latest_poo_status(
session=poo_session,
settings=settings,
homeassistant_client=homeassistant_client,
)
def _handle_ticktick_message(
envelope: HomeAssistantPublishEnvelope,
ticktick_client: TickTickClient | None,
tests/test_homeassistant_inbound.py
+158
View File
@@ -1,5 +1,21 @@
from sqlalchemy import text
import app.db as app_db
import app.poo_db as poo_db
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
@@ -141,6 +157,148 @@ def test_homeassistant_publish_rejects_invalid_ticktick_content(location_client)
assert response.text == "bad request"
def test_homeassistant_publish_poo_get_latest_publishes_latest_status(
ready_location_database,
ready_poo_database,
auth_database,
monkeypatch,
) -> None:
location_engine = app_db.create_engine(
ready_location_database["location_url"],
connect_args={"check_same_thread": False},
)
location_session_local = app_db.sessionmaker(
bind=location_engine,
autoflush=False,
autocommit=False,
)
poo_engine = poo_db.create_engine(
ready_poo_database["poo_url"],
connect_args={"check_same_thread": False},
)
poo_session_local = poo_db.sessionmaker(
bind=poo_engine,
autoflush=False,
autocommit=False,
)
fake_ha = _FakeHomeAssistantClient()
settings = Settings(
poo_sensor_entity_name="sensor.test_poo_status",
poo_sensor_friendly_name="Poo Status",
)
monkeypatch.setattr(app_db, "engine", location_engine)
monkeypatch.setattr(app_db, "SessionLocal", location_session_local)
monkeypatch.setattr(poo_db, "poo_engine", poo_engine)
monkeypatch.setattr(poo_db, "PooSessionLocal", poo_session_local)
test_app = create_app()
test_app.dependency_overrides[get_homeassistant_client] = lambda: fake_ha
test_app.dependency_overrides[get_app_settings] = lambda: settings
with poo_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:
from fastapi.testclient import TestClient
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()
location_engine.dispose()
poo_engine.dispose()
def test_homeassistant_publish_returns_internal_error_for_unknown_poo_action(
ready_location_database,
ready_poo_database,
auth_database,
monkeypatch,
) -> None:
location_engine = app_db.create_engine(
ready_location_database["location_url"],
connect_args={"check_same_thread": False},
)
location_session_local = app_db.sessionmaker(
bind=location_engine,
autoflush=False,
autocommit=False,
)
poo_engine = poo_db.create_engine(
ready_poo_database["poo_url"],
connect_args={"check_same_thread": False},
)
poo_session_local = poo_db.sessionmaker(
bind=poo_engine,
autoflush=False,
autocommit=False,
)
fake_ha = _FakeHomeAssistantClient()
settings = Settings(
poo_sensor_entity_name="sensor.test_poo_status",
poo_sensor_friendly_name="Poo Status",
)
monkeypatch.setattr(app_db, "engine", location_engine)
monkeypatch.setattr(app_db, "SessionLocal", location_session_local)
monkeypatch.setattr(poo_db, "poo_engine", poo_engine)
monkeypatch.setattr(poo_db, "PooSessionLocal", poo_session_local)
test_app = create_app()
test_app.dependency_overrides[get_homeassistant_client] = lambda: fake_ha
test_app.dependency_overrides[get_app_settings] = lambda: settings
try:
from fastapi.testclient import TestClient
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()
location_engine.dispose()
poo_engine.dispose()
def test_homeassistant_publish_returns_not_implemented_for_unknown_location_action(
location_client,
) -> None: