ha-nvidia-shield/tests/test_ha_api_drift.py

"""Run the Home-Assistant API-drift sweep (t37).

Skips cleanly when homeassistant is not installed — every other test file runs
against hand-written stubs, and this is the one check that must see the real
package. CI's `ha-api-drift` job installs it:

    uvx --with homeassistant --with pytest-asyncio pytest tests/test_ha_api_drift.py -q

The sweep itself lives in ``tests/ha_api_drift.py`` (documented there).
"""

from __future__ import annotations

import ast
import importlib
import subprocess
import sys
import threading
from pathlib import Path
from types import SimpleNamespace

import ha_api_drift
import pytest

pytest.importorskip(
    "homeassistant",
    reason="the API-drift sweep resolves our call sites against a real install",
)

#: The three t36 drifts plus the `options_updates` kwarg t35 removed — the
#: sweep must flag every one of them, or it cannot protect us from the next.
PRE_FIX_SOURCES = {
    "d1_browse_media_source.py": (
        "from homeassistant.components.media_player import BrowseMediaSource\n"
    ),
    "d2_util_diagnostic.py": (
        "from homeassistant.util.diagnostic import async_redact_data\n"
    ),
    "d3_unawaited_discovery.py": (
        "from homeassistant import config_entries\n"
        "\n"
        "\n"
        "class PreFixFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
        "    async def async_step_zeroconf(self, discovery_info):\n"
        "        self._async_handle_discovery_without_unique_id()\n"
    ),
    # t57 UNIT 0a: a wrong kwarg on a FLOW-DISPLAY API. `async_show_menu` takes
    # no `errors` in HA 2026.10.1, and the integration passed one — so every
    # ordinary setup path raised TypeError at the moment a user submitted the
    # first form, which is invisible to a stub suite whose stubs accept it.
    "t57_show_menu_errors.py": (
        "from homeassistant import config_entries\n"
        "\n"
        "\n"
        "class PreFixSetupMenu(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
        "    async def async_step_user(self, user_input=None):\n"
        "        return self.async_show_menu(\n"
        "            step_id='menu',\n"
        "            menu_options=['finish'],\n"
        "            errors={\'base\': \'nope\'},\n"
        "        )\n"
    ),
    "t35_options_updates.py": (
        "from homeassistant import config_entries\n"
        "\n"
        "\n"
        "class PreFixReauthFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
        "    async def async_step_reauth_confirm(self, user_input=None):\n"
        "        self.async_update_reload_and_abort(\n"
        "            self.entry, options_updates={'protocol_enabled': True}\n"
        "        )\n"
        "        return self.async_abort(reason='reauth_successful')\n"
    ),
}


def _require_installed_homeassistant():
    """Skip when sys.modules holds a test stub rather than an install."""
    import sys

    module = sys.modules["homeassistant"]
    path = getattr(module, "__file__", None)
    if path is None or not Path(path).exists():
        pytest.skip("homeassistant in sys.modules is a test stub, not an install")


def test_no_api_drift_in_the_production_tree():
    """The gate: every HA import, kwarg and await matches the real package."""
    _require_installed_homeassistant()

    report = ha_api_drift.scan()

    assert report.findings == [], "\n" + str(report)


def test_the_sweep_is_not_vacuous():
    """The counters prove the walk really covered the production tree."""
    _require_installed_homeassistant()

    report = ha_api_drift.scan()

    # Floors, not targets: the numbers live in ha_api_drift.FLOORS — one source
    # of truth for this wrapper and the standalone entry point (t39 F1) — so a
    # broken or mis-targeted walk fails here instead of looking green.
    violations = report.floor_violations()

    assert violations == [], "\n".join(violations)


def test_the_sweep_flags_the_pre_fix_defects(tmp_path):
    """Red proof: the t36 drifts and the options_updates kwarg are all caught."""
    _require_installed_homeassistant()

    for name, source in PRE_FIX_SOURCES.items():
        (tmp_path / name).write_text(source, encoding="utf-8")

    report = ha_api_drift.scan(tmp_path)

    found = {(finding.path, finding.kind) for finding in report.findings}
    assert found == {
        ("d1_browse_media_source.py", "broken-import"),
        ("d2_util_diagnostic.py", "broken-import"),
        ("d3_unawaited_discovery.py", "missing-await"),
        ("t35_options_updates.py", "unknown-kwarg"),
        ("t57_show_menu_errors.py", "unknown-kwarg"),
    }, "\n" + str(report)


def test_the_sweep_accepts_the_fixed_forms(tmp_path):
    """The same call sites written the way production writes them are clean."""
    _require_installed_homeassistant()

    (tmp_path / "fixed.py").write_text(
        "from homeassistant.components.media_player import BrowseMedia\n"
        "from homeassistant.helpers.redact import async_redact_data\n"
        "from homeassistant import config_entries\n"
        "\n"
        "\n"
        "class FixedFlow(config_entries.ConfigFlow, domain='nvidia_shield'):\n"
        "    async def async_step_zeroconf(self, discovery_info):\n"
        "        await self._async_handle_discovery_without_unique_id()\n"
        "\n"
        "    def finish(self):\n"
        "        return self.async_update_reload_and_abort(\n"
        "            self.entry, options={'protocol_enabled': True}\n"
        "        )\n",
        encoding="utf-8",
    )

    report = ha_api_drift.scan(tmp_path)

    assert report.findings == [], "\n" + str(report)


def _run_drift_module(*paths):
    """Run the standalone entry point in this (HA-bearing) interpreter."""
    return subprocess.run(
        [
            sys.executable,
            str(Path(ha_api_drift.__file__).resolve()),
            *[str(path) for path in paths],
        ],
        capture_output=True,
        text=True,
        check=False,
    )


def test_the_standalone_entry_point_passes_on_the_production_tree():
    """F1: the CLI is the sweep, with the same verdict on the real tree."""
    result = _run_drift_module()

    assert result.returncode == 0, result.stderr
    assert "no Home-Assistant API drift" in result.stdout


def test_the_standalone_entry_point_fails_on_an_empty_directory(tmp_path):
    """t39 F1: zeros must not exit 0 — the floors apply to the CLI too."""
    result = _run_drift_module(tmp_path)

    assert result.returncode != 0
    assert "0 modules" in result.stdout  # the counter that used to look green
    assert "FLOOR:" in result.stderr
    assert "drift sweep FAILED" in result.stderr


def test_the_standalone_entry_point_fails_on_a_missing_path(tmp_path):
    """A mis-specified path is a failure, not a silent zero-module sweep."""
    result = _run_drift_module(tmp_path / "not-a-directory")

    assert result.returncode != 0
    assert "bad-path" in result.stdout


def test_no_floor_constants_are_duplicated_in_this_wrapper():
    """F1: the floors come from ha_api_drift.FLOORS — nowhere else."""
    source = Path(__file__).read_text(encoding="utf-8")
    tree = ast.parse(source)

    local_ints = [
        target.id
        for node in tree.body
        if isinstance(node, ast.Assign)
        for target in node.targets
        if isinstance(target, ast.Name)
        and isinstance(node.value, ast.Constant)
        and isinstance(node.value.value, int)
    ]

    assert local_ints == [], f"floor-like constants belong in ha_api_drift: {local_ints}"
    assert "floor_violations" in source


def test_the_declared_python_floor_never_lags_homeassistant():
    """F2: the CI guard's floor is HA's own Requires-Python, not a guess."""
    _require_installed_homeassistant()

    import importlib.metadata as metadata

    declared = metadata.metadata("homeassistant")["Requires-Python"]
    parsed = ha_api_drift.parse_requires_python(declared)

    assert parsed is not None, f"unparsable Requires-Python: {declared!r}"
    assert tuple(parsed) <= ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR, (
        f"tests/ha_api_drift.py declares {ha_api_drift.HOMEASSISTANT_PYTHON_FLOOR} "
        f"but the installed homeassistant requires {declared} — raise the constant "
        f"and the setup-python version in .gitea/workflows/test.yml"
    )


# ------------------------------------------------- entity contract (t40, real HA)
#
# The stub suites cannot see this class of defect: their harness replaces
# ``async_write_ha_state``, so a non-property ``extra_state_attributes`` looks
# fine there. Only HA's own ``Entity.__async_calculate_state`` — which does
# ``attr |= self.extra_state_attributes`` — turns it into
# ``TypeError: 'method' object is not iterable`` and stops the entity from ever
# rendering. These tests drive that real path.

_REPO_ROOT = Path(__file__).resolve().parent.parent


class _RecordingStates:
    """Minimal ``hass.states``: keeps what HA's entity writes."""

    def __init__(self):
        self.registry: dict = {}
        self.written: list = []

    def get(self, entity_id):
        return self.registry.get(entity_id)

    def async_set_internal(self, entity_id, state, attr, *args):
        self.written.append((entity_id, state, dict(attr)))


def _real_ha_entity(module_name, class_name, *, options=None, entity_id=None):
    """Build a production entity under real homeassistant (no platform)."""
    if str(_REPO_ROOT) not in sys.path:
        sys.path.insert(0, str(_REPO_ROOT))
    module = importlib.import_module(f"custom_components.nvidia_shield.{module_name}")
    states = _RecordingStates()
    hass = SimpleNamespace(
        loop_thread_id=threading.get_ident(),
        states=states,
        data={},
        config=SimpleNamespace(components=set()),
    )
    entry = SimpleNamespace(
        entry_id="e1", data={}, options=dict(options or {}), title="Shield"
    )
    entity = getattr(module, class_name)(hass, entry)
    entity.entity_id = entity_id
    # HA's real translation-key paths (``unit_of_measurement``, device-class
    # names) need the platform's translation data; without it a translation-keyed
    # entity raises "cannot have a translation key ... before being added to the
    # entity platform". PlatformData is the real container, so the entity is
    # exercised with the same shape it sees in production.
    from homeassistant.helpers.entity_platform import PlatformData

    entity.platform_data = PlatformData(
        hass, domain=module_name, platform_name="nvidia_shield"
    )
    # Not added through an EntityComponent: HA would warn once via
    # _suggest_report_issue (which needs hass.data["integrations"]) — set the
    # flag it uses to report that once instead of faking a whole platform.
    entity._no_platform_reported = True
    return entity, states


def _write_ha_state(entity, states):
    """Run HA's real state write and return what reached the state machine."""
    entity.async_write_ha_state()
    assert states.written, f"{type(entity).__name__} wrote no state"
    return states.written[-1]


#: Every concrete entity class in the integration, exercised above against the
#: REAL state machine. t40: the media player's defect was a state-write crash, so
#: "the whole tree renders" is the contract under test, not a single entity.
_REAL_HA_MATRIX = (
    ("media_player", "NvidiaShieldMediaPlayer", None, "media_player.shield"),
    ("sensor", "NvidiaShieldCurrentAppSensor", None, "sensor.shield_current_app"),
    (
        "sensor",
        "NvidiaShieldNowPlayingSensor",
        {"jellyfin_entity": "media_player.jellyfin"},
        "sensor.shield_now_playing",
    ),
    (
        "sensor",
        "NvidiaShieldAppsSensor",
        {"adb_enabled": True},
        "sensor.shield_installed_apps",
    ),
    ("remote", "NvidiaShieldRemoteEntity", None, "remote.shield"),
    ("button", "NvidiaShieldPairAdbButton", None, "button.shield_pair_adb"),
)

#: Classified as an entity by the sweep, but an abstract intermediate base HA
#: never instantiates — its concrete subclasses are in the matrix.
_INTERMEDIATE_BASES = frozenset({"NvidiaShieldSensorBase"})


@pytest.mark.parametrize(
    ("module_name", "class_name", "options", "entity_id"), _REAL_HA_MATRIX
)
def test_entities_calculate_state_under_real_homeassistant(
    module_name, class_name, options, entity_id
):
    """t40 blocker: HA's own state write must not raise for ANY entity.

    ``Entity.async_write_ha_state`` → ``__async_calculate_state`` merges the
    extra attributes with ``attr |= ...``; the pre-fix media player raised
    ``TypeError: 'method' object is not iterable`` here, so it could never render.
    """
    _require_installed_homeassistant()

    entity, states = _real_ha_entity(
        module_name, class_name, options=options, entity_id=entity_id
    )

    written_id, state, attrs = _write_ha_state(entity, states)

    assert written_id == entity_id
    assert isinstance(state, str)
    assert isinstance(attrs, dict)


def test_the_media_player_extra_attributes_reach_the_state_machine():
    """The property's payload must survive HA's merge, not merely not raise."""
    _require_installed_homeassistant()

    entity, states = _real_ha_entity(
        "media_player", "NvidiaShieldMediaPlayer", entity_id="media_player.shield"
    )

    _entity_id, _state, attrs = _write_ha_state(entity, states)

    assert {"app_id", "app_name"} <= set(attrs), sorted(attrs)


def test_the_real_ha_matrix_covers_every_entity_class_in_the_tree():
    """A new entity class must join the matrix above, or this test fails.

    The sweep classifies entities from the AST; this keeps the real-HA coverage
    in step with that classification instead of silently shrinking.
    """
    _require_installed_homeassistant()

    classified: set[str] = set()
    for path in sorted(ha_api_drift.SHIELD_DIR.glob("*.py")):
        classified |= ha_api_drift.entity_class_names(
            ast.parse(path.read_text(encoding="utf-8"))
        )
    covered = {class_name for _module, class_name, _options, _entity_id in _REAL_HA_MATRIX}

    uncovered = classified - covered - _INTERMEDIATE_BASES

    assert uncovered == set(), (
        f"entity classes never exercised against real HA: {sorted(uncovered)} — "
        f"add them to _REAL_HA_MATRIX"
    )


# ------------------------------------------------- t62: dead code and the floor
#
# Both defects pinned below shipped while every other gate was green:
# connection.py carried a duplicated copy of the label-drain body AFTER its
# `return cached` (valid Python, dead, invisible to this sweep), and the
# call-argument floor sat at 20 while the counts had moved to ~46, so deleting a
# flow-display API owner from HA_CALL_OWNERS did not trip it.

#: t62 F2: the shape that shipped — a bare docstring plus the whole drain left
#: after `return cached`, inside the same method. Line 6 is the dead statement.
UNREACHABLE_PRE_FIX = (
    "class Sweep:\n"
    "    def labels(self, requests):\n"
    "        cached = {}\n"
    "        return cached\n"
    "\n"
    '        """Start at most one label fetch per watchdog cycle."""\n'
    "        if not requests:\n"
    "            return\n"
    "        requests.discard(min(requests))\n"
)


def test_the_sweep_flags_the_pre_fix_unreachable_body(tmp_path):
    """t62 F2 red proof: the duplicated drain body after `return cached` is dead."""
    (tmp_path / "dead_sweep.py").write_text(UNREACHABLE_PRE_FIX, encoding="utf-8")

    findings = ha_api_drift.scan_unreachable(tmp_path)

    assert [(finding.path, finding.kind) for finding in findings] == [
        ("dead_sweep.py", "unreachable-statement")
    ], findings
    assert findings[0].lineno == 6  # the dead docstring, not the return


def test_every_terminal_statement_kind_is_flagged(tmp_path):
    """`return`, `raise`, `continue` and `break` all end a block (t62 F2)."""
    (tmp_path / "terminals.py").write_text(
        "def a():\n"
        "    return 1\n"
        "    x = 1\n"
        "\n"
        "\n"
        "def b():\n"
        "    raise SystemExit(1)\n"
        "    y = 2\n"
        "\n"
        "\n"
        "def c(items):\n"
        "    for item in items:\n"
        "        continue\n"
        "        z = 3\n"
        "    while True:\n"
        "        break\n"
        "        w = 4\n",
        encoding="utf-8",
    )

    findings = ha_api_drift.scan_unreachable(tmp_path)

    assert [finding.kind for finding in findings] == ["unreachable-statement"] * 4, findings


def test_the_production_tree_has_no_unreachable_statements():
    """The gate itself: dead code is a finding, so it cannot ship silently."""
    assert ha_api_drift.scan_unreachable() == []


def test_the_unreachable_pass_walks_real_statement_blocks():
    """The floor proves "no findings" is not a pass that stopped parsing."""
    report = ha_api_drift.scan()

    assert report.statement_blocks >= ha_api_drift.FLOORS["statement_blocks"]


def test_removing_a_flow_display_owner_trips_the_call_argument_floor(monkeypatch):
    """t62: the floor must notice an API OWNER going away, not just a bad walk."""
    _require_installed_homeassistant()

    intact = ha_api_drift.scan()
    assert intact.floor_violations() == []

    reduced = dict(ha_api_drift.HA_CALL_OWNERS)
    del reduced["async_show_menu"]  # the t57 UNIT 0a owner: 6 checked arguments
    monkeypatch.setattr(ha_api_drift, "HA_CALL_OWNERS", reduced)

    fewer = ha_api_drift.scan()

    assert fewer.call_args < intact.call_args
    assert fewer.call_args < ha_api_drift.FLOORS["call_args"], (
        f"removing async_show_menu left {fewer.call_args} checked arguments, "
        f"still at or above the floor of {ha_api_drift.FLOORS['call_args']} — "
        "the flow-display owners are not load-bearing"
    )
    assert any("call_args" in line for line in fewer.floor_violations()), (
        fewer.floor_violations()
    )