"""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 WTF_CSRF_ENABLED = False # disable CSRF validation in tests 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") # "55%" still appears once in history; the info-table row is gone (current value is None) assert resp2.data.count(b"55%") == 1 # --------------------------------------------------------------------------- # Group 3 additions — skip-on-no-change + history logging (poller) # --------------------------------------------------------------------------- def test_poll_skips_update_when_percentage_unchanged(ha_app, ha_client_f): """No write and no pct_log entry when the polled value matches the stored value.""" ha_client_f.post("/device/add", data={"name": "Dev NoCh", "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 NoCh", "battery_slots": "1", "ha_entity_id": "sensor.noch" }) from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Battery, BatteryPctLog engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"]) s = sessionmaker(bind=engine)() s.get(Battery, 1).battery_percentage = 50 s.commit(); s.close() from ha_client import HomeAssistantClient from ha_poller import HaPoller mock_ha = MagicMock(spec=HomeAssistantClient) mock_ha.enabled = True mock_ha.get_state.return_value = 50 # same value — should be a no-op Session = _make_session_factory(ha_app) HaPoller(mock_ha, Session, interval=300)._poll_once() s2 = sessionmaker(bind=engine)() assert s2.query(BatteryPctLog).count() == 0 s2.close() def test_poll_creates_pct_log_on_change(ha_app, ha_client_f): """A pct_log entry with source='poll' is written when the value changes.""" ha_client_f.post("/device/add", data={"name": "Dev Chg", "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 Chg", "battery_slots": "1", "ha_entity_id": "sensor.chg" }) from ha_client import HomeAssistantClient from ha_poller import HaPoller from models import BatteryPctLog mock_ha = MagicMock(spec=HomeAssistantClient) mock_ha.enabled = True mock_ha.get_state.return_value = 72 Session = _make_session_factory(ha_app) HaPoller(mock_ha, Session, interval=300)._poll_once() s = Session() logs = s.query(BatteryPctLog).filter_by(battery_id=1).all() assert len(logs) == 1 assert logs[0].percentage == 72 assert logs[0].source == "poll" s.close() # --------------------------------------------------------------------------- # Group 4 additions — history logging (routes / UI) # --------------------------------------------------------------------------- def test_charge_log_creates_pct_log_entry(ha_app, ha_client_f): """Logging a charge creates a pct_log entry with percentage=100, source='charge'.""" ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"}) ha_client_f.post("/battery/1/charge-log/add", data={"charged_date": "2026-04-13"}) from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import BatteryPctLog engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"]) s = sessionmaker(bind=engine)() logs = s.query(BatteryPctLog).filter_by(battery_id=1).all() assert len(logs) == 1 assert logs[0].percentage == 100 assert logs[0].source == "charge" s.close() def test_manual_edit_creates_pct_log_entry(ha_client_f): """Saving a new battery_percentage via edit-details creates a history entry.""" ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"}) ha_client_f.post("/battery/1/edit-details", data={"battery_percentage": "65"}) resp = ha_client_f.get("/battery/1") assert b"65%" in resp.data assert b"manual" in resp.data # source shown in history table def test_manual_edit_no_log_when_unchanged(ha_client_f): """Saving the same percentage twice produces only one history entry.""" ha_client_f.post("/battery/add", data={"brand": "X", "count": "1"}) ha_client_f.post("/battery/1/edit-details", data={"battery_percentage": "40"}) ha_client_f.post("/battery/1/edit-details", data={"battery_percentage": "40"}) resp = ha_client_f.get("/battery/1") # "40%" appears once in the info section and once in history — not twice in history assert resp.data.count(b"40%") == 2 # --------------------------------------------------------------------------- # Group 5 — live fetch on device page load + /ha/entities endpoint # --------------------------------------------------------------------------- def test_device_detail_shows_live_pct(ha_app, ha_client_f): """Opening a device page fetches live % from HA and displays it.""" ha_client_f.post("/device/add", data={"name": "Dev Live", "battery_slots": "1"}) ha_client_f.post("/device/1/edit", data={ "name": "Dev Live", "battery_slots": "1", "ha_entity_id": "sensor.live_test" }) mock_resp = MagicMock() mock_resp.json.return_value = {"state": "78", "entity_id": "sensor.live_test"} mock_resp.raise_for_status.return_value = None with patch("ha_client.requests.get", return_value=mock_resp): resp = ha_client_f.get("/device/1") assert resp.status_code == 200 assert b"78%" in resp.data def test_device_detail_updates_battery_on_load(ha_app, ha_client_f): """Battery percentage is updated (with pct_log) when device page is loaded and value changed.""" ha_client_f.post("/device/add", data={"name": "Dev Update", "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 Update", "battery_slots": "1", "ha_entity_id": "sensor.update_test" }) mock_resp = MagicMock() mock_resp.json.return_value = {"state": "63"} mock_resp.raise_for_status.return_value = None with patch("ha_client.requests.get", return_value=mock_resp): ha_client_f.get("/device/1") from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Battery, BatteryPctLog engine = create_engine(ha_app.config["SQLALCHEMY_DATABASE_URI"]) s = sessionmaker(bind=engine)() b = s.get(Battery, 1) assert b.battery_percentage == 63 logs = s.query(BatteryPctLog).filter_by(battery_id=1, source="poll").all() assert len(logs) == 1 assert logs[0].percentage == 63 s.close() def test_ha_entities_endpoint(ha_app, ha_client_f): """GET /ha/entities returns only battery-related entities from HA.""" mock_resp = MagicMock() mock_resp.json.return_value = [ {"entity_id": "sensor.tv_battery", "attributes": {"device_class": "battery", "friendly_name": "TV Battery"}}, {"entity_id": "sensor.remote_battery", "attributes": {"friendly_name": "Remote"}}, # 'battery' in id {"entity_id": "sensor.temperature", "attributes": {"device_class": "temperature", "friendly_name": "Temp"}}, {"entity_id": "light.living_room", "attributes": {}}, ] mock_resp.raise_for_status.return_value = None with patch("ha_client.requests.get", return_value=mock_resp): resp = ha_client_f.get("/ha/entities") assert resp.status_code == 200 data = resp.get_json() entity_ids = [e["entity_id"] for e in data] assert "sensor.tv_battery" in entity_ids assert "sensor.remote_battery" in entity_ids assert "sensor.temperature" not in entity_ids assert "light.living_room" not in entity_ids # friendly names present tv = next(e for e in data if e["entity_id"] == "sensor.tv_battery") assert tv["friendly_name"] == "TV Battery" def test_ha_entities_only_sensor_domain(ha_app, ha_client_f): """Only sensor. entities are returned — all other domains are excluded.""" mock_resp = MagicMock() mock_resp.json.return_value = [ {"entity_id": "sensor.remote_battery", "attributes": {"device_class": "battery", "friendly_name": "Remote"}}, {"entity_id": "binary_sensor.door_battery_low", "attributes": {"device_class": "battery", "friendly_name": "Door low"}}, {"entity_id": "binary_sensor.smoke_battery", "attributes": {"friendly_name": "Smoke"}}, {"entity_id": "input_number.battery_threshold", "attributes": {}}, ] mock_resp.raise_for_status.return_value = None with patch("ha_client.requests.get", return_value=mock_resp): resp = ha_client_f.get("/ha/entities") assert resp.status_code == 200 data = resp.get_json() entity_ids = [e["entity_id"] for e in data] assert entity_ids == ["sensor.remote_battery"]