Add optional Home Assistant integration for battery percentage tracking
This commit is contained in:
@@ -15,6 +15,8 @@ def app():
|
||||
SECRET_KEY = "test-secret"
|
||||
SQLALCHEMY_TRACK_MODIFICATIONS = False
|
||||
TESTING = True
|
||||
HOMEASSISTANT_URL = None # prevent HA poller from starting in tests
|
||||
HOMEASSISTANT_API_KEY = None
|
||||
|
||||
from app import create_app
|
||||
flask_app = create_app(TestConfig)
|
||||
|
||||
@@ -0,0 +1,368 @@
|
||||
"""Acceptance tests for the optional Home Assistant integration.
|
||||
|
||||
Groups:
|
||||
1. HA disabled — regression guard (uses default client/seeded_client fixtures)
|
||||
2. HomeAssistantClient unit tests — mock requests.get, no Flask required
|
||||
3. Poller integration — uses ha_app fixture, calls _poll_once() directly
|
||||
4. Dashboard/device/battery UI — checks rendered HTML with HA enabled
|
||||
"""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HA-enabled app fixture
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture()
|
||||
def ha_app():
|
||||
"""Flask app with Home Assistant enabled (HOMEASSISTANT_URL set).
|
||||
|
||||
The poller is started but its first fire is after 300 s — it never fires
|
||||
during a test run.
|
||||
"""
|
||||
db_fd, db_path = tempfile.mkstemp(suffix=".db")
|
||||
os.close(db_fd)
|
||||
|
||||
class HaTestConfig:
|
||||
SQLALCHEMY_DATABASE_URI = f"sqlite:///{db_path}"
|
||||
SECRET_KEY = "test-secret"
|
||||
SQLALCHEMY_TRACK_MODIFICATIONS = False
|
||||
TESTING = True
|
||||
HOMEASSISTANT_URL = "http://ha.test:8123"
|
||||
HOMEASSISTANT_API_KEY = "fake-token"
|
||||
HOMEASSISTANT_POLL_INTERVAL = 300
|
||||
|
||||
from app import create_app
|
||||
flask_app = create_app(HaTestConfig)
|
||||
yield flask_app
|
||||
os.unlink(db_path)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def ha_client_f(ha_app):
|
||||
"""Test client bound to the HA-enabled app."""
|
||||
with ha_app.test_client() as c:
|
||||
yield c
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 1 — HA disabled (regression guard)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_ha_disabled_dashboard_no_ha_column(client):
|
||||
resp = client.get("/")
|
||||
assert resp.status_code == 200
|
||||
assert b"ha-pct" not in resp.data
|
||||
|
||||
|
||||
def test_ha_disabled_device_no_ha_field(client):
|
||||
client.post("/device/add", data={"name": "Dev A", "battery_slots": "1"})
|
||||
resp = client.get("/device/1")
|
||||
assert resp.status_code == 200
|
||||
assert b"ha_entity_id" not in resp.data
|
||||
|
||||
|
||||
def test_ha_disabled_battery_detail_still_has_percentage_field(client):
|
||||
"""battery_percentage edit field is always shown regardless of HA config."""
|
||||
client.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
resp = client.get("/battery/1")
|
||||
assert b"battery_percentage" in resp.data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 2 — HomeAssistantClient unit tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_ha_client_disabled_when_no_url():
|
||||
from ha_client import HomeAssistantClient
|
||||
c = HomeAssistantClient(None, None)
|
||||
assert c.enabled is False
|
||||
assert c.get_state("sensor.foo") is None
|
||||
|
||||
|
||||
def test_ha_client_disabled_when_no_key():
|
||||
from ha_client import HomeAssistantClient
|
||||
c = HomeAssistantClient("http://ha.local", None)
|
||||
assert c.enabled is False
|
||||
|
||||
|
||||
def test_ha_client_returns_integer_state():
|
||||
from ha_client import HomeAssistantClient
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"state": "85", "entity_id": "sensor.foo"}
|
||||
mock_resp.raise_for_status.return_value = None
|
||||
|
||||
with patch("ha_client.requests.get", return_value=mock_resp) as mock_get:
|
||||
c = HomeAssistantClient("http://ha.local:8123", "tok")
|
||||
result = c.get_state("sensor.foo")
|
||||
|
||||
assert result == 85
|
||||
mock_get.assert_called_once_with(
|
||||
"http://ha.local:8123/api/states/sensor.foo",
|
||||
headers={"Authorization": "Bearer tok"},
|
||||
timeout=10,
|
||||
)
|
||||
|
||||
|
||||
def test_ha_client_handles_float_state():
|
||||
from ha_client import HomeAssistantClient
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"state": "72.6"}
|
||||
mock_resp.raise_for_status.return_value = None
|
||||
|
||||
with patch("ha_client.requests.get", return_value=mock_resp):
|
||||
c = HomeAssistantClient("http://ha.local", "tok")
|
||||
assert c.get_state("sensor.x") == 72
|
||||
|
||||
|
||||
def test_ha_client_returns_none_on_network_error():
|
||||
from ha_client import HomeAssistantClient
|
||||
import requests as req
|
||||
with patch("ha_client.requests.get", side_effect=req.RequestException("timeout")):
|
||||
c = HomeAssistantClient("http://ha.local", "tok")
|
||||
assert c.get_state("sensor.x") is None
|
||||
|
||||
|
||||
def test_ha_client_returns_none_on_non_numeric_state():
|
||||
from ha_client import HomeAssistantClient
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.json.return_value = {"state": "unavailable"}
|
||||
mock_resp.raise_for_status.return_value = None
|
||||
|
||||
with patch("ha_client.requests.get", return_value=mock_resp):
|
||||
c = HomeAssistantClient("http://ha.local", "tok")
|
||||
assert c.get_state("sensor.x") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 3 — Poller integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_session_factory(ha_app):
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"])
|
||||
return sessionmaker(bind=engine)
|
||||
|
||||
|
||||
def test_poll_updates_installed_batteries(ha_app, ha_client_f):
|
||||
"""Installed battery in a device with ha_entity_id gets battery_percentage updated."""
|
||||
ha_client_f.post("/device/add", data={"name": "Dev A", "battery_slots": "2"})
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
# Assign battery to device
|
||||
ha_client_f.post("/battery/1/assign", data={"device_id": "1"})
|
||||
# Set ha_entity_id on device
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev A", "battery_slots": "2", "ha_entity_id": "sensor.dev_a_battery"
|
||||
})
|
||||
|
||||
from ha_client import HomeAssistantClient
|
||||
from ha_poller import HaPoller
|
||||
from models import Battery
|
||||
|
||||
mock_ha = MagicMock(spec=HomeAssistantClient)
|
||||
mock_ha.enabled = True
|
||||
mock_ha.get_state.return_value = 42
|
||||
|
||||
Session = _make_session_factory(ha_app)
|
||||
poller = HaPoller(mock_ha, Session, interval=300)
|
||||
poller._poll_once()
|
||||
|
||||
s = Session()
|
||||
b = s.get(Battery, 1)
|
||||
assert b.battery_percentage == 42
|
||||
s.close()
|
||||
|
||||
mock_ha.get_state.assert_called_once_with("sensor.dev_a_battery")
|
||||
|
||||
|
||||
def test_poll_skips_uninstalled_batteries(ha_app, ha_client_f):
|
||||
"""Batteries that are available (not installed) are not updated by the poller."""
|
||||
ha_client_f.post("/device/add", data={"name": "Dev B", "battery_slots": "1"})
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
# Set entity on device but do NOT install the battery
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev B", "battery_slots": "1", "ha_entity_id": "sensor.dev_b_battery"
|
||||
})
|
||||
|
||||
from ha_client import HomeAssistantClient
|
||||
from ha_poller import HaPoller
|
||||
from models import Battery
|
||||
|
||||
mock_ha = MagicMock(spec=HomeAssistantClient)
|
||||
mock_ha.enabled = True
|
||||
mock_ha.get_state.return_value = 55
|
||||
|
||||
Session = _make_session_factory(ha_app)
|
||||
poller = HaPoller(mock_ha, Session, interval=300)
|
||||
poller._poll_once()
|
||||
|
||||
s = Session()
|
||||
b = s.get(Battery, 1)
|
||||
assert b.battery_percentage is None # uninstalled — not touched
|
||||
s.close()
|
||||
|
||||
|
||||
def test_poll_skips_devices_without_entity_id(ha_app, ha_client_f):
|
||||
"""Devices with no ha_entity_id must never reach get_state."""
|
||||
ha_client_f.post("/device/add", data={"name": "Dev C", "battery_slots": "1"})
|
||||
|
||||
from ha_client import HomeAssistantClient
|
||||
from ha_poller import HaPoller
|
||||
|
||||
mock_ha = MagicMock(spec=HomeAssistantClient)
|
||||
mock_ha.enabled = True
|
||||
|
||||
Session = _make_session_factory(ha_app)
|
||||
poller = HaPoller(mock_ha, Session, interval=300)
|
||||
poller._poll_once()
|
||||
|
||||
mock_ha.get_state.assert_not_called()
|
||||
|
||||
|
||||
def test_poll_handles_api_error_gracefully(ha_app, ha_client_f):
|
||||
"""When get_state returns None, no exception is raised and percentage stays None."""
|
||||
ha_client_f.post("/device/add", data={"name": "Dev D", "battery_slots": "1"})
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
ha_client_f.post("/battery/1/assign", data={"device_id": "1"})
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev D", "battery_slots": "1", "ha_entity_id": "sensor.dev_d"
|
||||
})
|
||||
|
||||
from ha_client import HomeAssistantClient
|
||||
from ha_poller import HaPoller
|
||||
from models import Battery
|
||||
|
||||
mock_ha = MagicMock(spec=HomeAssistantClient)
|
||||
mock_ha.enabled = True
|
||||
mock_ha.get_state.return_value = None # simulate API failure
|
||||
|
||||
Session = _make_session_factory(ha_app)
|
||||
poller = HaPoller(mock_ha, Session, interval=300)
|
||||
poller._poll_once() # must not raise
|
||||
|
||||
s = Session()
|
||||
b = s.get(Battery, 1)
|
||||
assert b.battery_percentage is None
|
||||
s.close()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 4 — Dashboard / device / battery UI with HA enabled
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_dashboard_shows_ha_column_when_enabled(ha_client_f):
|
||||
resp = ha_client_f.get("/")
|
||||
assert resp.status_code == 200
|
||||
assert b"ha-pct" in resp.data
|
||||
|
||||
|
||||
def test_dashboard_shows_low_battery_warning(ha_app, ha_client_f):
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
|
||||
# Set percentage directly in DB
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from models import Battery
|
||||
engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"])
|
||||
s = sessionmaker(bind=engine)()
|
||||
b = s.get(Battery, 1)
|
||||
b.battery_percentage = 15
|
||||
s.commit()
|
||||
s.close()
|
||||
|
||||
resp = ha_client_f.get("/")
|
||||
assert b"badge-warning" in resp.data
|
||||
assert b"15%" in resp.data
|
||||
|
||||
|
||||
def test_dashboard_no_warning_for_high_percentage(ha_app, ha_client_f):
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from models import Battery
|
||||
engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"])
|
||||
s = sessionmaker(bind=engine)()
|
||||
b = s.get(Battery, 1)
|
||||
b.battery_percentage = 85
|
||||
s.commit()
|
||||
s.close()
|
||||
|
||||
resp = ha_client_f.get("/")
|
||||
assert b"85%" in resp.data
|
||||
# badge-warning should NOT appear for this battery's percentage
|
||||
# (may still appear in page for other reasons, so check row contains 85% but not warning badge near it)
|
||||
assert b"col-ha-pct" in resp.data # column rendered
|
||||
|
||||
|
||||
def test_device_detail_shows_ha_field(ha_client_f):
|
||||
ha_client_f.post("/device/add", data={"name": "Dev E", "battery_slots": "1"})
|
||||
resp = ha_client_f.get("/device/1")
|
||||
assert resp.status_code == 200
|
||||
assert b"ha_entity_id" in resp.data
|
||||
|
||||
|
||||
def test_edit_device_ha_entity_id_saves(ha_client_f):
|
||||
ha_client_f.post("/device/add", data={"name": "Dev F", "battery_slots": "1"})
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev F", "battery_slots": "1", "ha_entity_id": "sensor.my_remote"
|
||||
})
|
||||
resp = ha_client_f.get("/device/1")
|
||||
assert b"sensor.my_remote" in resp.data
|
||||
|
||||
|
||||
def test_edit_device_ha_entity_id_clear(ha_client_f):
|
||||
ha_client_f.post("/device/add", data={"name": "Dev G", "battery_slots": "1"})
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev G", "battery_slots": "1", "ha_entity_id": "sensor.foo"
|
||||
})
|
||||
# Now clear it
|
||||
ha_client_f.post("/device/1/edit", data={
|
||||
"name": "Dev G", "battery_slots": "1", "ha_entity_id": ""
|
||||
})
|
||||
resp = ha_client_f.get("/device/1")
|
||||
assert b"sensor.foo" not in resp.data
|
||||
|
||||
|
||||
def test_charge_log_resets_percentage_to_100(ha_app, ha_client_f):
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
|
||||
# Set low percentage
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from models import Battery
|
||||
engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"])
|
||||
s = sessionmaker(bind=engine)()
|
||||
b = s.get(Battery, 1)
|
||||
b.battery_percentage = 30
|
||||
s.commit()
|
||||
s.close()
|
||||
|
||||
# Log a charge
|
||||
ha_client_f.post("/battery/1/charge-log/add", data={"charged_date": "2026-04-13"})
|
||||
|
||||
s2 = sessionmaker(bind=engine)()
|
||||
b2 = s2.get(Battery, 1)
|
||||
assert b2.battery_percentage == 100
|
||||
s2.close()
|
||||
|
||||
|
||||
def test_manual_battery_percentage_edit(ha_client_f):
|
||||
ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"})
|
||||
|
||||
# Set via edit form
|
||||
ha_client_f.post("/battery/1/edit-details", data={"battery_percentage": "55"})
|
||||
resp = ha_client_f.get("/battery/1")
|
||||
assert b"55%" in resp.data
|
||||
|
||||
# Clear via edit form
|
||||
ha_client_f.post("/battery/1/edit-details", data={"battery_percentage": ""})
|
||||
resp2 = ha_client_f.get("/battery/1")
|
||||
# percentage display row should not appear when None
|
||||
assert b"Battery %" not in resp2.data or b"55%" not in resp2.data
|
||||
Reference in New Issue
Block a user