ha-nvidia-shield/tests/test_now_playing_publication.py

"""tests/test_now_playing_publication.py — which entity states may carry a reading (t17/O2).

THE OBSERVATION THIS CLOSES. The batch review read `extra_state_attributes` and
argued the now-playing row publishes "the entity state verbatim", so a dashboard
could see `now_playing_state: on`, while the README documents only `playing`,
`paused`, `buffering` and `idle`. t10's author had reported the opposite (an
`on` or `off` device publishes no row at all). Both accounts were about a
different thing, and this file settles it BY EXECUTION:

* the t10 account is true for the NATURAL ladder — `on` means no source reported
  a power or playback fact, and a reading only exists when a tier answered, so
  the two cannot meet in practice;
* the review was right about the RULE, which was "publish unless the state is
  `off`" — an unenforced assumption, and the probe below shows it publishing
  `now_playing_state: on` (and `standby`) the moment a reading exists beside that
  state. `standby` is not even in the README, and nothing in the code stopped it.

WHAT WAS CHANGED. The publication rule now names the four values a reading may
carry (`NOW_PLAYING_STATE_VALUES`) and drops the reading when the entity state is
anything else, exactly as a state of `off` always did. The README says the same
thing in words. The two tests here walk EVERY ``MediaPlayerState`` member and
compare the README against the code constant, so neither can drift again.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import entity_stublib  # noqa: E402
import pytest  # noqa: E402
from entity_stublib import MediaPlayerState, load_entity_modules  # noqa: E402


@pytest.fixture(autouse=True)
def scoped_entity_stubs():
    """Container for the HA stubs + the loaded entity modules (t7 discipline)."""
    snapshot = entity_stublib.snapshot_modules()
    entity_stublib.install()
    yield load_entity_modules()
    entity_stublib.restore_modules(snapshot)


README = Path(__file__).resolve().parent.parent / "README.md"

#: The states the README documents, read from the file itself (never from the
#: constant under test) so a code change cannot silently rewrite the pin.
def _documented_states() -> set[str]:
    text = README.read_text(encoding="utf-8")
    row = re.search(r"^\| `now_playing_state` \|(.*)\|$", text, re.M)
    assert row, "the README's now_playing_state table row is gone"
    return {value for value in re.findall(r"`([a-z_]+)`", row.group(1)) if value != "on"}


def _player_for_state(modules, member):
    """A player whose ladder answers `member` while a reading is available.

    The state is forced by patching the ``state`` property, which is the only way
    to ask "what would be published IF the ladder ever answered this?" — the
    question the review raised. A reading is genuine: the foreground tier answers
    it through the same code path production uses (``_now_playing``).
    """
    from test_command_routing import TIVIMATE_PACKAGE, _now_playing_attrs, _state_player

    player, _manager = _state_player(modules, foreground=TIVIMATE_PACKAGE)
    player._last_channel = "101"
    media_player_cls = type(player)
    original = media_player_cls.__dict__.get("state")
    media_player_cls.state = property(lambda self, forced=member: forced)
    try:
        reading = player._now_playing()
        attrs = _now_playing_attrs(player)
    finally:
        if original is None:  # pragma: no cover - the class always declares it
            del media_player_cls.state
        else:
            media_player_cls.state = original
    return reading, attrs


class TestThePublicationRule:
    """One reading in, one row out — for EVERY member the enum has."""

    def test_the_documented_four_publish_the_reading(self, scoped_entity_stubs):
        modules = scoped_entity_stubs
        documented = {
            MediaPlayerState.PLAYING: "playing",
            MediaPlayerState.PAUSED: "paused",
            MediaPlayerState.BUFFERING: "buffering",
            MediaPlayerState.IDLE: "idle",
        }

        for member, value in documented.items():
            reading, attrs = _player_for_state(modules, member)

            assert reading, f"{value}: the fixture must have a reading to publish"
            assert attrs["now_playing_state"] == value
            assert attrs["now_playing_title"] == "101"

    def test_a_state_outside_the_documented_four_publishes_nothing(self, scoped_entity_stubs):
        """The review's case, executed: `on` publishes no row — by construction now."""
        modules = scoped_entity_stubs
        reading_available = {
            MediaPlayerState.ON: "on",
            MediaPlayerState.OFF: "off",
            MediaPlayerState.STANDBY: "standby",
        }

        for member, value in reading_available.items():
            reading, attrs = _player_for_state(modules, member)

            assert reading, f"{value}: the fixture still HAS a reading (that is the point)"
            assert attrs == {}, f"{value} must publish no now-playing row, got {attrs}"

    def test_every_enum_member_is_accounted_for(self, scoped_entity_stubs):
        """The matrix is exhaustive: no state is left untested as the enum grows."""
        modules = scoped_entity_stubs
        published = {}
        for member in MediaPlayerState:
            reading, attrs = _player_for_state(modules, member)
            published[member.value] = attrs.get("now_playing_state")

        assert set(published) == {member.value for member in MediaPlayerState}
        for value in ("playing", "paused", "buffering", "idle"):
            assert published[value] == value
        for value in ("on", "off", "standby"):
            assert published[value] is None, f"{value} published a state: {published[value]}"

    def test_the_gate_reads_the_named_constant(self, scoped_entity_stubs):
        """The rule is the constant, not a hard-coded pair of states."""
        from nsh.media_player import NOW_PLAYING_STATE_VALUES

        assert NOW_PLAYING_STATE_VALUES == ("playing", "paused", "buffering", "idle")
        assert self._published_for(scoped_entity_stubs, MediaPlayerState.ON) == {}

    @staticmethod
    def _published_for(modules, member) -> dict:
        _reading, attrs = _player_for_state(modules, member)
        return attrs


class TestTheReadmeAgrees:
    """The docs and the code say the same four values (t17/O2's anti-drift pin)."""

    def test_the_table_row_lists_exactly_the_constant(self, scoped_entity_stubs):
        from nsh.media_player import NOW_PLAYING_STATE_VALUES

        assert _documented_states() == set(NOW_PLAYING_STATE_VALUES)

    def test_the_prose_names_the_same_set_and_excludes_on(self, scoped_entity_stubs):
        text = README.read_text(encoding="utf-8")
        from nsh.media_player import NOW_PLAYING_STATE_VALUES

        paragraph = re.search(
            r"\*\*`now_playing_state` is one of FOUR values[^*]*\*\*(.*?)\n-", text, re.S
        )
        assert paragraph, "the README's now_playing_state paragraph is gone"
        body = paragraph.group(1)
        # the sentence that names the set, read apart from the sentence that
        # excludes `on`/`off` — otherwise the exclusion would look like a member
        sentence = re.search(r"full set is ([^.]+)\.", body)
        assert sentence, "the README no longer names the full set"
        named = set(re.findall(r"`([a-z_]+)`", sentence.group(1)))
        assert named == set(NOW_PLAYING_STATE_VALUES), named
        assert "`on`" in body and "not one of them" in text, (
            "the paragraph must say `on` is excluded"
        )

    def test_the_off_and_on_prose_still_says_no_rows(self, scoped_entity_stubs):
        text = README.read_text(encoding="utf-8")

        assert "**A device that reads `off` publishes none of them.**" in text
        assert "publishes none either" in text