"""t66: the opt-in screen-capture IMAGE entity. WHAT IS PROVEN HERE, AND WHAT IS NOT ------------------------------------ Proven: the bytes that reach Home Assistant are REAL PNG bytes — ``PNG_B64`` below is a real 2x1 PNG, base64'd on the wire exactly the way ``screencap -p | base64`` sends it — the entity hands HA ``bytes`` and never a ``str``, a disconnected link or a bad answer makes the entity UNAVAILABLE instead of serving a blank/zero-byte image, one capture is coalesced and reused so a dashboard cannot hammer the device, and the option defaults OFF so no entity exists until the user opts in. NOT proven: anything about a real Shield. No device was available in this environment, so ``screencap``'s flags, the device's base64 and — the important one — how the persistent ADB socket copes with a few hundred kilobytes of base64 in a single ``shell()`` call are INFERRED, not measured. ``FakeAdbPoller`` answers the way a working Shield should. """ import asyncio import base64 import json import sys from datetime import datetime from types import SimpleNamespace import entity_stublib import pytest from conftest import _SHIELD_DIR from entity_stublib import install, load_entity_modules #: A REAL 2x1 PNG (one red pixel, one green pixel) — 72 bytes, base64. It is the #: wire form the device sends, so the fixture exercises the real decode path #: rather than an invented one. PNG_B64 = ( "iVBORw0KGgoAAAANSUhEUgAAAAIAAAABCAIAAAB7QOjdAAAAD0lEQVR42mP4z8DA8J8BAAf/" "Af+4BDXgAAAAAElFTkSuQmCC" ) PNG_BYTES = base64.b64decode(PNG_B64) PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" def wrapped(text: str, width: int = 76) -> str: """``base64``'s own line wrapping — the device wraps, and that is not corruption.""" return "\n".join(text[i : i + width] for i in range(0, len(text), width)) + "\n" @pytest.fixture(autouse=True) def scoped_entity_stubs(): """Container for the HA stubs + loaded entity modules (t7 discipline).""" snapshot = entity_stublib.snapshot_modules() install() yield load_entity_modules() entity_stublib.restore_modules(snapshot) @pytest.fixture(name="modules") def load_modules_fixture(): return load_entity_modules() class FakeAdbPoller: """The ADB seam the entity is allowed to use: ``connected`` + ``shell``.""" def __init__(self, answer=None, *, connected=True, delay=0.0): self.connected = connected self.answer = answer self.delay = delay self.calls: list[str] = [] async def shell(self, cmd: str): self.calls.append(cmd) if self.delay: await asyncio.sleep(self.delay) if isinstance(self.answer, Exception): raise self.answer if callable(self.answer): return self.answer(cmd) return self.answer class StrictManager: """Only ``adb_poller`` may be touched — nothing else exists on purpose. "Never a state source" is a claim about what the capture path reads and writes: it reads the ADB poller for a screen grab and writes nothing at all. A manager that raises on every other attribute turns a future reach-in (a store write, a state push) into a failing test instead of a quiet coupling between screen capture and the state pipeline. """ def __init__(self, poller): self.adb_poller = poller def __getattr__(self, name): raise AssertionError(f"the screen-capture path touched manager.{name}") def make_entry(options=None, *, entry_id="e1", title="Shield"): return SimpleNamespace( entry_id=entry_id, title=title, options=dict(options or {}), runtime_data=None ) def make_entity(modules, poller, options=None): """The entity over a fake manager on the entry, plus the entry it belongs to.""" entry = make_entry(options or {"screen_capture_enabled": True}) if poller is not None: entry.runtime_data = StrictManager(poller) entity = modules["image"].NvidiaShieldScreenImage(SimpleNamespace(), entry) return entity, entry def fake_clock(modules, monkeypatch, start=1000.0): """Freeze ``time.monotonic`` for the capture-window tests. Returns the cell.""" now = [start] monkeypatch.setattr( modules["image"], "time", SimpleNamespace(monotonic=lambda: now[0]) ) return now # ── the decode path ────────────────────────────────────────────────────────── class TestScreenCaptureDecode: """``decode_screen_capture`` — the one place an image could be corrupted.""" def test_a_real_png_survives_the_wire_format(self, modules): decode = modules["image"].decode_screen_capture data = decode(wrapped(PNG_B64)) assert data == PNG_BYTES # byte-for-byte, wrapping and all assert data.startswith(PNG_SIGNATURE) assert isinstance(data, bytes) # not a str: this is base64-decoded @pytest.mark.parametrize( "answer", [ None, # the transport never answered (AdbClientBase.shell's failure) "", # an empty answer is not an image "not base64 at all !!", base64.b64encode(b"hello world, definitely not a png").decode(), # a signature and nothing else: not a PNG (the IHDR chunk is missing) base64.b64encode(PNG_SIGNATURE + b"\x00" * 8).decode(), ], ) def test_everything_that_is_not_a_png_is_rejected(self, modules, answer): assert modules["image"].decode_screen_capture(answer) is None def test_a_device_error_line_is_not_an_image(self, modules): polluted = "error: no display\n" + wrapped(PNG_B64) assert modules["image"].decode_screen_capture(polluted) is None # ── the entity ─────────────────────────────────────────────────────────────── class TestScreenCaptureEntity: async def test_the_entity_serves_real_png_bytes_and_never_a_str(self, modules): poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) data = await entity.async_image() assert data == PNG_BYTES assert data.startswith(PNG_SIGNATURE) assert not isinstance(data, str) assert entity.available is True assert entity.capture_status == "ok" assert entity.extra_state_attributes["capture_status"] == "ok" async def test_the_command_is_the_base64_route_and_needs_no_device_file(self, modules): """Route 2 of the module docstring, and the reason it was chosen.""" poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) await entity.async_image() assert poller.calls == ["screencap -p | base64"] # No device-side artifact exists, so there is nothing to clean up... assert "/sdcard" not in poller.calls[0] assert "rm " not in poller.calls[0] async def test_a_disconnected_link_is_unavailable_and_is_never_asked(self, modules): poller = FakeAdbPoller(wrapped(PNG_B64), connected=False) entity, _entry = make_entity(modules, poller) assert entity.available is False assert entity.capture_status == "adb_disconnected" assert await entity.async_image() is None assert poller.calls == [] # a dead link is not poked for a screenshot async def test_adb_disabled_has_its_own_reason(self, modules): entry = make_entry({"screen_capture_enabled": True}) manager = SimpleNamespace(adb_poller=None) entry.runtime_data = manager entity = modules["image"].NvidiaShieldScreenImage(SimpleNamespace(), entry) assert entity.available is False assert entity.capture_status == "adb_disabled" assert await entity.async_image() is None async def test_an_empty_answer_is_unavailable_not_a_blank_image(self, modules): poller = FakeAdbPoller(None) entity, _entry = make_entity(modules, poller) assert await entity.async_image() is None assert entity.available is False assert entity.capture_status == "capture_failed" async def test_a_transport_that_raises_is_a_failure_not_a_crash(self, modules): poller = FakeAdbPoller(RuntimeError("socket closed under us")) entity, _entry = make_entity(modules, poller) assert await entity.async_image() is None assert entity.available is False assert entity.capture_status == "capture_failed" async def test_a_failed_capture_recovers_on_the_next_request(self, modules): poller = FakeAdbPoller(None) entity, _entry = make_entity(modules, poller) assert await entity.async_image() is None assert entity.available is False poller.answer = wrapped(PNG_B64) # the device came back assert await entity.async_image() == PNG_BYTES assert entity.available is True assert entity.capture_status == "ok" async def test_one_capture_is_reused_inside_the_window(self, modules, monkeypatch): """A dashboard repaint is not a screen grab per repaint.""" now = fake_clock(modules, monkeypatch) poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) first = await entity.async_image() now[0] += modules["image"].CAPTURE_REUSE_SECONDS - 1.0 second = await entity.async_image() assert first == second == PNG_BYTES assert len(poller.calls) == 1 # one shell, two image requests async def test_the_window_expires_so_a_later_view_refreshes(self, modules, monkeypatch): now = fake_clock(modules, monkeypatch) poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) await entity.async_image() now[0] += modules["image"].CAPTURE_REUSE_SECONDS + 1.0 await entity.async_image() assert len(poller.calls) == 2 async def test_an_explicit_refresh_skips_the_reuse_window(self, modules, monkeypatch): """``homeassistant.update_entity`` means "grab it now".""" fake_clock(modules, monkeypatch) poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) await entity.async_image() await entity.async_update() assert len(poller.calls) == 2 async def test_concurrent_requests_share_one_capture(self, modules): """Parallel image fetches must not leak ADB shells.""" poller = FakeAdbPoller(wrapped(PNG_B64), delay=0.01) entity, _entry = make_entity(modules, poller) results = await asyncio.gather( entity.async_image(), entity.async_image(), entity.async_image() ) assert list(results) == [PNG_BYTES, PNG_BYTES, PNG_BYTES] assert len(poller.calls) == 1 async def test_capture_reads_the_adb_poller_and_drives_nothing(self, modules): """The strict manager raises if the capture path touches anything else.""" poller = FakeAdbPoller(wrapped(PNG_B64)) entity, _entry = make_entity(modules, poller) # StrictManager assert await entity.async_image() == PNG_BYTES # ...and the state HA would record is the timestamp, never the bytes assert entity.image_last_updated is not None assert isinstance(entity.image_last_updated, datetime) assert entity.image_last_updated.tzinfo is not None async def test_image_last_updated_is_stamped_only_by_a_success(self, modules, monkeypatch): fake_clock(modules, monkeypatch) poller = FakeAdbPoller(None) entity, _entry = make_entity(modules, poller) assert entity.image_last_updated is None # nothing captured yet assert await entity.async_image() is None assert entity.image_last_updated is None # a failure stamps nothing poller.answer = wrapped(PNG_B64) await entity.async_image() stamped = entity.image_last_updated assert stamped is not None poller.answer = None # a later failure must not move it await entity.async_image() assert entity.image_last_updated is stamped def test_entity_hygiene(self, modules): """Naming, content type, recorder hygiene, and the no-poll rule.""" poller = FakeAdbPoller(wrapped(PNG_B64)) entity, entry = make_entity(modules, poller) cls = type(entity) assert cls._attr_has_entity_name is True assert entity.translation_key == "screen_capture" assert entity.content_type == "image/png" # HA polls image entities never by default; this one must not either assert cls._attr_should_poll is False assert entity.should_poll is False assert entity.unique_id == f"nvidia_shield_{entry.entry_id}_screen_capture" assert entity.device_info["identifiers"] == {("nvidia_shield", entry.entry_id)} # the status changes with a capture, so it stays out of the database, and # no attribute ever carries the image bytes assert "capture_status" in cls._unrecorded_attributes for value in entity.extra_state_attributes.values(): assert not isinstance(value, (bytes, bytearray)) def test_the_platform_is_forwarded(self, modules): """The new platform is in the entry's list, and the manifest still parses.""" platforms = [str(p) for p in sys.modules["nsh.init"].PLATFORMS] assert "image" in platforms assert platforms.count("image") == 1 manifest = json.loads( (_SHIELD_DIR / "manifest.json").read_text(encoding="utf-8") ) # HA discovers platforms from the forwarded list; a "platforms" key in the # manifest is not a thing (and adding one would be a silent no-op). assert "platforms" not in manifest assert "adb-shell>=0.4.0" in manifest["requirements"] # ── opt-in ─────────────────────────────────────────────────────────────────── class TestScreenCaptureOptIn: """Default OFF: no option, no entity, no cost.""" async def _setup(self, modules, options): added: list = [] entry = make_entry(options) entry.runtime_data = StrictManager(FakeAdbPoller(wrapped(PNG_B64))) await modules["image"].async_setup_entry(SimpleNamespace(), entry, added.extend) return added async def test_the_option_defaults_off_and_creates_no_entity(self, modules): assert await self._setup(modules, {}) == [] assert sys.modules["nsh.const"].CONF_SCREEN_CAPTURE_ENABLED == ( "screen_capture_enabled" ) assert await self._setup(modules, {"screen_capture_enabled": False}) == [] async def test_switching_it_on_creates_exactly_one_entity(self, modules): added = await self._setup(modules, {"screen_capture_enabled": True}) assert len(added) == 1 assert type(added[0]).__name__ == "NvidiaShieldScreenImage" async def test_an_entry_without_a_manager_is_skipped_not_crashed(self, modules): added: list = [] entry = make_entry({"screen_capture_enabled": True}) entry.runtime_data = None await modules["image"].async_setup_entry(SimpleNamespace(), entry, added.extend) assert added == []