rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-10-04 14:25:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(starlark,on-demand): the third-party fixes worth taking, plus a Home Assistant MQTT bridge (#538)

Browse Source 
* feat(starlark,on-demand): the third-party fixes worth taking, plus an MQTT bridge

Analysis of ant456/ledmatrix-fixes-repo, a third-party collection of
patches and services built while running this project on Starlark apps
under MQTT control. Its patches are whole-file copies taken against an
older tree, so applying them as written would revert #523's frame
pacing, #534's display() bool returns and the GitHub token masking in
plugins_manager.js. Three of its claimed fixes are already in main, and
its api_v3 Starlark routes are #535's. What follows is the rest --
verified against current code, and reimplemented where the patch's
approach did not hold up.

**On-demand display.** `pinned` reached the controller from the API, was
stored on it and republished in the status payload, but never narrowed
the rotation -- a pinned request still cycled every mode its plugin
owns. Right for a sports plugin, whose modes are views of one subject;
wrong for a plugin whose modes are unrelated, which is every Starlark
app. Now honoured, and it survives a restart.

Restarting while on-demand was active loaded *only* the on-demand
plugin, so normal rotation had nothing to return to for the life of the
process -- and a restart mid-session is routine, since that is how an
update is applied. The panel came back cycling one plugin's modes with
no way out but clearing the cache by hand. Every enabled plugin loads
now; on-demand still resumes on its saved mode.

Stop requests are exempt from the duplicate guards on purpose, so that a
second click stops a mode a race left running -- which means consuming
the mailbox is the only thing that ends one. It was never consumed, so
the same stop was re-read and re-processed on every poll, forever. Both
paths now share one compare-before-delete helper.

**Starlark rendering.** `extract_schema` parsed the source with a regex,
which can only see option lists written out literally: an app whose
dropdown is filled from a live API call inside `get_schema()` came back
empty, and the config form offered nothing to pick. Now runs `pixlet
schema`, which executes the app, and falls back to the parser when
Pixlet is absent, too old for the subcommand, or the app fails to run.
The third-party patch replaced the parser outright and hardcoded
/usr/local/bin/pixlet; this keeps the fallback and the binary search.

A `|` in a config value was dropped by a shell-metacharacter filter,
though the command is a list with no shell involved -- and apps do use
it as a separator inside one value. The key went missing silently and
the app rendered its own "not configured" screen with nothing to say
why. And a 0-byte render was reported as success: Pixlet exits 0 and
writes nothing when an app has no content, which read downstream as a
working app drawing a black panel.

**Starlark display.** `display()` ignored the mode it was called with,
so a specific app could not be addressed. It now accepts `display_mode`
-- which is the whole mechanism, since the controller inspects the
signature before passing it. Found while there: `_select_next_app` ran
only while `current_app` was unset, so with several apps installed the
first was picked once and shown forever while the rest were rendered on
schedule and never displayed. And `enable_scrolling` was missing, so
multi-frame apps were called once per rotation slot and never advanced
past frame one.

**GET /api/v3/display/modes.** Every mode that can be requested
on-demand, with the plugin that owns it. Nothing exposed this, so
anything driving the display from outside the web UI read each plugin's
manifest.json off disk and reimplemented PluginManager's fallbacks. It
also triggers discovery, which is otherwise lazy and normally happens
because a person opened the dashboard.

**integrations/mqtt_bridge.** Home Assistant control over MQTT
Discovery: a mode select, a stop button, power, brightness. Rewritten
against the API rather than the filesystem, so it needs no read access
to config.json and cannot drift from the web UI. paho-mqtt 2.x
VERSION2, TLS, an availability topic that is also the last will, and
secrets from the environment.

**Two opt-in extras.** A DNS single-request unit, for glibc's parallel
A/AAAA lookup stalling ~5s per name on routers that answer only the A
query -- which makes any plugin calling an external API slow and
Starlark apps, which have a render timeout, fail outright. And a Pixlet
config editor: a script you run and Ctrl+C rather than the third-party
version's always-on unauthenticated Flask service, since it stops the
display for the length of a session. Neither is installed by default.

Long Starlark app names now wrap instead of overflowing their card.

115 new tests across 5 files. Also unblocked
test_starlark_display_contract.py, which was silently skipping wherever
fcntl is absent. Whole suite: no new failures against main.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(mqtt_bridge): the five issues Codacy flagged on this branch

All in the new bridge, all real:

  * requests floor was 2.31.0, which carries CVE-2024-35195,
    CVE-2024-47081 and CVE-2026-25645. Raised to >=2.33.0,<3.0.0, which
    is what the project's own requirements.txt already pins.
  * `import time` was never used.
  * `"mqtt_password": None` in DEFAULTS read as a hardcoded credential.
    It is the "no password configured" default; marked nosec B105, the
    convention used elsewhere in the repo.

Also dropped an unused `build_app` from the display-modes test imports.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: the review findings on this PR

Nine of CodeRabbit's ten, plus the CodeQL alert. The tenth is wrong and
is answered below.

**One bad config section blanked the whole mode list.**
`/display/modes` read `full_config.get(plugin_id, {}).get('enabled')`,
so a non-dict under a plugin id -- a shape DisplayController already
guards, so it happens -- raised AttributeError mid-loop and answered 500
with no modes at all. Every MQTT bridge entity is built from that list.
Now skipped with a warning.

**The DNS scripts reported success they had not earned.** Three separate
paths: `resolvconf -u` failing was swallowed by `|| true`; the
systemd-resolved branch exited 0 without applying anything, so the
oneshot unit recorded success while the workaround was inactive; and the
installer's `|| echo` turned a failed start into "installation
complete." with exit 0. All three now fail loudly. `single-request` is a
glibc resolv.conf option with no resolved.conf equivalent, so on those
hosts the honest answer is that it cannot be applied.

A NetworkManager-generated resolv.conf is regenerated on connection
changes, not only at boot, and the unit is oneshot with RemainAfterExit
-- so the option can vanish mid-boot with nothing to put it back. Now
detected and stated plainly rather than implied to be permanent.

**`Before=` does not order a manual restart.** It only orders units
already in the same transaction, so `systemctl restart ledmatrix` could
bypass the fix. install_dns_fix.sh now writes a ledmatrix.service
drop-in with Wants= and After=. Wants=, not Requires=: a DNS workaround
failing should not stop the display.

**The Pixlet editor's `--lan` is gone.** `pixlet serve` has no
authentication, and a printed warning is not access control. Loopback
only, with the SSH port-forward in the header where the flag used to be
documented -- SSH does the authenticating and nothing is left listening.

**The MQTT example config now defaults to TLS** on 8883. The installer
copies it verbatim, and without TLS the broker password and every
command cross the network in cleartext. A plaintext broker is still
supported and documented, and the bridge warns once at startup when a
password is configured without TLS.

**Not taken: "the upstream Pixlet CLI has no `schema` subcommand."**
Upstream tidbyt/pixlet has none, but `scripts/download_pixlet.sh`
installs `tronbyt/pixlet`, whose `cmd/schema.go` is
`schema [PATH]` -> JSON on stdout, built on
`runtime.NewAppletFromPath`, so it does execute `get_schema()`. That is
exactly what extract_schema_via_pixlet calls. A binary without the
subcommand exits non-zero and falls back to the source parser, which is
already covered by a test.

**CodeQL stack-trace exposure: not taken either.** I removed `details`
first and that broke
test_web_error_detail.py::test_no_api_v3_handler_discards_its_exception,
which enforces `describe_exception` across all ~75 handlers -- written
because a device with failing storage answered "see logs for details"
from the log viewer itself. describe_exception redacts credentials; the
trade-off is the project's and is already made. Restored, with the
reasoning in a comment.

11 new tests. Whole suite: no new failures against main, 4127 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-09-08 08:34:52 -04:00
committed by GitHub
co-authored by Claude Opus 5
parent 793b988d33
commit e23f1f45d3
24 changed files with 2597 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/test_api_v3_display_modes.py
+157
View File
@@ -0,0 +1,157 @@
"""GET /api/v3/display/modes -- the list of modes an on-demand request can name.
Anything driving the display from outside the web UI needs two things that no
existing endpoint gave it: the set of display modes, and which plugin owns each
one. /plugins/installed carries neither, so callers read every plugin's
manifest.json off disk and reimplemented PluginManager's own fallbacks.
"""
import json
from unittest.mock import MagicMock
import pytest
from test._api_v3_test_helpers import ( # noqa: F401 - fixtures
api_v3_client, api_v3_module,
)
MANIFESTS = {
'clock-simple': {'name': 'Simple Clock', 'display_modes': ['clock-simple']},
'football-scoreboard': {
'name': 'Football Scoreboard',
'display_modes': ['nfl_live', 'nfl_recent', 'nfl_upcoming'],
},
'ledmatrix-weather': {'name': 'Weather', 'display_modes': ['weather']},
}
CONFIG = {
'clock-simple': {'enabled': True},
'football-scoreboard': {'enabled': True},
'ledmatrix-weather': {'enabled': False},
}
@pytest.fixture
def client(api_v3_module, api_v3_client):
pm = api_v3_module.api_v3.plugin_manager
pm.plugin_manifests = MANIFESTS
pm.discover_plugins = MagicMock(return_value=list(MANIFESTS))
pm.get_plugin_display_modes = MagicMock(
side_effect=lambda pid: MANIFESTS[pid]['display_modes'])
api_v3_module.api_v3.config_manager.load_config = MagicMock(return_value=CONFIG)
return api_v3_client
def _modes(response):
return {m['mode']: m for m in response.get_json()['data']['modes']}
class TestTheModeListing:
def test_enabled_plugins_contribute_every_mode(self, client):
modes = _modes(client.get('/api/v3/display/modes'))
assert set(modes) == {'clock-simple', 'nfl_live', 'nfl_recent', 'nfl_upcoming'}
def test_each_mode_names_its_plugin(self, client):
"""on-demand/start's find_plugin_for_mode fallback cannot see generated
modes, so the caller has to send plugin_id -- it must come from here."""
modes = _modes(client.get('/api/v3/display/modes'))
assert modes['nfl_live']['plugin_id'] == 'football-scoreboard'
def test_disabled_plugins_are_left_out_by_default(self, client):
modes = _modes(client.get('/api/v3/display/modes'))
assert 'weather' not in modes
def test_disabled_plugins_can_be_asked_for(self, client):
"""They are still valid on-demand targets: the controller enables them
for the duration of the request."""
modes = _modes(client.get('/api/v3/display/modes?include_disabled=1'))
assert modes['weather']['enabled'] is False
def test_a_single_mode_plugin_is_labelled_with_its_own_name(self, client):
modes = _modes(client.get('/api/v3/display/modes'))
assert modes['clock-simple']['name'] == 'Simple Clock'
def test_a_multi_mode_plugin_labels_each_mode_distinctly(self, client):
"""There is no per-mode name anywhere, and three modes all called
"Football Scoreboard" are not a usable dropdown."""
modes = _modes(client.get('/api/v3/display/modes'))
assert modes['nfl_live']['name'] == 'nfl_live'
assert modes['nfl_live']['plugin_name'] == 'Football Scoreboard'
class TestItWorksForACallerThatNeverOpensTheDashboard:
def test_discovery_is_triggered(self, client, api_v3_module):
"""Discovery is lazy and normally runs because a person loaded the
dashboard; a bridge or script would otherwise get an empty list."""
client.get('/api/v3/display/modes')
api_v3_module.api_v3.plugin_manager.discover_plugins.assert_called_once()
def test_no_plugin_manager_is_a_clean_error(self, api_v3_module, api_v3_client):
api_v3_module.api_v3.plugin_manager = None
response = api_v3_client.get('/api/v3/display/modes')
assert response.status_code == 500
assert response.get_json()['status'] == 'error'
def test_a_plugin_with_no_declared_modes_still_appears(self, client, api_v3_module):
"""Its mode is its own id -- the same fallback the controller uses."""
pm = api_v3_module.api_v3.plugin_manager
pm.plugin_manifests = {'starlark-apps': {'name': 'Starlark Apps', 'display_modes': []}}
pm.get_plugin_display_modes = MagicMock(return_value=[])
api_v3_module.api_v3.config_manager.load_config = MagicMock(
return_value={'starlark-apps': {'enabled': True}})
modes = _modes(client.get('/api/v3/display/modes'))
assert modes['starlark-apps']['plugin_id'] == 'starlark-apps'
class TestOneBadConfigSectionDoesNotBlankTheList:
"""config.json can hold a non-dict under a plugin id.
DisplayController guards the same shape, so it happens in practice. Here it
used to raise AttributeError mid-loop and answer 500 with no modes at all --
and the MQTT bridge builds every one of its entities from this list, so one
hand-edited section would empty the Home Assistant dropdown.
"""
@pytest.fixture
def client_with_bad_section(self, api_v3_module, api_v3_client):
pm = api_v3_module.api_v3.plugin_manager
pm.plugin_manifests = MANIFESTS
pm.discover_plugins = MagicMock(return_value=list(MANIFESTS))
pm.get_plugin_display_modes = MagicMock(
side_effect=lambda pid: MANIFESTS[pid]['display_modes'])
api_v3_module.api_v3.config_manager.load_config = MagicMock(return_value={
'clock-simple': {'enabled': True},
'football-scoreboard': "true", # a string, not an object
'ledmatrix-weather': {'enabled': True},
})
return api_v3_client
def test_the_endpoint_still_answers(self, client_with_bad_section):
assert client_with_bad_section.get('/api/v3/display/modes').status_code == 200
def test_the_healthy_plugins_are_still_listed(self, client_with_bad_section):
modes = _modes(client_with_bad_section.get('/api/v3/display/modes'))
assert 'clock-simple' in modes and 'weather' in modes
def test_the_bad_section_is_treated_as_disabled(self, client_with_bad_section):
modes = _modes(client_with_bad_section.get('/api/v3/display/modes'))
assert 'nfl_live' not in modes
def test_a_failure_is_reported_the_way_every_other_handler_reports_one(
self, api_v3_module, api_v3_client):
"""describe_exception, per test_web_error_detail's contract -- an
opaque "see logs for details" is what that test exists to prevent."""
api_v3_module.api_v3.plugin_manager.discover_plugins = MagicMock(
side_effect=RuntimeError("disk is gone"))
resp = api_v3_client.get('/api/v3/display/modes')
assert resp.status_code == 500
assert 'disk is gone' in resp.get_json()['details']
def test_credentials_in_the_exception_are_redacted(self, api_v3_module, api_v3_client):
"""describe_exception is what makes returning detail safe."""
api_v3_module.api_v3.plugin_manager.discover_plugins = MagicMock(
side_effect=RuntimeError("GET https://x/y?api_key=SEC123 failed"))
body = api_v3_client.get('/api/v3/display/modes').get_json()
assert 'SEC123' not in json.dumps(body)
test/test_mqtt_bridge.py
+329
View File
@@ -0,0 +1,329 @@
"""The Home Assistant MQTT bridge, without a broker or a matrix.
The bridge is a translation layer: one MQTT payload in, one api_v3 call out.
Everything worth pinning is on that path -- attaching the plugin_id a mode
needs, rejecting values HA can produce but the API cannot take, and the
discovery configs HA reads once and caches -- so it is all reachable with a
fake API client and a pure function.
"""
import importlib.util
import json
import logging
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
BRIDGE_PATH = (Path(__file__).resolve().parent.parent
/ "integrations" / "mqtt_bridge" / "ledmatrix_mqtt_bridge.py")
@pytest.fixture(scope="module")
def bridge_module():
if not BRIDGE_PATH.exists():
pytest.skip("mqtt bridge is not present")
try:
spec = importlib.util.spec_from_file_location("ledmatrix_mqtt_bridge", BRIDGE_PATH)
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
except ImportError as e:
pytest.skip(f"mqtt bridge dependencies are not installed here: {e}")
finally:
sys.modules.pop("ledmatrix_mqtt_bridge", None)
MODES = [
{"mode": "clock-simple", "plugin_id": "clock-simple",
"plugin_name": "Simple Clock", "name": "Simple Clock", "enabled": True},
{"mode": "nfl_live", "plugin_id": "football-scoreboard",
"plugin_name": "Football Scoreboard", "name": "nfl_live", "enabled": True},
{"mode": "aquarium", "plugin_id": "starlark-apps",
"plugin_name": "Starlark Apps", "name": "aquarium", "enabled": True},
]
@pytest.fixture
def client():
"""A stand-in for LEDMatrixClient that records what it was asked to do."""
fake = MagicMock()
fake.list_modes.return_value = list(MODES)
fake.start_on_demand.return_value = {"request_id": "r1"}
fake.stop_on_demand.return_value = {}
fake.set_power.return_value = {}
fake.set_brightness.return_value = {}
fake.get_brightness.return_value = 90
fake.display_status.return_value = {
"service": {"active": True},
"state": {"active": True, "mode": "nfl_live"},
}
return fake
@pytest.fixture
def handler(bridge_module, client):
h = bridge_module.CommandHandler(client)
h.refresh_modes()
return h
class TestDisplayCommands:
def test_a_mode_is_forced_on_demand(self, handler, client):
result = handler.handle({"action": "display", "mode": "nfl_live"})
assert result["status"] == "success"
assert client.start_on_demand.call_args.kwargs["mode"] == "nfl_live"
def test_the_owning_plugin_is_sent_with_the_mode(self, handler, client):
"""on-demand/start's find_plugin_for_mode fallback cannot resolve a
generated mode, so an omitted plugin_id 404s for every Starlark app."""
handler.handle({"action": "display", "mode": "aquarium"})
assert client.start_on_demand.call_args.kwargs["plugin_id"] == "starlark-apps"
def test_a_home_assistant_label_resolves_to_its_mode(self, handler, client):
"""The select entity shows labels, so that is what comes back."""
handler.handle({"action": "display", "mode": "Simple Clock"})
assert client.start_on_demand.call_args.kwargs["mode"] == "clock-simple"
def test_an_unknown_mode_refreshes_before_giving_up(self, handler, client):
"""A plugin installed since the last refresh is the usual reason."""
client.list_modes.return_value = MODES + [
{"mode": "new_app", "plugin_id": "starlark-apps", "name": "new_app"}]
handler.handle({"action": "display", "mode": "new_app"})
assert client.start_on_demand.call_args.kwargs["plugin_id"] == "starlark-apps"
def test_an_explicit_plugin_id_is_respected(self, handler, client):
handler.handle({"action": "display", "mode": "x", "plugin_id": "custom"})
assert client.start_on_demand.call_args.kwargs["plugin_id"] == "custom"
def test_duration_and_pinned_are_passed_through(self, handler, client):
handler.handle({"action": "display", "mode": "aquarium",
"duration": 300, "pinned": True})
kwargs = client.start_on_demand.call_args.kwargs
assert kwargs["duration"] == 300 and kwargs["pinned"] is True
def test_display_with_nothing_to_show_is_rejected(self, handler, client):
result = handler.handle({"action": "display"})
assert result["status"] == "error"
client.start_on_demand.assert_not_called()
def test_stop_returns_to_normal_rotation(self, handler, client):
assert handler.handle({"action": "stop_display"})["status"] == "success"
client.stop_on_demand.assert_called_once()
class TestPowerAndBrightness:
@pytest.mark.parametrize("state,expected", [("on", True), ("OFF", False), (" On ", True)])
def test_power_states_are_accepted(self, handler, client, state, expected):
assert handler.handle({"action": "power", "state": state})["status"] == "success"
client.set_power.assert_called_with(expected)
def test_an_unknown_power_state_is_rejected(self, handler, client):
assert handler.handle({"action": "power", "state": "maybe"})["status"] == "error"
client.set_power.assert_not_called()
def test_brightness_is_applied(self, handler, client):
assert handler.handle({"action": "brightness", "value": 75})["status"] == "success"
client.set_brightness.assert_called_with(75)
def test_a_float_from_home_assistant_is_accepted(self, handler, client):
"""The number entity publishes "75.0"; int() would raise on that."""
handler.handle({"action": "brightness", "value": "75.0"})
client.set_brightness.assert_called_with(75)
@pytest.mark.parametrize("value", [-1, 101, "bright", None])
def test_out_of_range_brightness_never_reaches_the_api(self, handler, client, value):
assert handler.handle({"action": "brightness", "value": value})["status"] == "error"
client.set_brightness.assert_not_called()
class TestFailuresAreReportedNotRaised:
"""A failed command must publish an error, not kill the MQTT loop."""
def test_an_api_error_becomes_an_error_result(self, handler, client):
client.start_on_demand.side_effect = RuntimeError("Display service is not running")
result = handler.handle({"action": "display", "mode": "nfl_live"})
assert result["status"] == "error"
assert "not running" in result["message"]
def test_an_unknown_action_lists_the_known_ones(self, handler):
result = handler.handle({"action": "explode"})
assert result["status"] == "error"
assert "brightness" in result["message"]
def test_a_missing_action_is_an_error(self, handler):
assert handler.handle({})["status"] == "error"
class TestDiscoveryPayloads:
"""HA reads these once and caches them; a wrong shape is a dead entity."""
@pytest.fixture
def messages(self, bridge_module):
return bridge_module.discovery_messages(
"ledmatrix/command", "ledmatrix/command/state",
"ledmatrix/command/availability", ["Simple Clock", "nfl_live"])
def test_every_entity_is_published(self, messages):
components = {m["topic"].split("/")[1] for m in messages}
assert components == {"select", "button", "switch", "number"}
def test_each_entity_has_a_stable_unique_id(self, messages):
"""Without one HA cannot let a user rename or reassign the entity."""
ids = [m["payload"]["unique_id"] for m in messages]
assert len(ids) == len(set(ids)) and all(ids)
def test_the_select_offers_the_modes_it_was_given(self, messages):
select = next(m for m in messages if "/select/" in m["topic"])
assert select["payload"]["options"] == ["Simple Clock", "nfl_live"]
def test_every_entity_shares_the_availability_topic(self, messages):
"""It is also the bridge's last will, so HA greys the controls out
when the bridge dies rather than leaving them silently inert."""
assert all(m["payload"]["availability_topic"] == "ledmatrix/command/availability"
for m in messages)
def test_every_entity_belongs_to_one_device(self, messages):
assert all(m["payload"]["device"]["identifiers"] == ["ledmatrix"] for m in messages)
def test_the_command_templates_are_valid_json_the_handler_accepts(self, messages, handler):
"""A template that renders malformed JSON fails only at runtime, in HA."""
select = next(m for m in messages if "/select/" in m["topic"])
rendered = select["payload"]["command_template"].replace("{{ value }}", "nfl_live")
assert handler.handle(json.loads(rendered))["status"] == "success"
number = next(m for m in messages if "/number/" in m["topic"])
rendered = number["payload"]["command_template"].replace("{{ value }}", "40")
assert handler.handle(json.loads(rendered))["status"] == "success"
def test_the_switch_payloads_are_valid_json_the_handler_accepts(self, messages, handler):
switch = next(m for m in messages if "/switch/" in m["topic"])
for key in ("payload_on", "payload_off"):
assert handler.handle(json.loads(switch["payload"][key]))["status"] == "success"
def test_the_button_payload_is_valid_json_the_handler_accepts(self, messages, handler):
button = next(m for m in messages if "/button/" in m["topic"])
assert handler.handle(json.loads(button["payload"]["payload_press"]))["status"] == "success"
class TestStateReporting:
def test_state_reflects_the_running_display(self, bridge_module, client):
state = bridge_module.read_state(client)
assert state == {"power": True, "mode": "nfl_live", "brightness": 90}
def test_normal_rotation_reports_no_forced_mode(self, bridge_module, client):
client.display_status.return_value = {
"service": {"active": True}, "state": {"active": False}}
assert bridge_module.read_state(client)["mode"] is None
def test_an_unreachable_field_does_not_blank_the_others(self, bridge_module, client):
"""A stopped display service still has a brightness worth showing."""
client.display_status.side_effect = RuntimeError("connection refused")
state = bridge_module.read_state(client)
assert state["brightness"] == 90 and state["power"] is False
class TestConfigLoading:
def test_defaults_apply_when_no_file_exists(self, bridge_module, tmp_path):
config = bridge_module.load_config(str(tmp_path / "absent.json"))
assert config["mqtt_topic"] == "ledmatrix/command"
assert config["mqtt_port"] == 1883
def test_the_file_overrides_defaults(self, bridge_module, tmp_path):
path = tmp_path / "bridge_config.json"
path.write_text(json.dumps({"mqtt_host": "broker.local", "mqtt_port": "8883"}))
config = bridge_module.load_config(str(path))
assert config["mqtt_host"] == "broker.local"
assert config["mqtt_port"] == 8883, "a port read from JSON must still be an int"
def test_the_environment_overrides_the_file(self, bridge_module, tmp_path, monkeypatch):
"""So a password need not sit in a file the service user can read."""
path = tmp_path / "bridge_config.json"
path.write_text(json.dumps({"mqtt_password": "from-file"}))
monkeypatch.setenv("LEDMATRIX_MQTT_MQTT_PASSWORD", "from-env")
assert bridge_module.load_config(str(path))["mqtt_password"] == "from-env"
def test_the_example_placeholder_password_is_refused(self, bridge_module, tmp_path):
"""Copying the example unedited must fail loudly, not silently fail to
authenticate against the broker."""
path = tmp_path / "bridge_config.json"
path.write_text(json.dumps(
{"mqtt_password": "REPLACE_WITH_YOUR_ACTUAL_MQTT_PASSWORD"}))
with pytest.raises(bridge_module.ConfigError):
bridge_module.load_config(str(path))
def test_malformed_json_is_a_clear_error(self, bridge_module, tmp_path):
path = tmp_path / "bridge_config.json"
path.write_text("{not json")
with pytest.raises(bridge_module.ConfigError):
bridge_module.load_config(str(path))
def test_a_non_numeric_port_is_a_clear_error(self, bridge_module, tmp_path):
path = tmp_path / "bridge_config.json"
path.write_text(json.dumps({"mqtt_port": "not-a-port"}))
with pytest.raises(bridge_module.ConfigError):
bridge_module.load_config(str(path))
class TestTheExampleConfigIsSecureByDefault:
"""The installer copies bridge_config.example.json verbatim on first run.
Without TLS the broker password and every display command cross the network
in cleartext, so the shipped default has to be the safe one -- a plaintext
broker is a deliberate edit, not something you get by not reading.
"""
@pytest.fixture
def example(self):
path = (Path(__file__).resolve().parent.parent
/ "integrations" / "mqtt_bridge" / "bridge_config.example.json")
if not path.exists():
pytest.skip("mqtt bridge example config is not present")
return json.loads(path.read_text(encoding="utf-8"))
def test_tls_is_on(self, example):
assert example["mqtt_tls"] is True
def test_the_port_is_the_tls_one(self, example):
"""1883 with mqtt_tls on would just fail to connect."""
assert example["mqtt_port"] == 8883
def test_certificate_verification_is_not_disabled(self, example):
assert example.get("mqtt_tls_insecure", False) is False
def test_no_password_ships_in_the_example(self, example):
assert example["mqtt_password"] is None
def test_the_example_loads(self, bridge_module, tmp_path):
"""It is copied verbatim, so it must survive load_config."""
path = (Path(__file__).resolve().parent.parent
/ "integrations" / "mqtt_bridge" / "bridge_config.example.json")
config = bridge_module.load_config(str(path))
assert config["mqtt_tls"] is True and config["mqtt_port"] == 8883
class TestCleartextIsCalledOut:
"""Turning TLS off is allowed -- the Mosquitto add-on is plaintext on 1883 --
but it should not be silent when a password is going over it."""
def test_a_password_without_tls_warns(self, bridge_module, caplog):
with caplog.at_level(logging.WARNING):
warned = bridge_module.warn_if_cleartext(
{"mqtt_tls": False, "mqtt_password": "hunter2"})
assert warned is True
assert "unencrypted" in caplog.text
def test_the_warning_does_not_repeat_the_password(self, bridge_module, caplog):
with caplog.at_level(logging.WARNING):
bridge_module.warn_if_cleartext({"mqtt_tls": False, "mqtt_password": "hunter2"})
assert "hunter2" not in caplog.text
def test_no_password_means_nothing_to_lose(self, bridge_module):
assert bridge_module.warn_if_cleartext(
{"mqtt_tls": False, "mqtt_password": None}) is False
def test_tls_on_does_not_warn(self, bridge_module):
assert bridge_module.warn_if_cleartext(
{"mqtt_tls": True, "mqtt_password": "hunter2"}) is False
test/test_on_demand_pinning_and_restart.py
+208
View File
@@ -0,0 +1,208 @@
"""On-demand behaviour that a person can ask for but the controller ignored.
Three separate gaps, all reachable from the web UI's force-display dialog:
* `pinned` was accepted by the API, stored on the controller and published
back in the status payload, but never narrowed the rotation -- a pinned
request still cycled every mode its plugin owns;
* restarting while on-demand was active loaded *only* the on-demand plugin,
so normal rotation had nothing to return to for the life of the process;
* a stop request was exempt from the duplicate guards on purpose and was
never removed from the mailbox, so it was re-processed on every poll
forever.
"""
from unittest.mock import MagicMock
import pytest
def _plugin_with_modes(controller, plugin_id, modes):
"""Register `modes` as loaded modes belonging to `plugin_id`."""
controller.plugin_display_modes[plugin_id] = list(modes)
for mode in modes:
controller.plugin_modes[mode] = MagicMock(spec=[])
controller.mode_to_plugin_id[mode] = plugin_id
class TestPinnedNarrowsTheRotation:
"""A pinned request shows the one mode that was asked for.
Unpinned stays the default: a sports plugin's modes are views of one
subject, so rotating them is right. A Starlark plugin's modes are
unrelated widgets, so it is not.
"""
MODES = ['app_a', 'app_b', 'app_c']
def _activate(self, controller, pinned):
_plugin_with_modes(controller, 'starlark-apps', self.MODES)
controller.available_modes = list(self.MODES)
controller._activate_on_demand({
'request_id': 'r1',
'action': 'start',
'plugin_id': 'starlark-apps',
'mode': 'app_b',
'pinned': pinned,
})
def test_pinned_rotation_holds_the_requested_mode(self, test_display_controller):
c = test_display_controller
self._activate(c, pinned=True)
assert c.on_demand_modes == ['app_b']
def test_unpinned_still_rotates_the_whole_plugin(self, test_display_controller):
c = test_display_controller
self._activate(c, pinned=False)
assert set(c.on_demand_modes) == set(self.MODES)
def test_pinned_still_starts_on_the_requested_mode(self, test_display_controller):
c = test_display_controller
self._activate(c, pinned=True)
assert c.current_display_mode == 'app_b'
def test_the_pin_is_recorded_for_the_status_payload(self, test_display_controller):
c = test_display_controller
self._activate(c, pinned=True)
assert c.on_demand_pinned is True
def test_an_unresolvable_pin_does_not_empty_the_rotation(self, test_display_controller):
"""A pin naming a mode outside the plugin must not leave nothing to show."""
c = test_display_controller
_plugin_with_modes(c, 'starlark-apps', self.MODES)
assert c._apply_on_demand_pin(list(self.MODES), 'not_a_mode', True) == self.MODES
class TestPinSurvivesARestart:
"""The pin is part of the request being resumed, not a per-session flag."""
def test_restored_pinned_state_narrows_the_rotation(self, test_display_controller):
c = test_display_controller
_plugin_with_modes(c, 'starlark-apps', ['app_a', 'app_b', 'app_c'])
c.on_demand_active = True
c.on_demand_plugin_id = 'starlark-apps'
c.on_demand_mode = 'app_c'
c.on_demand_pinned = True
c._populate_on_demand_modes_from_plugin()
assert c.on_demand_modes == ['app_c']
def test_restored_unpinned_state_keeps_every_mode(self, test_display_controller):
c = test_display_controller
_plugin_with_modes(c, 'starlark-apps', ['app_a', 'app_b', 'app_c'])
c.on_demand_active = True
c.on_demand_plugin_id = 'starlark-apps'
c.on_demand_mode = 'app_c'
c.on_demand_pinned = False
c._populate_on_demand_modes_from_plugin()
assert set(c.on_demand_modes) == {'app_a', 'app_b', 'app_c'}
class TestRestartDoesNotStarveTheOtherPlugins:
"""Restarting mid-on-demand used to load only the on-demand plugin.
Restarts during an on-demand session are routine -- it is how an update or
a config change is applied -- and the panel came back cycling one plugin's
modes and nothing else until the on-demand cache was cleared by hand.
"""
DISCOVERED = ['clock', 'weather', 'starlark-apps', 'disabled-one']
@pytest.fixture
def controller(self, test_display_controller):
c = test_display_controller
c.config.update({
'clock': {'enabled': True},
'weather': {'enabled': True},
'starlark-apps': {'enabled': True},
'disabled-one': {'enabled': False},
})
return c
def test_every_enabled_plugin_still_loads(self, controller):
selected = controller._select_startup_plugins(
self.DISCOVERED, {'plugin_id': 'starlark-apps', 'mode': 'app_a'})
assert set(selected) == {'clock', 'weather', 'starlark-apps'}
def test_disabled_plugins_are_still_left_out(self, controller):
selected = controller._select_startup_plugins(
self.DISCOVERED, {'plugin_id': 'starlark-apps', 'mode': 'app_a'})
assert 'disabled-one' not in selected
def test_the_on_demand_state_is_still_restored(self, controller):
controller._select_startup_plugins(
self.DISCOVERED,
{'plugin_id': 'starlark-apps', 'mode': 'app_a', 'pinned': True})
assert controller.on_demand_active is True
assert controller.on_demand_plugin_id == 'starlark-apps'
assert controller.on_demand_mode == 'app_a'
assert controller.on_demand_pinned is True
def test_a_disabled_on_demand_plugin_is_enabled_and_loaded(self, controller):
"""Otherwise the mode being resumed has nothing behind it."""
selected = controller._select_startup_plugins(
self.DISCOVERED, {'plugin_id': 'disabled-one', 'mode': 'x'})
assert 'disabled-one' in selected
assert controller.config['disabled-one']['enabled'] is True
def test_an_unknown_on_demand_plugin_falls_back_to_normal(self, controller):
selected = controller._select_startup_plugins(
self.DISCOVERED, {'plugin_id': 'uninstalled', 'mode': 'x'})
assert set(selected) == {'clock', 'weather', 'starlark-apps'}
assert controller.on_demand_active is False
def test_no_on_demand_config_is_a_normal_startup(self, controller):
selected = controller._select_startup_plugins(self.DISCOVERED, None)
assert set(selected) == {'clock', 'weather', 'starlark-apps'}
assert controller.on_demand_active is False
class TestStopRequestsAreConsumed:
"""A stop request is exempt from the duplicate guards, so the mailbox
delete is the only thing that ends it."""
STOP = {'request_id': 'S1', 'action': 'stop'}
def _arrange(self, controller, active):
controller.on_demand_active = active
controller.on_demand_status = 'active' if active else 'idle'
controller._last_on_demand_poll = None
controller.cache_manager.get = MagicMock(
side_effect=lambda key, *a, **kw:
self.STOP if key == 'display_on_demand_request' else None)
controller.cache_manager.set = MagicMock()
controller.cache_manager.delete = MagicMock()
controller._clear_on_demand = MagicMock()
def test_a_handled_stop_is_removed_from_the_mailbox(self, test_display_controller):
c = test_display_controller
self._arrange(c, active=True)
c._poll_on_demand_requests()
c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
def test_a_stop_arriving_while_idle_is_also_removed(self, test_display_controller):
"""Otherwise a stop sent to an idle display re-fires forever."""
c = test_display_controller
self._arrange(c, active=False)
c._poll_on_demand_requests()
c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
def test_the_stop_is_still_acted_on(self, test_display_controller):
c = test_display_controller
self._arrange(c, active=True)
c._poll_on_demand_requests()
c._clear_on_demand.assert_called_once_with(reason='requested-stop')
def test_a_start_racing_in_behind_a_stop_is_not_discarded(self, test_display_controller):
"""The compare-before-delete applies to stops too."""
c = test_display_controller
self._arrange(c, active=True)
newer = {'request_id': 'S2', 'action': 'start', 'plugin_id': 'p', 'mode': 'm'}
reads = iter([self.STOP, newer])
c.cache_manager.get = MagicMock(
side_effect=lambda key, *a, **kw:
next(reads, newer) if key == 'display_on_demand_request' else None)
c._poll_on_demand_requests()
assert c.cache_manager.delete.call_count == 0
test/test_pixlet_renderer_contract.py
+210
View File
@@ -0,0 +1,210 @@
"""starlark-apps PixletRenderer: what reaches Pixlet, and what counts as a render.
Three things a Starlark app can do that the renderer got wrong:
* put a "|" in a config value -- a shell metacharacter filter dropped the
whole key, though the command is a list and no shell is involved;
* render nothing -- Pixlet exits 0 and writes a 0-byte file, which was
reported as a successful render;
* compute its schema at runtime -- the source parser can only read option
lists that are written out literally, so a dropdown fed by a live API call
came back empty.
"""
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
PLUGIN_DIR = Path(__file__).resolve().parent.parent / "plugin-repos" / "starlark-apps"
@pytest.fixture(scope="module")
def renderer_module():
if not PLUGIN_DIR.exists():
pytest.skip("starlark-apps plugin is not checked out")
sys.path.insert(0, str(PLUGIN_DIR))
try:
spec = importlib.util.spec_from_file_location(
"pixlet_renderer_under_test", PLUGIN_DIR / "pixlet_renderer.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
except Exception as e: # noqa: BLE001 - optional deps may be absent
pytest.skip(f"pixlet_renderer is not importable here: {e}")
finally:
sys.path.remove(str(PLUGIN_DIR))
@pytest.fixture
def renderer(renderer_module):
"""A renderer with a known binary and __init__'s binary search bypassed."""
r = renderer_module.PixletRenderer.__new__(renderer_module.PixletRenderer)
r.timeout = 30
r.pixlet_binary = "/usr/local/bin/pixlet"
return r
def _completed(returncode=0, stdout="", stderr=""):
return subprocess.CompletedProcess(args=[], returncode=returncode,
stdout=stdout, stderr=stderr)
class TestConfigValuesReachPixlet:
"""cmd is a list and there is no shell=True, so nothing here is ever
interpreted by a shell -- the filter is defence in depth, not a boundary."""
def _args_for(self, renderer, tmp_path, config):
star = tmp_path / "app.star"
star.write_text("# app", encoding="utf-8")
out = tmp_path / "out.webp"
def fake_run(cmd, **kw):
out.write_bytes(b"webp-bytes")
fake_run.cmd = cmd
return _completed()
with patch.object(subprocess, "run", side_effect=fake_run):
renderer.render(str(tmp_path / "app.star"), str(out), config=config)
return fake_run.cmd
def test_a_pipe_in_a_value_is_passed_through(self, renderer, tmp_path):
"""Real apps use "|" as a separator inside one config value."""
args = self._args_for(renderer, tmp_path, {"sign_id": "I-476 North|175659"})
assert "sign_id=I-476 North|175659" in args
def test_a_dropped_value_does_not_take_the_key_with_it(self, renderer, tmp_path):
args = self._args_for(renderer, tmp_path, {"sign_id": "I-476 North|175659"})
assert any(a.startswith("sign_id=") for a in args)
@pytest.mark.parametrize("value", [
"$(rm -rf /)",
"`whoami`",
"a;b",
"a&b",
"a>b",
"a<b",
])
def test_actual_shell_metacharacters_are_still_dropped(self, renderer, tmp_path, value):
args = self._args_for(renderer, tmp_path, {"k": value})
assert not any(a.startswith("k=") for a in args)
def test_an_invalid_key_is_still_dropped(self, renderer, tmp_path):
args = self._args_for(renderer, tmp_path, {"bad key!": "x"})
assert not any("bad key!" in a for a in args)
class TestAnEmptyRenderIsNotASuccess:
"""Pixlet exits 0 and writes 0 bytes when an app renders no content."""
def _render(self, renderer, tmp_path, write_bytes):
star = tmp_path / "app.star"
star.write_text("# app", encoding="utf-8")
out = tmp_path / "out.webp"
def fake_run(cmd, **kw):
out.write_bytes(write_bytes)
return _completed()
with patch.object(subprocess, "run", side_effect=fake_run):
return renderer.render(str(tmp_path / "app.star"), str(out))
def test_a_zero_byte_output_is_reported_as_a_failure(self, renderer, tmp_path):
ok, error = self._render(renderer, tmp_path, b"")
assert ok is False
assert "empty" in (error or "").lower()
def test_a_real_output_still_succeeds(self, renderer, tmp_path):
ok, error = self._render(renderer, tmp_path, b"RIFF....WEBP")
assert (ok, error) == (True, None)
def test_a_missing_output_still_fails(self, renderer, tmp_path):
star = tmp_path / "app.star"
star.write_text("# app", encoding="utf-8")
out = tmp_path / "out.webp"
with patch.object(subprocess, "run", return_value=_completed()):
ok, error = renderer.render(str(tmp_path / "app.star"), str(out))
assert ok is False
assert "not found" in (error or "").lower()
class TestSchemaComesFromPixletWhenItCan:
"""`pixlet schema` runs get_schema(); the source parser only reads it.
An app whose dropdown options come from a live API call has no option list
anywhere in its source, so the parser reports an empty dropdown and the
config form offers nothing to choose.
"""
PIXLET_OUTPUT = json.dumps({
"version": "1",
"schema": [
{"id": "roadway", "name": "Roadway", "type": "dropdown",
"description": "Pick a road", "options": [{"display": "I-476", "value": "476"}]},
],
})
def test_the_pixlet_output_is_used(self, renderer, tmp_path):
star = tmp_path / "app.star"
star.write_text("# no schema literal anywhere\n", encoding="utf-8")
with patch.object(subprocess, "run",
return_value=_completed(stdout=self.PIXLET_OUTPUT)):
ok, schema, error = renderer.extract_schema(str(star))
assert ok is True and error is None
assert schema["schema"][0]["options"][0]["value"] == "476"
def test_pixlet_field_keys_are_remapped_for_the_config_form(self, renderer, tmp_path):
"""The UI reads typeOf/desc; Pixlet emits type/description."""
star = tmp_path / "app.star"
star.write_text("#\n", encoding="utf-8")
with patch.object(subprocess, "run",
return_value=_completed(stdout=self.PIXLET_OUTPUT)):
_, schema, _ = renderer.extract_schema(str(star))
field = schema["schema"][0]
assert field["typeOf"] == "dropdown"
assert field["desc"] == "Pick a road"
assert "type" not in field and "description" not in field
@pytest.mark.parametrize("outcome", [
{"return_value": _completed(returncode=1, stderr="unknown command \"schema\"")},
{"return_value": _completed(stdout="not json at all")},
{"side_effect": subprocess.TimeoutExpired(cmd="pixlet", timeout=20)},
{"side_effect": OSError("no such binary")},
])
def test_it_falls_back_to_the_source_parser(self, renderer, tmp_path, outcome):
"""Older Pixlet has no `schema` subcommand, and an app can fail to run.
Neither should cost the schema an app does state literally."""
star = tmp_path / "app.star"
star.write_text(
'def get_schema():\n'
' return schema.Schema(version = "1", fields = [\n'
' schema.Text(id = "city", name = "City", desc = "Your city",'
' icon = "gear"),\n'
' ])\n',
encoding="utf-8")
with patch.object(subprocess, "run", **outcome):
ok, parsed, error = renderer.extract_schema(str(star))
assert ok is True and error is None
assert [f["id"] for f in parsed["schema"]] == ["city"]
def test_no_pixlet_binary_skips_straight_to_the_parser(self, renderer, tmp_path):
renderer.pixlet_binary = None
star = tmp_path / "app.star"
star.write_text("#\n", encoding="utf-8")
assert renderer.extract_schema_via_pixlet(str(star)) is None
def test_a_missing_star_file_is_still_an_error(self, renderer, tmp_path):
ok, schema, error = renderer.extract_schema(str(tmp_path / "gone.star"))
assert ok is False and schema is None and "not found" in error
def test_an_unexpected_pixlet_shape_is_not_trusted(self, renderer, tmp_path):
"""A bare list, or a dict with no schema array, must not reach the UI."""
star = tmp_path / "app.star"
star.write_text("#\n", encoding="utf-8")
with patch.object(subprocess, "run",
return_value=_completed(stdout='["not", "a", "schema"]')):
assert renderer.extract_schema_via_pixlet(str(star)) is None
test/test_starlark_display_contract.py
+124 -1
View File
@@ -7,6 +7,7 @@ of defect as a display() that returns None.
"""
import sys
import types
from pathlib import Path
from unittest.mock import MagicMock
@@ -20,6 +21,17 @@ def manager_module():
if not PLUGIN_DIR.exists():
pytest.skip("starlark-apps plugin is not checked out")
sys.path.insert(0, str(PLUGIN_DIR))
# fcntl is POSIX-only and the manager imports it at module scope for
# manifest locking. None of the display paths under test touch it, so a
# stub is what keeps these assertions running on a developer's machine
# rather than skipping everywhere but the Pi and CI.
injected_fcntl = "fcntl" not in sys.modules
if injected_fcntl:
stub = types.ModuleType("fcntl")
stub.LOCK_EX = 2
stub.LOCK_UN = 8
stub.flock = lambda *a, **kw: None
sys.modules["fcntl"] = stub
try:
import importlib
spec = importlib.util.spec_from_file_location(
@@ -27,10 +39,12 @@ def manager_module():
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
except Exception as e: # noqa: BLE001 - optional deps (pixlet, fcntl) may be absent
except Exception as e: # noqa: BLE001 - optional deps (Pillow, pixlet) may be absent
pytest.skip(f"starlark-apps manager is not importable here: {e}")
finally:
sys.path.remove(str(PLUGIN_DIR))
if injected_fcntl:
sys.modules.pop("fcntl", None)
def _plugin(manager_module):
@@ -80,3 +94,112 @@ class TestDisplayFramePropagates:
assert result is False, "display() claimed success over a failed frame update"
assert isinstance(result, bool), "the controller only skips on a real bool"
class _SelectableApp(_App):
"""An app that display() can pick, render and show."""
def __init__(self, app_id, enabled=True):
super().__init__(frames=[("frame", 100)])
self.app_id = app_id
self._enabled = enabled
def is_enabled(self):
return self._enabled
def _plugin_with_apps(manager_module, *app_ids):
p = _plugin(manager_module)
p.apps = {app_id: _SelectableApp(app_id) for app_id in app_ids}
p._display_frame = lambda: True
return p
class TestDisplayModeSelectsTheApp:
"""The controller passes the mode it is rotating to; the plugin must use it.
It inspects display()'s signature to decide whether to pass display_mode at
all, so the parameter is the whole mechanism -- without it there is no way
to address one specific app, including from an on-demand request pinned to
one.
"""
def test_display_accepts_display_mode(self, manager_module):
import inspect
params = inspect.signature(manager_module.StarlarkAppsPlugin.display).parameters
assert "display_mode" in params, \
"the controller only passes display_mode to plugins whose signature accepts it"
def test_a_named_app_is_the_one_shown(self, manager_module):
p = _plugin_with_apps(manager_module, "aquarium", "printer", "nowplaying")
assert p.display(display_mode="printer") is True
assert p.current_app.app_id == "printer"
def test_a_named_app_wins_over_the_one_already_showing(self, manager_module):
p = _plugin_with_apps(manager_module, "aquarium", "printer")
p.display(display_mode="aquarium")
p.display(display_mode="printer")
assert p.current_app.app_id == "printer"
def test_an_unknown_mode_falls_back_to_rotation(self, manager_module):
"""The plugin id arrives here when no per-app modes are exposed."""
p = _plugin_with_apps(manager_module, "aquarium")
assert p.display(display_mode="starlark-apps") is True
assert p.current_app.app_id == "aquarium"
def test_force_clear_does_not_override_a_named_app(self, manager_module):
p = _plugin_with_apps(manager_module, "aquarium", "printer")
p.display(display_mode="printer", force_clear=True)
assert p.current_app.app_id == "printer"
class TestInstalledAppsTakeTurns:
"""Every installed app gets shown, not just whichever was picked first.
_select_next_app ran only while current_app was unset, so the first enabled
app was chosen once and displayed forever; the rest were rendered on
schedule and never reached the panel.
"""
def test_each_entry_to_the_mode_advances(self, manager_module):
p = _plugin_with_apps(manager_module, "a", "b", "c")
seen = []
for _ in range(3):
p.display(force_clear=True) # force_clear = "you are up"
seen.append(p.current_app.app_id)
assert seen == ["a", "b", "c"]
def test_the_rotation_wraps(self, manager_module):
p = _plugin_with_apps(manager_module, "a", "b")
seen = []
for _ in range(4):
p.display(force_clear=True)
seen.append(p.current_app.app_id)
assert seen == ["a", "b", "a", "b"]
def test_frames_within_one_slot_do_not_advance_the_app(self, manager_module):
"""Only entry to the mode rotates -- the high-FPS loop must not."""
p = _plugin_with_apps(manager_module, "a", "b", "c")
p.display(force_clear=True)
for _ in range(20):
p.display(force_clear=False)
assert p.current_app.app_id == "a"
def test_a_disabled_app_is_skipped(self, manager_module):
p = _plugin_with_apps(manager_module, "a", "b")
p.apps["a"]._enabled = False
p.display(force_clear=True)
assert p.current_app.app_id == "b"
def test_no_enabled_apps_reports_nothing_displayed(self, manager_module):
p = _plugin_with_apps(manager_module, "a")
p.apps["a"]._enabled = False
assert p.display(force_clear=True) is False
class TestAnimationsRunAtFrameRate:
def test_the_plugin_asks_for_the_high_fps_loop(self, manager_module):
"""The controller reads this attribute; a multi-frame app is otherwise
called once per rotation slot and never advances past frame one."""
plugin = manager_module.StarlarkAppsPlugin
assert getattr(plugin, "enable_scrolling", False) is True