goodwe-addon/goodwe_controller/test_haentities.py

"""UX-13 P0: the entity list for the picker.  `python3 test_haentities.py`

`GET ha/entities` reads Home Assistant's states through the add-on's own
session (hass.states()) and must never fail the page: part 1 drives the real
HomeAssistant client against a local server that answers 502, garbage, the
wrong shape, nothing at all, or too slowly. Part 2 checks the 30 s cache and
what is served while Home Assistant is down. Part 3 serves the real route.
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
"""

import asyncio
import json
import logging
import sys
import time

from aiohttp import ClientSession, web as aioweb

from app import hass as H, haentities as E, web

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


STATES = [
    {"entity_id": "sensor.grid", "state": "812",
     "attributes": {"friendly_name": "Grid power", "unit_of_measurement": "W",
                    "device_class": "power"}},
    {"entity_id": "sensor.soc", "state": "54",
     "attributes": {"friendly_name": "Battery level", "unit_of_measurement": "%",
                    "device_class": "battery"}},
    {"entity_id": "number.setpoint", "state": "0", "attributes": {}},
    {"entity_id": "sensor.pv_kw", "state": "1.2",
     "attributes": {"unit_of_measurement": "kW"}},
    {"no_entity_id": True},
]


async def serve(handler):
    app = aioweb.Application()
    app.router.add_get("/api/states", handler)
    runner = aioweb.AppRunner(app)
    await runner.setup()
    site = aioweb.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    return runner, site._server.sockets[0].getsockname()[1]


async def read_with(handler, timeout_s=10):
    runner, port = await serve(handler)
    H.CORE_API, H.STATES_TIMEOUT_S = f"http://127.0.0.1:{port}/api", timeout_s
    try:
        async with ClientSession() as session:
            t0 = time.monotonic()
            got = await H.HomeAssistant(session, "token").states()
            read_with.took = time.monotonic() - t0
            return got
    finally:
        await runner.cleanup()


async def ok(_r):
    return aioweb.json_response(STATES)


async def bad_gateway(_r):
    return aioweb.Response(status=502)


async def garbage(_r):
    return aioweb.Response(text="<html>", content_type="application/json")


async def not_a_list(_r):
    return aioweb.json_response({"message": "hi"})


async def slow(_r):
    await asyncio.sleep(2)
    return aioweb.json_response(STATES)


async def closed_port():
    H.CORE_API = "http://127.0.0.1:9/api"
    async with ClientSession() as session:
        return await H.HomeAssistant(session, "token").states()


print("1. hass.states() answers None, never an exception")
check("a normal answer is the state list", asyncio.run(read_with(ok)) == STATES)
for name, handler in (("502 while HA restarts", bad_gateway),
                      ("a body that is not JSON", garbage),
                      ("JSON that is not a list", not_a_list)):
    check(f"{name} -> None", asyncio.run(read_with(handler)) is None)
class _Lines(logging.Handler):
    def __init__(self):
        super().__init__()
        self.lines = []

    def emit(self, record):
        if record.levelno >= logging.WARNING:
            self.lines.append(record.getMessage())


_lines = _Lines()
logging.getLogger().addHandler(_lines)
check("an answer slower than the timeout -> None, at the timeout",
      asyncio.run(read_with(slow, timeout_s=0.3)) is None
      and read_with.took < 1.0)
# UX-13 P1 test finding (c): a timeout's own text is empty, which left
# "Reading the entity list failed: " with no reason.
check("...and its WARNING line gives the reason, not an empty one "
      f"({_lines.lines})", any(
          x == "Reading the entity list failed: no answer within 0.3 seconds."
          for x in _lines.lines)
      and not any(x.rstrip().endswith("failed:") for x in _lines.lines))
logging.getLogger().removeHandler(_lines)
check("no server at all -> None", asyncio.run(closed_port()) is None)
check("the timeout for the list is the same 10 s as every other read",
      "STATES_TIMEOUT_S = 10 " in open(H.__file__, encoding="utf-8").read())


class FakeHass:
    def __init__(self):
        self.answer, self.reads = STATES, 0

    async def states(self):
        self.reads += 1
        await asyncio.sleep(0.01)
        return self.answer


class Clock:
    def __init__(self):
        self.t = 1000.0

    def __call__(self):
        return self.t


print("\n2. the cache, and Home Assistant being down")
_h, _c = FakeHass(), Clock()
_cache = E.EntityCache(_h, clock=_c)


async def burst():
    return await asyncio.gather(*(_cache.payload("power") for _ in range(5)))


_p = asyncio.run(burst())
check("five pickers opening at once cost one read of Home Assistant",
      _h.reads == 1 and all(x == _p[0] for x in _p))
_ids = [r["entity_id"] for r in _p[0]["entities"]]
check("power entities come first, then the rest, and a broken state is left "
      f"out ({_ids})", _ids == ["sensor.grid", "sensor.pv_kw",
                                "number.setpoint", "sensor.soc"]
      and [r["match"] for r in _p[0]["entities"]] == [True, True, False, False])
check("each row carries id, name, state, unit and device class",
      _p[0]["entities"][0] == {"entity_id": "sensor.grid", "name": "Grid power",
                               "state": "812", "unit": "W",
                               "device_class": "power", "match": True})
check("the kinds filter as the pickers need",
      [r["entity_id"] for r in _p[0]["entities"]
       if E.matches(r, "pct")] == ["sensor.soc"]
      and [r["entity_id"] for r in _p[0]["entities"]
           if E.matches(r, "number")] == ["number.setpoint"])
_c.t += 10
asyncio.run(_cache.payload())
check("within 30 s the cached list is served", _h.reads == 1)
check("peek() hands the validator the fresh list without a read",
      set(_cache.peek()) == set(_ids) and _h.reads == 1)
_h.answer = None
_c.t += 30
_down = asyncio.run(_cache.payload())
check("Home Assistant down after a good read: the old list, marked old",
      _h.reads == 2 and _down["available"] and _down["text"] == E.STALE_TEXT
      and len(_down["entities"]) == 4 and _down["age_s"] >= 30)
check("...and peek() then gives nothing, so no warning rests on an old list",
      _cache.peek() is None)


async def down_windows():
    """One loop throughout: a contended asyncio.Lock binds to its loop."""
    hass, clock = FakeHass(), Clock()
    hass.answer = None
    cache = E.EntityCache(hass, clock=clock)
    counts = []
    for step in (0, 1, 30):
        clock.t += step
        await asyncio.gather(*(cache.payload() for _ in range(5)))
        counts.append(hass.reads)
    return counts


check("five polls while Home Assistant is down cost one read, the polls just "
      "after wait for the next window, and that window tries once",
      asyncio.run(down_windows()) == [1, 1, 2])
_h2 = FakeHass()
_h2.answer = None
_empty = asyncio.run(E.EntityCache(_h2).payload("power"))
check("Home Assistant down from the start: an empty list that says why",
      _empty == {"available": False, "text": E.DOWN_TEXT, "entities": []})


print("\n3. the route")


class FakeController:
    def __init__(self, hass):
        self.hass, self.o = hass, {}


async def route(hass, query=""):
    app = web.build_app(FakeController(hass), selfcfg=object())
    runner = aioweb.AppRunner(app)
    await runner.setup()
    site = aioweb.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    try:
        async with ClientSession() as s:
            async with s.get(f"http://127.0.0.1:{port}/ha/entities{query}") as r:
                return r.status, await r.json()
    finally:
        await runner.cleanup()


_st, _body = asyncio.run(route(FakeHass(), "?kind=pct"))
check("GET ha/entities?kind=pct answers 200 with the matching entity first",
      _st == 200 and _body["entities"][0]["entity_id"] == "sensor.soc")
_st, _body = asyncio.run(route(_h2))
check("with Home Assistant down it still answers 200, with the reason",
      _st == 200 and _body["available"] is False and _body["entities"] == [])
check("the words shown are short and plain (STYLE: one sentence)",
      all(len(t.split()) <= 12 and t.count(".") == 1
          for t in (E.DOWN_TEXT, E.STALE_TEXT)))
print(json.dumps(_body))

print()
if fails:
    print(f"{len(fails)} FAILED")
    sys.exit(1)
print("all passed")