473 lines
17 KiB
Python
473 lines
17 KiB
Python
"""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")
|
|
# "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
|