diff --git a/custom_components/opendisplay/delivery.py b/custom_components/opendisplay/delivery.py index 8378b89..4a18d72 100644 --- a/custom_components/opendisplay/delivery.py +++ b/custom_components/opendisplay/delivery.py @@ -135,7 +135,9 @@ def __init__(self, hass: HomeAssistant, entry: OpenDisplayConfigEntry) -> None: self._pending_config_resync: bool = False self._last_error: str | None = None + self._active_upload: PendingUpload | None = None self._delivering: bool = False + self._delivery_requested: bool = False self._delivery_task: asyncio.Task[None] | None = None self._unsub_device_seen: CALLBACK_TYPE | None = None @@ -165,7 +167,9 @@ async def async_shutdown(self) -> None: except Exception: # noqa: BLE001 - shutdown must not raise _LOGGER.debug("Delivery task raised during shutdown", exc_info=True) self._delivery_task = None + self._active_upload = None self._delivering = False + self._delivery_requested = False # -- public state ------------------------------------------------------- @@ -237,10 +241,17 @@ def _on_device_seen(self) -> None: @callback def notify_device_seen(self, source: str = "ble") -> None: """Start a delivery drain if there is pending work and none in flight.""" - if self._delivering or not self._has_pending_work(): + if not self._has_pending_work(): + return + if self._delivering: + # A newer submission can arrive while an older frame is already + # streaming. Remember the request so latest-wins work is drained + # after the current BLE operation releases the single connection. + self._delivery_requested = True return _LOGGER.debug("%s: device seen (%s); starting delivery", self._address, source) self._delivering = True + self._delivery_requested = False self._delivery_task = self._entry.async_create_background_task( self._hass, self._deliver(), f"opendisplay_delivery_{self._address}" ) @@ -267,6 +278,7 @@ async def _deliver(self) -> None: # read/connect timeouts as BLETimeoutError, so a bare TimeoutError here # can only be the asyncio.timeout(DELIVERY_DEADLINE_S) firing.) self._register_attempt_failure( + self._active_upload, f"delivery deadline exceeded ({DELIVERY_DEADLINE_S:.0f}s)" ) _LOGGER.error( @@ -283,21 +295,27 @@ async def _deliver(self) -> None: except (BLEConnectionError, BLETimeoutError) as err: # Device slept mid-connect or the wake window was missed; keep the # work queued and try again on the next wake. - self._register_attempt_failure(str(err)) + self._register_attempt_failure(self._active_upload, str(err)) except (AuthenticationFailedError, AuthenticationRequiredError): # Bad/rotated key: pause the upload and prompt the user to reauth. _LOGGER.warning("%s: delivery auth failed; starting reauth", self._address) - if self._pending_upload is not None: - self._pending_upload.paused = True + upload = self._active_upload or self._pending_upload + if upload is not None and self._pending_upload is upload: + upload.paused = True self._last_error = "auth" self._entry.async_start_reauth(self._hass) self._notify_state() except OpenDisplayError as err: _LOGGER.warning("%s: delivery failed: %s", self._address, err) - self._register_attempt_failure(str(err)) + self._register_attempt_failure(self._active_upload, str(err)) finally: + delivery_requested = self._delivery_requested + self._active_upload = None self._delivering = False self._delivery_task = None + self._delivery_requested = False + if delivery_requested and self._has_pending_work(): + self.notify_device_seen("follow-up") async def _drain_once(self) -> None: """The connection + drain body (see `_deliver` for error handling).""" @@ -313,7 +331,7 @@ async def _drain_once(self) -> None: ) if ble_device is None: # The advertisement that woke us has already aged out; retry next. - self._register_attempt_failure("device not connectable") + self._register_attempt_failure(None, "device not connectable") return upload = self._pending_upload @@ -335,7 +353,9 @@ async def _drain_once(self) -> None: ) as device, ): if upload is not None and not upload.paused: + self._active_upload = upload await self._drain_upload(device, upload) + self._active_upload = None if self._pending_config_resync: await self._drain_resync(device) @@ -350,6 +370,15 @@ async def _drain_upload( ) if upload.cancel_deadline: upload.cancel_deadline() + if self._pending_upload is not upload: + # A newer image was queued while this frame was in flight. The BLE + # stream used the captured upload object, so the panel gets one whole + # old frame; keep the newer slot pending for the next drain. + _LOGGER.debug( + "%s: delivered superseded content; newer content remains queued", + self._address, + ) + return self._pending_upload = None self._last_error = None # Bump the image entity's timestamp to when the frame actually landed. @@ -434,14 +463,25 @@ def _give_up_upload(self, slot: PendingUpload, reason: str) -> None: # -- helpers ------------------------------------------------------------ - def _register_attempt_failure(self, reason: str) -> None: + def _register_attempt_failure( + self, upload: PendingUpload | None, reason: str + ) -> None: """Record a failed wake attempt and keep the work queued. After ``MAX_DELIVERY_ATTEMPTS`` failures the upload is given up and dropped (``_give_up_upload``) so a frame that can never be delivered does not retry on every wake until ``queue_timeout``. """ - upload = self._pending_upload + upload = upload or self._pending_upload + if upload is not None and self._pending_upload is not upload: + # The failed BLE operation belonged to a superseded frame. Do not + # increment attempts or set errors on the newer pending frame. + _LOGGER.debug( + "%s: superseded delivery attempt failed (%s); newer content remains queued", + self._address, + reason, + ) + return if upload is not None: upload.attempts += 1 if upload.attempts >= MAX_DELIVERY_ATTEMPTS: diff --git a/custom_components/opendisplay/services.py b/custom_components/opendisplay/services.py index b8dcd0a..08369ae 100644 --- a/custom_components/opendisplay/services.py +++ b/custom_components/opendisplay/services.py @@ -305,18 +305,6 @@ class _DeviceUnavailable(Exception): """ -# Probe budget for a probably-asleep tag: one connect attempt, short timeout. -# 5 s is >2x the observed 1-3 s connect-during-window latency (plus proxy -# slack), so a genuinely awake tag connects comfortably, while a dark ESP32 -# (radio fully off in timer deep sleep) costs at most ~5 s before queuing — -# vs the ~40 s default budget (4 attempts x 10 s). Deliberately below both the -# 10 s wake window and the 15 s freshness horizon (wake_window_s + -# FRESHNESS_SLACK_S), so a probe triggered by a just-missed advert still lands -# inside the window it is betting on. -PROBE_CONNECT_TIMEOUT_S = 5.0 -PROBE_MAX_ATTEMPTS = 1 - - async def _async_connect_and_run( hass: HomeAssistant, entry: "OpenDisplayConfigEntry", @@ -463,17 +451,7 @@ async def _async_send_image( rotate: Rotation = Rotation.ROTATE_0, use_measured_palettes: bool = False, ) -> DeliveryReceipt: - """Upload a PIL image, delivering live or queuing it for the next wake. - - Returns a receipt describing whether the frame was delivered immediately or - queued (for a sleeping device). Non-sleepy devices always deliver live and - keep the original strict-failure behavior. - """ - # Split the upload into its heavy CPU half and its BLE-I/O half. The CPU - # work (rotate + fit + dither + encode + zlib on a full frame) is offloaded - # to an executor thread so it never blocks the event loop; only the BLE - # transfer runs on the loop. This mirrors what device.upload_image() does - # internally, but that call ran _prepare_image synchronously on the loop. + """Prepare and queue a PIL image for asynchronous delivery.""" config = entry.runtime_data.device_config display_cfg = config.displays[0] if config and config.displays else None # Match upload_image(): only ask prepare_image() to build compressed data @@ -513,77 +491,36 @@ async def _async_send_image( # image entity immediately (D6), not only after a successful delivery. jpeg = await hass.async_add_executor_job(_pil_to_jpeg, img) - profile = runtime.sleep_profile + # Image services are queue-only: a successful service call means "accepted + # for delivery", never "visible on the panel". Every set-up entry should + # have a DeliveryManager; fail loudly if setup invariants are broken. manager = runtime.delivery - sleepy = manager is not None and profile.is_sleepy - - def _queue() -> DeliveryReceipt: - assert manager is not None - return manager.submit_upload( - prepared=prepared, - refresh_mode=refresh_mode, - partial_state=state, - use_measured_palettes=use_measured_palettes, - preview_jpeg=jpeg, - device_id=device_id, - ) - - async def _upload(device: OpenDisplayDevice) -> None: - await device.upload_prepared_image( - prepared, refresh_mode=refresh_mode, state=state + if manager is None: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="upload_error", + translation_placeholders={"error": "delivery manager unavailable"}, ) - # Freshness gate: a probably-asleep tag will not usually answer a live - # connect. Instead of queuing blind, spend one short connect attempt (the - # "probe") in case the tag is actually awake: its wake adverts may have - # been missed by the scanner, and Silabs tags advertise continuously even - # when their power config looks sleepy. A dark ESP32 costs at most - # ~PROBE_CONNECT_TIMEOUT_S before falling back to the queue. HA drops the - # connectable BLEDevice ~3-5 min after the last advert, so a long-dark or - # never-seen tag short-circuits to the queue at near-zero cost (no - # BLEDevice -> _DeviceUnavailable without any radio traffic). - probing = False - connect_kwargs: dict[str, Any] = {} - if sleepy: - last_seen = ( - runtime.coordinator.data.last_seen if runtime.coordinator.data else None - ) - if profile.probably_asleep(last_seen): - if not profile.probe_before_queue: - return _queue() - probing = True - connect_kwargs = { - "connect_timeout": PROBE_CONNECT_TIMEOUT_S, - "max_attempts": PROBE_MAX_ATTEMPTS, - } + # DeliveryManager owns latest-wins replacement, queue expiry, preview state, + # retries, and delivered/expired events. The service does not open BLE. + receipt = manager.submit_upload( + prepared=prepared, + refresh_mode=refresh_mode, + partial_state=state, + use_measured_palettes=use_measured_palettes, + preview_jpeg=jpeg, + device_id=device_id, + ) - try: - await _async_connect_and_run( - hass, - entry, - _upload, - use_measured_palettes=use_measured_palettes, - reraise_ble=sleepy, - **connect_kwargs, - ) - except _DeviceUnavailable: - # The device was dark, dropped, or refused the link; defer. - receipt = _queue() - if probing: - # Race: a wake advert arriving DURING the failed probe found no - # pending work (nothing queued yet), so no drain started. If the - # tag now looks fresh, kick a drain instead of waiting out a full - # sleep cycle. notify_device_seen is a no-op while one is running. - last_seen = ( - runtime.coordinator.data.last_seen if runtime.coordinator.data else None - ) - if not profile.probably_asleep(last_seen): - assert manager is not None - manager.notify_device_seen("post-probe") - return receipt + # Always ask the queue to drain in the background after a new submission. + # The service does not decide whether the device is currently reachable; the + # drain path owns reachability checks and leaves the frame queued if there is + # no connectable device yet. "queued-submit" is only an internal source label + # for logging/diagnostics, not a Home Assistant or BLE protocol term. + manager.notify_device_seen("queued-submit") - async_dispatcher_send(hass, f"{SIGNAL_IMAGE_UPDATED}_{entry.unique_id}", jpeg) - return DeliveryReceipt(status="delivered", expires_at=None) + return receipt def _receipt_response(receipt: DeliveryReceipt) -> ServiceResponse: @@ -623,12 +560,10 @@ async def _async_upload_image(call: ServiceCall) -> ServiceResponse: translation_placeholders={"url": image_data}, ) - # Latest-wins for upload_image specifically: a newer upload cancels an - # older still-running one instead of queuing behind it (an automation - # pushing a fresh snapshot supersedes the stale one). This composes with the - # per-MAC ble_lock without deadlocking: the cancel + await below happens - # before _async_send_image acquires the lock, so the cancelled task releases - # the lock (its `async with ble_lock` unwinds) before this one takes it. + # Latest-wins for upload_image specifically: a newer upload cancels older + # still-running download/prepare work before it can replace the queue slot. + # Once a frame reaches DeliveryManager.submit_upload(), the queue's own + # latest-wins behavior supersedes the previous pending frame. current = asyncio.current_task() if (prev := entry.runtime_data.upload_task) is not None and not prev.done(): prev.cancel() @@ -997,4 +932,4 @@ def async_setup_services(hass: HomeAssistant) -> None: supports_response=SupportsResponse.OPTIONAL, ) hass.services.async_register(DOMAIN, "activate_led", _async_activate_led, schema=SCHEMA_ACTIVATE_LED) - hass.services.async_register(DOMAIN, "activate_buzzer", _async_activate_buzzer, schema=SCHEMA_ACTIVATE_BUZZER) \ No newline at end of file + hass.services.async_register(DOMAIN, "activate_buzzer", _async_activate_buzzer, schema=SCHEMA_ACTIVATE_BUZZER) diff --git a/tests/test_delivery.py b/tests/test_delivery.py index 994d752..fab99fe 100644 --- a/tests/test_delivery.py +++ b/tests/test_delivery.py @@ -67,13 +67,13 @@ def _make_env(profile=None, entry_data=None, last_seen=None, options=None): return hass, entry, coordinator -def _submit(mgr, device_id="dev1"): +def _submit(mgr, device_id="dev1", data=b"img", preview_jpeg=b"jpeg"): return mgr.submit_upload( - prepared=(b"img", None, object()), + prepared=(data, None, object()), refresh_mode=RefreshMode.FULL, partial_state=MagicMock(), use_measured_palettes=False, - preview_jpeg=b"jpeg", + preview_jpeg=preview_jpeg, device_id=device_id, ) @@ -216,6 +216,87 @@ async def test_drain_delivers_upload(): assert EVENT_CONTENT_DELIVERED in fired +@pytest.mark.asyncio +async def test_drain_success_does_not_clear_newer_upload_queued_mid_send(): + """A completed superseded upload must not clear the newer pending slot.""" + hass, entry, _ = _make_env() + scheduled = [] + + def _capture(_hass, coro, _name): + scheduled.append(coro) + if len(scheduled) > 1: + coro.close() # follow-up drain is asserted, not run in this test + return MagicMock() + + entry.async_create_background_task = MagicMock(side_effect=_capture) + device = MagicMock() + + async def _upload(_prepared, **_kwargs): + _submit(mgr, device_id="b", data=b"b", preview_jpeg=b"b-jpeg") + mgr.notify_device_seen("queued-submit") + + device.upload_prepared_image = AsyncMock(side_effect=_upload) + with ( + patch.object(delivery_mod, "async_call_later", return_value=MagicMock()), + patch.object(delivery_mod, "async_dispatcher_send"), + patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()), + patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_fake_device_ctx(device)), + ): + mgr = DeliveryManager(hass, entry) + _submit(mgr, device_id="a", data=b"a", preview_jpeg=b"a-jpeg") + mgr.notify_device_seen("ble") + await scheduled[0] + + device.upload_prepared_image.assert_awaited_once() + assert device.upload_prepared_image.await_args.args[0][0] == b"a" + assert mgr._pending_upload is not None + assert mgr._pending_upload.device_id == "b" + assert mgr._pending_upload.attempts == 0 + assert mgr.state.last_error is None + assert len(scheduled) == 2 # B requested a follow-up drain after A finished. + + +@pytest.mark.asyncio +async def test_drain_failure_does_not_penalize_newer_upload_queued_mid_send(): + """A failed superseded upload must not count as an attempt against the replacement.""" + hass, entry, _ = _make_env() + scheduled = [] + + def _capture(_hass, coro, _name): + scheduled.append(coro) + if len(scheduled) > 1: + coro.close() # follow-up drain is asserted, not run in this test + return MagicMock() + + entry.async_create_background_task = MagicMock(side_effect=_capture) + device = MagicMock() + + async def _upload(_prepared, **_kwargs): + _submit(mgr, device_id="b", data=b"b", preview_jpeg=b"b-jpeg") + mgr.notify_device_seen("queued-submit") + raise BLEConnectionError("lost link") + + device.upload_prepared_image = AsyncMock(side_effect=_upload) + with ( + patch.object(delivery_mod, "async_call_later", return_value=MagicMock()), + patch.object(delivery_mod, "async_dispatcher_send"), + patch.object(delivery_mod, "async_ble_device_from_address", return_value=MagicMock()), + patch.object(delivery_mod, "OpenDisplayDevice", side_effect=_fake_device_ctx(device)), + ): + mgr = DeliveryManager(hass, entry) + _submit(mgr, device_id="a", data=b"a", preview_jpeg=b"a-jpeg") + mgr.notify_device_seen("ble") + await scheduled[0] + + device.upload_prepared_image.assert_awaited_once() + assert device.upload_prepared_image.await_args.args[0][0] == b"a" + assert mgr._pending_upload is not None + assert mgr._pending_upload.device_id == "b" + assert mgr._pending_upload.attempts == 0 + assert mgr.state.last_error is None + assert len(scheduled) == 2 # B requested a follow-up drain after A failed. + + @pytest.mark.asyncio async def test_drain_ble_failure_keeps_slot_and_counts_attempt(): hass, entry, _ = _make_env() diff --git a/tests/test_services.py b/tests/test_services.py index 2065e97..4d6370d 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -1,33 +1,23 @@ -"""Unit tests for the _async_send_image sleep gate and probe-before-queue. +"""Unit tests for the image-service queue contract. -Mirrors the test_delivery.py approach: no Home Assistant test harness; the -service module's HA/BLE touchpoints (``prepare_image``, ``_pil_to_jpeg``, -``async_dispatcher_send``, ``async_ble_device_from_address`` and -``OpenDisplayDevice``) are patched in the services module namespace. +The service module's HA/BLE touchpoints are patched in its own namespace so the +tests do not need a full Home Assistant harness. """ import asyncio import time from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import MagicMock, patch -from opendisplay import BLEConnectionError, RefreshMode, DitherMode +from opendisplay import DitherMode, RefreshMode from PIL import Image as PILImage import pytest +from homeassistant.exceptions import HomeAssistantError + from custom_components.opendisplay import services as services_mod -from custom_components.opendisplay.const import ( - CONF_BLOCKS_PER_ACK, - CONF_MAX_QUEUE_SIZE, - DEFAULT_BLOCKS_PER_ACK, - DEFAULT_MAX_QUEUE_SIZE, -) from custom_components.opendisplay.delivery import DeliveryReceipt -from custom_components.opendisplay.services import ( - PROBE_CONNECT_TIMEOUT_S, - PROBE_MAX_ATTEMPTS, - _async_send_image, -) +from custom_components.opendisplay.services import _async_send_image from custom_components.opendisplay.sleep import SleepProfile ADDRESS = "AA:BB:CC:DD:EE:FF" @@ -46,16 +36,17 @@ def _profile(**overrides): return SleepProfile.create(**params) -def _make_env(profile=None, last_seen=None, options=None): +def _make_env(profile=None, last_seen=None, manager=None): """Return (hass, entry, coordinator, manager) with a fake runtime.""" profile = profile or _profile() coordinator = MagicMock() coordinator.data = SimpleNamespace(last_seen=last_seen) - manager = MagicMock() - manager.submit_upload = MagicMock( - return_value=DeliveryReceipt(status="queued", expires_at=123.0) - ) - manager.notify_device_seen = MagicMock() + if manager is None: + manager = MagicMock() + manager.submit_upload = MagicMock( + return_value=DeliveryReceipt(status="queued", expires_at=123.0) + ) + manager.notify_device_seen = MagicMock() runtime = SimpleNamespace( coordinator=coordinator, sleep_profile=profile, @@ -68,7 +59,7 @@ def _make_env(profile=None, last_seen=None, options=None): entry.unique_id = ADDRESS entry.runtime_data = runtime entry.data = {} # no encryption key - entry.options = options if options is not None else {} + entry.options = {} # no custom transfer options hass = MagicMock() async def _executor_job(fn, *args): @@ -78,27 +69,6 @@ async def _executor_job(fn, *args): return hass, entry, coordinator, manager -def _device_ctx_factory(device=None, exc=None, on_enter=None): - """MagicMock construct-recording factory for OpenDisplayDevice. - - Produces an async context manager that yields ``device``, or raises - ``exc`` from ``__aenter__`` (after calling ``on_enter`` if given). - """ - - class _Ctx: - async def __aenter__(self): - if on_enter is not None: - on_enter() - if exc is not None: - raise exc - return device - - async def __aexit__(self, *exc_info): - return False - - return MagicMock(side_effect=lambda **kwargs: _Ctx()) - - def _send(hass, entry): img = PILImage.new("RGB", (1, 1)) return _async_send_image( @@ -110,220 +80,103 @@ def _send(hass, entry): ) -def _patches(od_factory, ble_device="ble-device"): +def _patches(): """Common patch set for a _async_send_image drive.""" return ( patch.object( services_mod, "prepare_image", return_value=(b"img", None, object()) ), patch.object(services_mod, "_pil_to_jpeg", return_value=b"jpeg"), - patch.object(services_mod, "async_dispatcher_send"), - patch.object( - services_mod, "async_ble_device_from_address", return_value=ble_device - ), - patch.object(services_mod, "OpenDisplayDevice", od_factory), + patch.object(services_mod, "OpenDisplayDevice"), ) @pytest.mark.asyncio -async def test_probe_success_delivers(): - """Stale device + probe on: one reduced-budget attempt that succeeds.""" - hass, entry, _, manager = _make_env(last_seen=None) # never seen -> asleep - device = MagicMock() - device.upload_prepared_image = AsyncMock() - od = _device_ctx_factory(device=device) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: - receipt = await _send(hass, entry) - - assert receipt.status == "delivered" - od.assert_called_once() - kwargs = od.call_args.kwargs - assert kwargs["timeout"] == PROBE_CONNECT_TIMEOUT_S - assert kwargs["max_attempts"] == PROBE_MAX_ATTEMPTS - manager.submit_upload.assert_not_called() - - -@pytest.mark.asyncio -async def test_probe_failure_queues(): - """Stale device + probe on: the failed attempt falls back to the queue.""" +async def test_image_send_always_queues_and_does_not_open_live_connection(): hass, entry, _, manager = _make_env(last_seen=None) - od = _device_ctx_factory(exc=BLEConnectionError("dark")) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: + p1, p2, p3 = _patches() + with p1, p2, p3 as od: receipt = await _send(hass, entry) assert receipt.status == "queued" - od.assert_called_once() manager.submit_upload.assert_called_once() - - -@pytest.mark.asyncio -async def test_probe_disabled_queues_immediately(): - """probe_before_queue=False restores the old queue-without-connect gate.""" - hass, entry, _, manager = _make_env( - profile=_profile(probe_before_queue=False), last_seen=None - ) - od = _device_ctx_factory() - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4 as ble_lookup, p5: - receipt = await _send(hass, entry) - - assert receipt.status == "queued" - ble_lookup.assert_not_called() od.assert_not_called() - manager.submit_upload.assert_called_once() - - -@pytest.mark.asyncio -async def test_not_sleepy_full_budget(): - """Non-sleepy devices keep the library's default connect budget.""" - hass, entry, _, manager = _make_env( - profile=_profile(sleep_mode="off"), last_seen=None - ) - device = MagicMock() - device.upload_prepared_image = AsyncMock() - od = _device_ctx_factory(device=device) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: - receipt = await _send(hass, entry) - - assert receipt.status == "delivered" - kwargs = od.call_args.kwargs - assert "timeout" not in kwargs - assert "max_attempts" not in kwargs - manager.submit_upload.assert_not_called() @pytest.mark.asyncio -async def test_probe_no_ble_device_queues_cheaply(): - """No connectable BLEDevice: the probe short-circuits to the queue.""" +async def test_queue_submission_threads_prepared_frame_state_and_refresh_mode(): hass, entry, _, manager = _make_env(last_seen=None) - od = _device_ctx_factory() - p1, p2, p3, p4, p5 = _patches(od, ble_device=None) - with p1, p2, p3, p4, p5: - receipt = await _send(hass, entry) + p1, p2, p3 = _patches() + with p1, p2, p3: + await _send(hass, entry) - assert receipt.status == "queued" - od.assert_not_called() - manager.submit_upload.assert_called_once() + kwargs = manager.submit_upload.call_args.kwargs + assert kwargs["prepared"][0] == b"img" + assert kwargs["refresh_mode"] is RefreshMode.FULL + assert kwargs["partial_state"] is entry.runtime_data.partial_state + assert kwargs["use_measured_palettes"] is False + assert kwargs["preview_jpeg"] == b"jpeg" @pytest.mark.asyncio -async def test_post_probe_fresh_advert_triggers_drain(): - """A wake advert landing during the failed probe kicks an immediate drain.""" - hass, entry, coordinator, manager = _make_env(last_seen=None) - - def _advert_arrives(): - coordinator.data = SimpleNamespace(last_seen=time.time()) - - od = _device_ctx_factory( - exc=BLEConnectionError("dropped"), on_enter=_advert_arrives - ) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: +async def test_fresh_sleepy_device_kicks_background_drain(): + hass, entry, _, manager = _make_env(last_seen=time.time()) + p1, p2, p3 = _patches() + with p1, p2, p3: receipt = await _send(hass, entry) assert receipt.status == "queued" - manager.notify_device_seen.assert_called_once_with("post-probe") + manager.notify_device_seen.assert_called_once_with("queued-submit") @pytest.mark.asyncio -async def test_post_probe_still_stale_no_drain_kick(): - """No advert during the failed probe: wait for the natural next wake.""" +async def test_stale_sleepy_device_still_kicks_background_drain(): hass, entry, _, manager = _make_env(last_seen=None) - od = _device_ctx_factory(exc=BLEConnectionError("dark")) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: + p1, p2, p3 = _patches() + with p1, p2, p3: receipt = await _send(hass, entry) assert receipt.status == "queued" - manager.notify_device_seen.assert_not_called() + manager.notify_device_seen.assert_called_once_with("queued-submit") @pytest.mark.asyncio -async def test_fresh_device_failure_no_probe_kwargs(): - """Sleepy but recently seen: full budget, queue on failure, no drain kick.""" +async def test_image_send_does_not_check_ble_reachability(): hass, entry, _, manager = _make_env(last_seen=time.time()) - od = _device_ctx_factory(exc=BLEConnectionError("dropped")) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: + p1, p2, p3 = _patches() + with ( + p1, + p2, + p3, + patch.object(services_mod, "async_ble_device_from_address") as ble_lookup, + ): receipt = await _send(hass, entry) assert receipt.status == "queued" - kwargs = od.call_args.kwargs - assert "timeout" not in kwargs - assert "max_attempts" not in kwargs - manager.notify_device_seen.assert_not_called() + ble_lookup.assert_not_called() + manager.notify_device_seen.assert_called_once_with("queued-submit") @pytest.mark.asyncio -async def test_pipe_kwargs_default(): - """No options set: library sliding-window defaults are threaded through.""" +async def test_not_sleepy_device_kicks_background_drain_when_connectable(): hass, entry, _, manager = _make_env( profile=_profile(sleep_mode="off"), last_seen=None ) - device = MagicMock() - device.upload_prepared_image = AsyncMock() - od = _device_ctx_factory(device=device) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: - receipt = await _send(hass, entry) - - assert receipt.status == "delivered" - kwargs = od.call_args.kwargs - assert kwargs["blocks_per_ack"] == DEFAULT_BLOCKS_PER_ACK - assert kwargs["max_queue_size"] == DEFAULT_MAX_QUEUE_SIZE - manager.submit_upload.assert_not_called() - - -@pytest.mark.asyncio -async def test_pipe_kwargs_custom(): - """Configured options reach the OpenDisplayDevice constructor.""" - hass, entry, _, _ = _make_env( - profile=_profile(sleep_mode="off"), - last_seen=None, - options={CONF_BLOCKS_PER_ACK: 4, CONF_MAX_QUEUE_SIZE: 1}, - ) - device = MagicMock() - device.upload_prepared_image = AsyncMock() - od = _device_ctx_factory(device=device) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: + p1, p2, p3 = _patches() + with p1, p2, p3: receipt = await _send(hass, entry) - assert receipt.status == "delivered" - kwargs = od.call_args.kwargs - assert kwargs["blocks_per_ack"] == 4 - assert kwargs["max_queue_size"] == 1 + assert receipt.status == "queued" + manager.notify_device_seen.assert_called_once_with("queued-submit") @pytest.mark.asyncio -async def test_live_send_passes_state_and_refresh_mode(): - """Live-send path pins the pipe-partial signature: the entry's PartialState - and the requested refresh_mode reach upload_prepared_image. - - The library keeps the ``upload_prepared_image(prepared, refresh_mode=, - state=)`` contract for pipe-partial, so the integration needs no behavior - change -- this guards that the live path keeps threading both kwargs. - """ - hass, entry, _, manager = _make_env( - profile=_profile(sleep_mode="off"), last_seen=None - ) - device = MagicMock() - device.upload_prepared_image = AsyncMock() - od = _device_ctx_factory(device=device) - p1, p2, p3, p4, p5 = _patches(od) - with p1, p2, p3, p4, p5: - receipt = await _send(hass, entry) - - assert receipt.status == "delivered" - device.upload_prepared_image.assert_awaited_once() - call = device.upload_prepared_image.await_args - # _send uses RefreshMode.FULL, so the service re-baselines partial_state to - # a fresh PartialState and hands that same object to the library. - assert call.kwargs["state"] is entry.runtime_data.partial_state - assert call.kwargs["refresh_mode"] is RefreshMode.FULL +async def test_missing_delivery_manager_raises_upload_error(): + hass, entry, _, _ = _make_env(manager=None) + entry.runtime_data.delivery = None + p1, p2, p3 = _patches() + with p1, p2, p3, pytest.raises(HomeAssistantError): + await _send(hass, entry) if __name__ == "__main__":