From 76b5a0eb305c6ceab3cb8d4f633482063599be65 Mon Sep 17 00:00:00 2001 From: Chuck <33324927+ChuckBuilds@users.noreply.github.com> Date: Sat, 3 Oct 2026 21:09:03 -0400 Subject: [PATCH] fix(ipc): ticks carry the volatile timestamps, so current-status stays known /api/v3/display/current-status answered mode: null over the control socket once the same mode had been on screen for more than two minutes (a live game, Vegas, a single plugin). On ledpi it returned nulls in 1059/1059 samples over 90 minutes while a direct state.get showed the mode with last_updated 0.15 s old. The hub's version ignores the keys a publisher names as volatile, so display.last_updated moving alone never sent a state event, and the subscriber's keepalive tick carried only loop and served_at. The web's copy kept the last_updated of the last real change, and current_status's 120 s rule called it unknown. The plugins section had the same problem with published_at: /plugins/state read stale after 180 s with nothing changing. The short changed:false answer (what a tick is) now carries volatile: {section: {key: value}}, the current values of those keys, and StateSubscription merges them into its copy. Freshness still follows the writers, so the verdicts from #726 are unchanged: a render thread that stops publishing reads stalled at 60 s and unknown at 120 s, a runtime publisher that stops goes stale, and a quiet subscription falls back to the cache. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 19 +++++ docs/IPC_CONTROL_SOCKET.md | 18 ++-- src/ipc/client.py | 32 ++++++- src/ipc/contract.py | 21 +++-- src/ipc/server.py | 33 +++++++- test/test_ipc_state_stream.py | 82 ++++++++++++++++++ test/test_state_stream_readers.py | 134 ++++++++++++++++++++++++++++++ 7 files changed, 322 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fa3a3862..35ac91820 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -576,6 +576,25 @@ policies are unchanged. even when the mode name stays the same. The display republished its current state only on a mode change or every 30 s, so `is_display_active` and `on_demand_active` could be up to 30 s out of date. +- `/api/v3/display/current-status` no longer answers `mode: null` over the + control socket (#735) once the same mode has been on screen for more than + two minutes: a live game under live priority, Vegas, or a single plugin. + On ledpi it returned nulls in every sample for 90 minutes while the + display was live. The state stream's version leaves out the timestamps + that move on every publish, and a subscriber's keepalive tick carried only + the loop's heartbeat. So the web interface's copy kept the + `display.last_updated` of the last real change, and the reader's 120 s + rule called it unknown. The plugin runtime section had the same problem: + with no plugin changing state, `/plugins/state` and the `runtime` in + `/plugins/installed` read `stale` after 180 s. A tick (and a `state.get` + answer with `since`) now carries `volatile`: the current values of those + timestamps (`display.last_updated`, `on_demand.last_updated` and + `remaining`, `plugins.published_at`), and the subscription merges them + into its copy. The verdicts are unchanged. A render thread that stops + publishing still reads as `stalled` after 60 s and as unknown after 120 s, + a runtime publisher that stops still goes `stale`, and a subscription that + goes quiet still falls back to the cache. The cache path's 120 s rule is + unchanged. ### Scrolling diff --git a/docs/IPC_CONTROL_SOCKET.md b/docs/IPC_CONTROL_SOCKET.md index 56f98e716..2ec19e1f9 100644 --- a/docs/IPC_CONTROL_SOCKET.md +++ b/docs/IPC_CONTROL_SOCKET.md @@ -146,7 +146,7 @@ connection until either side hangs up: ```json {"v": 1, "id": "", "event": "state", "result": {...a state snapshot...}} -{"v": 1, "id": "", "event": "tick", "result": {"version": 7, "epoch": "…", "pid": 812, "served_at": 1790000000.1, "changed": false, "loop": {...}}} +{"v": 1, "id": "", "event": "tick", "result": {"version": 7, "epoch": "…", "pid": 812, "served_at": 1790000000.1, "changed": false, "loop": {...}, "volatile": {"display": {"last_updated": 1790000000.0}, "...": "..."}}} ``` An event has `event` where a response has `ok`, which is how a reader tells @@ -216,8 +216,14 @@ socket. counts within an `epoch`, one run of the display process, so a reader that sees a new `epoch` has a restarted display. - `state.get` with `since` and `epoch` from an earlier answer gets just - `{changed: false, version, epoch, pid, served_at, loop}` while nothing has - changed. + `{changed: false, version, epoch, pid, served_at, loop, volatile}` while + nothing has changed. `volatile` is `{section: {key: value}}`: the current + values of those ignored timestamps, which the reader merges into the copy + it has. They don't make a new version, but they are still news: + `display.last_updated` is how a reader knows the render thread is still + publishing, and `plugins.published_at` the runtime publisher. Without + them a reader's copy kept the timestamps of the last real change, so a + mode on screen for over 120 s read as unknown. - A snapshot that would not fit in a message (hundreds of plugins) is sent without `plugins`, and `truncated: ["plugins"]` says so. Readers then use the cache for that section only. @@ -226,8 +232,10 @@ socket. - a `state` event (a full snapshot) whenever the version changes, and - a `tick` at least every 5 s (`SUBSCRIBE_KEEPALIVE_SECONDS`) when nothing - changed. It carries `loop`, so a stalled render loop shows up within one - tick, and it tells the reader the connection is alive. + changed. It is the short `changed: false` answer, so it carries `loop` + (a stalled render loop shows up within one tick) and `volatile` (the + timestamps stay as fresh as the writers keep them), and it tells the + reader the connection is alive. A slow reader is never sent a backlog: each event is the latest version, so one that falls behind skips the versions in between. A reader that has heard diff --git a/src/ipc/client.py b/src/ipc/client.py index 430859cd2..169fdb41c 100644 --- a/src/ipc/client.py +++ b/src/ipc/client.py @@ -283,6 +283,29 @@ def snapshot_loop_age(snapshot: Mapping[str, Any], return max(float(age), 0.0) + snapshot_age(snapshot, now_mono) +def _merge_volatile(state: Dict[str, Any], volatile: Any) -> None: + """Fold a tick's ``volatile`` values (``{section: {key: value}}``) into + ``state``, copying each section it touches. + + These are the timestamps the hub leaves out of its version -- + ``display.last_updated``, ``on_demand.last_updated``/``remaining``, + ``plugins.published_at`` -- and the readers judge freshness by them, so + a copy that only full ``state`` events updated would go stale while the + same mode stayed on screen. Only keys the section already has are taken: + a tick never adds a section or a key the last snapshot did not carry + (a section left out of a truncated snapshot stays out). + """ + if not isinstance(volatile, dict): + return # a display from before ticks carried them + for name, values in volatile.items(): + section = state.get(name) + if not isinstance(section, dict) or not isinstance(values, dict): + continue + fresh = {k: v for k, v in values.items() if k in section} + if fresh: + state[name] = dict(section, **fresh) + + #: A subscription that has heard nothing for this long is not trusted: the #: display sends a tick at least every SUBSCRIBE_KEEPALIVE_SECONDS. SUBSCRIPTION_SILENCE_SECONDS = 3 * SUBSCRIBE_KEEPALIVE_SECONDS @@ -449,12 +472,17 @@ def _store(self, result: Dict[str, Any], full: bool) -> None: self.snapshots += 1 elif (self._snapshot is not None and result.get('epoch') == self._snapshot.get('epoch')): - # A tick: nothing changed but the render loop's liveness. + # A tick: nothing changed but the render loop's liveness and + # the volatile keys (timestamps) the writers keep refreshing. snap = dict(self._snapshot) + state = dict(snap.get('state') or {}) + if result.get('version') == snap.get('version'): + _merge_volatile(state, result.get('volatile')) loop = result.get('loop') if isinstance(loop, dict): - snap['state'] = dict(snap.get('state') or {}, loop=loop) + state['loop'] = loop snap['loop'] = loop + snap['state'] = state snap['served_at'] = result.get('served_at', snap.get('served_at')) self._snapshot = snap else: diff --git a/src/ipc/contract.py b/src/ipc/contract.py index ee35c0ae0..ec5ef06bb 100644 --- a/src/ipc/contract.py +++ b/src/ipc/contract.py @@ -191,7 +191,7 @@ class Command: #: A subscriber hears from the display at least this often: a ``state`` #: event when something changed, else a ``tick`` carrying the render loop's -#: liveness. A client that has heard nothing for a few of these treats its +#: liveness and the latest volatile timestamps. A client that has heard nothing for a few of these treats its #: copy as unknown. SUBSCRIBE_KEEPALIVE_SECONDS = 5.0 @@ -530,8 +530,10 @@ class StateGetArgs: """``state.get``: the display's state, as a versioned snapshot. With ``since`` and the ``epoch`` it came from, the answer is only - ``{changed: false, version, epoch, served_at, loop}`` while the state is - still at that version, so a poller that already has it is sent no state. + ``{changed: false, version, epoch, served_at, loop, volatile}`` while the + state is still at that version, so a poller that already has it is sent + no state -- only the latest values of the keys that do not count as a + change (``volatile``, see :class:`StateSnapshot`). """ since: Optional[int] = None epoch: Optional[str] = None @@ -672,9 +674,13 @@ class StateSnapshot(TypedDict, total=False): ``version`` counts changes to the state within one ``epoch`` (one run of the display process): a reader that sees a new epoch starts over. ``changed`` is False only for a ``state.get`` whose ``since`` is still - current, and then ``state`` is absent. ``served_at`` is the display's - wall clock when it answered. ``loop`` is measured at that moment, so it - is also inside ``state``. + current, and then ``state`` is absent and ``volatile`` is there instead: + ``{section: {key: value}}``, the current values of the keys the version + ignores (``display.last_updated``, ``on_demand.last_updated`` and + ``remaining``, ``plugins.published_at``). A reader merges them into the + copy it has; they are how it can tell the writers are still publishing. + ``served_at`` is the display's wall clock when it answered. ``loop`` is + measured at that moment, so it is also inside ``state``. ``state`` holds the sections in :data:`STATE_SECTIONS`: @@ -696,12 +702,13 @@ class StateSnapshot(TypedDict, total=False): served_at: float changed: bool state: Dict[str, Any] + volatile: Dict[str, Dict[str, Any]] loop: LoopState class StateEventKind: STATE = 'state' # result: a full StateSnapshot, the latest version - TICK = 'tick' # result: {version, epoch, pid, served_at, loop}; nothing changed + TICK = 'tick' # result: {version, epoch, pid, served_at, loop, volatile}; nothing changed @dataclass(frozen=True) diff --git a/src/ipc/server.py b/src/ipc/server.py index 7075b76c5..e2b456f4e 100644 --- a/src/ipc/server.py +++ b/src/ipc/server.py @@ -222,6 +222,22 @@ def _fingerprint(value: Optional[Mapping[str, Any]], volatile: Iterable[str]) -> return {k: v for k, v in value.items() if k not in skip} if skip else dict(value) +def _volatile_values(sections: Mapping[str, Optional[Dict[str, Any]]], + volatile: Mapping[str, FrozenSet[str]]) -> Dict[str, Dict[str, Any]]: + """``{section: {key: value}}``: the volatile keys each published section + has now. The sections are never mutated after publish (a publish swaps + in a new dict), so reading them outside the lock is safe.""" + values: Dict[str, Dict[str, Any]] = {} + for name, keys in volatile.items(): + value = sections.get(name) + if not keys or not isinstance(value, dict): + continue + present = {k: value[k] for k in keys if k in value} + if present: + values[name] = present + return values + + def _unknown_loop() -> Dict[str, Any]: return {'heartbeat_age_seconds': None, 'armed': False, 'stale_after': None} @@ -259,7 +275,11 @@ class StateHub: (``loop_probe``), so it keeps ageing while the render thread is stuck. The version goes up when a section's value changes, ignoring the keys - the publisher names as volatile (timestamps). Publishing never blocks on + the publisher names as volatile (timestamps). Those keys still carry + news -- ``display.last_updated`` is the render thread's proof of life -- + so the short ``changed: false`` answer, which is what a subscriber's + tick carries, has their current values in ``volatile``; a reader merges + them into its copy. Publishing never blocks on a reader: the lock is held only to swap a dict reference and compare it, and every socket write happens on the reader's own thread, outside it. A reader that is slow gets the latest version when it next asks, not @@ -274,6 +294,7 @@ def __init__(self, loop_probe: Optional[LoopProbe] = None, *, self._cond = threading.Condition(threading.Lock()) self._sections: Dict[str, Optional[Dict[str, Any]]] = {} self._fingerprints: Dict[str, Any] = {} + self._volatile: Dict[str, FrozenSet[str]] = {} self._version = 0 self.epoch = epoch or uuid.uuid4().hex[:16] self.pid = os.getpid() if pid is None else pid @@ -301,9 +322,11 @@ def publish(self, section: str, value: Optional[Mapping[str, Any]], The value is copied (one level), so the caller may reuse its dict. """ stored = None if value is None else dict(value) - fingerprint = _fingerprint(stored, volatile) + skip = frozenset(volatile) + fingerprint = _fingerprint(stored, skip) with self._cond: self._sections[section] = stored + self._volatile[section] = skip if self._fingerprints.get(section, _MISSING) == fingerprint: return False self._fingerprints[section] = fingerprint @@ -333,11 +356,14 @@ def snapshot(self, since: Optional[int] = None, """The :class:`~src.ipc.contract.StateSnapshot` now. ``since`` with this hub's ``epoch``, still the current version, gives - the short ``changed: false`` form. + the short ``changed: false`` form, with ``volatile``: each section's + volatile keys at their latest values (the rest of the section is + what the reader already has). """ with self._cond: version = self._version sections = dict(self._sections) + volatile = dict(self._volatile) loop = self.loop() result: Dict[str, Any] = { 'schema': STATE_SCHEMA, @@ -349,6 +375,7 @@ def snapshot(self, since: Optional[int] = None, } if since is not None and epoch == self.epoch and since == version: result['changed'] = False + result['volatile'] = _volatile_values(sections, volatile) return result state: Dict[str, Any] = {name: sections.get(name) for name in STATE_SECTIONS if name != 'loop'} diff --git a/test/test_ipc_state_stream.py b/test/test_ipc_state_stream.py index f72a68f9b..d512d68ce 100644 --- a/test/test_ipc_state_stream.py +++ b/test/test_ipc_state_stream.py @@ -96,6 +96,27 @@ def test_since_the_current_version_is_the_short_answer(self, hub): assert hub.snapshot(since=1, epoch='other')['changed'] is True assert hub.snapshot(since=1)['changed'] is True + def test_the_short_answer_carries_the_latest_volatile_values(self, hub): + """A version that stays put must not freeze the timestamps a reader + judges freshness by: the short answer (a tick) carries them.""" + hub.publish('on_demand', {'active': False, 'last_updated': 2.0, 'remaining': 0.0}, + volatile=('last_updated', 'remaining')) + hub.publish('brightness', {'brightness': 50}) + version = hub.version + assert not hub.publish('display', _display(updated=1234.5), volatile=('last_updated',)) + short = hub.snapshot(since=version, epoch='e1') + assert short['changed'] is False and 'state' not in short + assert short['volatile'] == {'display': {'last_updated': 1234.5}, + 'on_demand': {'last_updated': 2.0, 'remaining': 0.0}} + # The full answer has them in place already. + assert 'volatile' not in hub.snapshot() + + def test_volatile_values_skip_sections_without_any(self): + hub = StateHub(epoch='e1') + hub.publish('brightness', {'brightness': 50}) + hub.publish('plugins', None, volatile=('published_at',)) + assert hub.snapshot(since=hub.version, epoch='e1')['volatile'] == {} + def test_each_display_run_has_its_own_epoch(self): assert StateHub().epoch != StateHub().epoch @@ -276,6 +297,40 @@ def test_a_tick_refreshes_the_loop_and_keeps_the_state(self, hub): assert latest['state']['loop']['heartbeat_age_seconds'] == 70.0 assert latest['received_mono'] == clock.now + def test_a_tick_refreshes_the_volatile_timestamps(self, hub): + """The bug from the ledpi rig: the same mode on screen for minutes + left the reader's display.last_updated at the last real change.""" + clock = FakeClock() + sub = client.StateSubscription(paths=['/nowhere'], clock=clock) + sub._store(hub.snapshot(), full=True) + hub.publish('display', _display(updated=9999.0), volatile=('last_updated',)) + sub._store(hub.snapshot(since=1, epoch='e1'), full=False) + latest = sub.latest() + assert latest['state']['display']['last_updated'] == 9999.0 + assert latest['state']['display']['mode'] == 'clock' + assert latest['version'] == 1 + + def test_a_tick_adds_no_section_or_key_and_ignores_another_version(self, hub): + clock = FakeClock() + sub = client.StateSubscription(paths=['/nowhere'], clock=clock) + snap = hub.snapshot() + sub._store(snap, full=True) + updated = snap['state']['display']['last_updated'] + sub._store({'version': 1, 'epoch': 'e1', + 'volatile': {'display': {'new_key': 1}, 'plugins': {'published_at': 5.0}, + 'on_demand': 'junk'}}, full=False) + state = sub.latest()['state'] + assert 'new_key' not in state['display'] and state['plugins'] is None + assert state['on_demand'] is None + # A tick for a version this copy is not at says nothing about it. + sub._store({'version': 7, 'epoch': 'e1', + 'volatile': {'display': {'last_updated': 5.0}}}, full=False) + assert sub.latest()['state']['display']['last_updated'] == updated + # A display from before ticks carried them: the loop still refreshes. + sub._store({'version': 1, 'epoch': 'e1', 'loop': {'heartbeat_age_seconds': 3.0}}, + full=False) + assert sub.latest()['loop'] == {'heartbeat_age_seconds': 3.0} + def test_a_tick_from_another_epoch_is_ignored(self, hub): clock = FakeClock() sub = client.StateSubscription(paths=['/nowhere'], clock=clock) @@ -369,6 +424,33 @@ def test_ticks_keep_a_quiet_subscription_alive(self, live, path): assert kinds == ['tick', 'tick', 'tick'] sock.close() + def test_ticks_carry_the_timestamps_the_version_ignores(self, live, path): + _, hub = live + sock, reader = self._subscribe(path) + pending = [] + self._next(sock, reader, pending) + hub.publish('display', _display(updated=4242.0), volatile=('last_updated',)) + for _ in range(3): # a tick already on its way may predate the publish + tick = self._next(sock, reader, pending) + assert tick['event'] == 'tick' + if tick['result']['volatile'] == {'display': {'last_updated': 4242.0}}: + break + else: + pytest.fail(f'no tick carried the new last_updated: {tick}') + sock.close() + + def test_a_subscription_keeps_last_updated_current(self, live, path): + _, hub = live + sub = client.StateSubscription(paths=[path]).start() + try: + assert _wait_until(lambda: sub.latest() is not None) + hub.publish('display', _display(updated=4242.0), volatile=('last_updated',)) + assert _wait_until(lambda: (sub.latest() or {}).get('state', {}) + .get('display', {}).get('last_updated') == 4242.0) + assert sub.snapshots == 1 # ticks, not new snapshots + finally: + sub.stop() + def test_a_burst_is_coalesced_to_the_latest(self, live, path): _, hub = live sock, reader = self._subscribe(path) diff --git a/test/test_state_stream_readers.py b/test/test_state_stream_readers.py index 6402264ea..c7272da41 100644 --- a/test/test_state_stream_readers.py +++ b/test/test_state_stream_readers.py @@ -468,6 +468,140 @@ def test_plugin_runtime_view_prefers_the_socket(self, web, socket_state): assert pkg._plugin_runtime_view().source == 'cache' +# --- freshness over a long-lived subscription --------------------------------------------- + +class _Rig: + """A display and one web subscription on fake clocks, with no socket. + + Each second the render thread runs its publish point (unless stopped); + every 5 s the plugin runtime publisher ticks and the subscription gets + what the server would push it: a ``state`` event when the version moved, + else a tick (the short ``changed: false`` answer). The loop's heartbeat + is the time since the render thread last went round. + """ + + MODE = 'ncaa_fb_live' + + def __init__(self): + from src.ipc import client as control_client + self.wall = FakeClock(1_800_000_000.0) + self.mono = FakeClock(5000.0) + self.last_render = self.mono.now + self.hub = StateHub(loop_probe=lambda: { + 'heartbeat_age_seconds': self.mono.now - self.last_render, 'armed': True, + 'stale_after': display_watchdog.HEARTBEAT_STALE_SECONDS}, + epoch='e1', clock=self.mono, wall_clock=self.wall) + self.dc = _controller(self.hub) + self.dc.current_display_mode = self.MODE + self.dc.mode_to_plugin_id[self.MODE] = 'football-scoreboard' + self.publisher = PluginRuntimePublisher(MagicMock(), _states(), clock=self.mono, + wall_clock=self.wall) + self.publisher.attach_hub(self.hub) + self.sub = control_client.StateSubscription(paths=['/nowhere'], clock=self.mono) + self.render = self.plugins = self.ticks = True + self.sent_version = None + self._publish() + self._push() + + def _publish(self): + from src import display_controller as dc_module + with patch.object(dc_module.time, 'monotonic', self.mono), \ + patch.object(dc_module.time, 'time', self.wall): + self.dc._publish_current_mode_state_if_changed() + self.last_render = self.mono.now + + def _push(self): + snap = self.hub.snapshot(since=self.sent_version, epoch='e1') + self.sub._store(snap, full=bool(snap.get('changed'))) + self.sent_version = snap['version'] + + def run(self, seconds): + for _ in range(int(seconds)): + self.wall.now += 1 + self.mono.now += 1 + if self.render: + self._publish() + if int(self.mono.now) % 5 == 0: + if self.plugins: + self.publisher.tick() + if self.ticks: + self._push() + + def status(self): + return display_state.current_status(self.sub.latest(), now=self.wall.now) + + def runtime(self): + return view_from_socket_state(self.sub.latest(), now=self.wall.now, + now_mono=self.mono.now) + + +class TestFreshnessOverTicks: + """#735 on the ledpi rig: with one mode on screen for more than two + minutes (a live game, Vegas, a single plugin) current-status read + ``mode: null`` from the socket. The version stayed put, so the + subscription's copy kept the ``last_updated`` of the last real change.""" + + def test_a_mode_on_screen_for_fifteen_minutes_stays_known(self): + rig = _Rig() + for _ in range(15): + rig.run(60) + status = rig.status() + assert status['mode'] == rig.MODE, status + assert rig.wall.now - status['last_updated'] <= 5 + # The plugin section, unchanged as long, stays live too. + assert rig.runtime().status == rt.LIVE + assert rig.sub.snapshots == 1 # all of it from ticks + + def test_the_route_keeps_answering_from_the_socket(self, web, monkeypatch): + client, cached = web + cached['display_current_state'] = {'mode': 'from-cache', 'last_updated': 1} + rig = _Rig() + rig.run(10 * 60) + monkeypatch.setattr(display_state, 'read_state', rig.sub.latest) + with patch.object(display_state.time, 'time', rig.wall): + data = _data(client, '/api/v3/display/current-status') + assert (data['mode'], data['source']) == (rig.MODE, 'socket') + + def test_a_render_thread_that_stops_still_reads_stalled_then_unknown(self): + """#726's verdicts are unchanged: the socket thread keeps ticking, + but nothing refreshes last_updated and the heartbeat ages.""" + rig = _Rig() + rig.run(5 * 60) + rig.render = False + rig.run(display_watchdog.HEARTBEAT_STALE_SECONDS + 5) + assert rig.runtime().status == rt.STALLED + assert display_state.loop_heartbeat_age(rig.sub.latest()) >= \ + display_watchdog.HEARTBEAT_STALE_SECONDS + assert rig.status()['mode'] == rig.MODE # not 120 s yet + rig.run(display_state.CURRENT_STATE_MAX_AGE_SECONDS) + assert rig.status() == {'mode': None, 'plugin_id': None, 'last_updated': None} + + def test_a_runtime_publisher_that_stops_goes_stale(self): + rig = _Rig() + rig.run(60) + rig.plugins = False + rig.run(rt.STALE_AFTER + 10) + assert rig.runtime().status == rt.STALE + assert rig.status()['mode'] == rig.MODE + + def test_when_the_ticks_stop_the_route_falls_back_to_the_cache(self, web, monkeypatch): + from src.ipc import client as control_client + client, cached = web + cached['display_current_state'] = {'mode': 'from-cache', 'last_updated': 1} + rig = _Rig() + rig.run(5 * 60) + rig.ticks = False + rig.run(control_client.SUBSCRIPTION_SILENCE_SECONDS + 1) + assert rig.sub.latest() is None + + def no_socket(*args, **kwargs): + raise control_client.ControlError('no_socket') + monkeypatch.setattr(display_state, '_subscription', lambda: rig.sub) + monkeypatch.setattr(control_client, 'state_get', no_socket) + data = _data(client, '/api/v3/display/current-status') + assert (data['mode'], data['source']) == ('from-cache', 'cache') + + # --- end to end over a real socket ------------------------------------------------------- @needs_unix_sockets