114 lines
3.7 KiB
Python
114 lines
3.7 KiB
Python
import json
|
|
from urllib import error
|
|
|
|
import pytest
|
|
|
|
from app.config import Settings
|
|
from app.integrations.homeassistant import (
|
|
HomeAssistantClient,
|
|
HomeAssistantConfigError,
|
|
HomeAssistantRequestError,
|
|
)
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int):
|
|
self.status_code = status_code
|
|
|
|
def getcode(self) -> int:
|
|
return self.status_code
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
|
|
def _configured_settings() -> Settings:
|
|
return Settings(
|
|
home_assistant_base_url="http://ha.local:8123",
|
|
home_assistant_auth_token="secret-token",
|
|
home_assistant_timeout_seconds=1.5,
|
|
)
|
|
|
|
|
|
def test_publish_sensor_posts_expected_request(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured = {}
|
|
client = HomeAssistantClient(
|
|
settings=_configured_settings(),
|
|
timeout_seconds=_configured_settings().home_assistant_timeout_seconds,
|
|
)
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["url"] = req.full_url
|
|
captured["timeout"] = timeout
|
|
captured["authorization"] = req.headers["Authorization"]
|
|
captured["content_type"] = req.headers["Content-type"]
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
return _FakeResponse(200)
|
|
|
|
monkeypatch.setattr("app.integrations.homeassistant.request.urlopen", fake_urlopen)
|
|
|
|
client.publish_sensor(
|
|
entity_id="sensor.test_poo_status",
|
|
state="happy",
|
|
attributes={"friendly_name": "Poo Status"},
|
|
)
|
|
|
|
assert captured["url"] == "http://ha.local:8123/api/states/sensor.test_poo_status"
|
|
assert captured["timeout"] == pytest.approx(1.5)
|
|
assert captured["authorization"] == "Bearer secret-token"
|
|
assert captured["content_type"] == "application/json"
|
|
assert captured["body"] == {
|
|
"entity_id": "sensor.test_poo_status",
|
|
"state": "happy",
|
|
"attributes": {"friendly_name": "Poo Status"},
|
|
}
|
|
|
|
|
|
def test_trigger_webhook_posts_expected_request(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
captured = {}
|
|
client = HomeAssistantClient(settings=_configured_settings())
|
|
|
|
def fake_urlopen(req, timeout):
|
|
captured["url"] = req.full_url
|
|
captured["body"] = json.loads(req.data.decode("utf-8"))
|
|
return _FakeResponse(201)
|
|
|
|
monkeypatch.setattr("app.integrations.homeassistant.request.urlopen", fake_urlopen)
|
|
|
|
client.trigger_webhook(webhook_id="poo-status", body={"status": "done"})
|
|
|
|
assert captured["url"] == "http://ha.local:8123/api/webhook/poo-status"
|
|
assert captured["body"] == {"status": "done"}
|
|
|
|
|
|
def test_homeassistant_client_raises_on_http_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client = HomeAssistantClient(settings=_configured_settings())
|
|
|
|
def fake_urlopen(req, timeout):
|
|
raise error.HTTPError(req.full_url, 500, "boom", hdrs=None, fp=None)
|
|
|
|
monkeypatch.setattr("app.integrations.homeassistant.request.urlopen", fake_urlopen)
|
|
|
|
with pytest.raises(HomeAssistantRequestError, match="HTTP 500"):
|
|
client.publish_sensor(entity_id="sensor.test_status", state="bad")
|
|
|
|
|
|
def test_homeassistant_client_raises_when_not_configured() -> None:
|
|
client = HomeAssistantClient(settings=Settings(_env_file=None))
|
|
|
|
with pytest.raises(HomeAssistantConfigError, match="not configured"):
|
|
client.publish_sensor(entity_id="sensor.test_status", state="ok")
|
|
|
|
|
|
def test_homeassistant_client_raises_on_invalid_arguments() -> None:
|
|
client = HomeAssistantClient(settings=_configured_settings())
|
|
|
|
with pytest.raises(ValueError, match="entity_id"):
|
|
client.publish_sensor(entity_id="", state="ok")
|
|
|
|
with pytest.raises(ValueError, match="webhook_id"):
|
|
client.trigger_webhook(webhook_id="", body={})
|