Compare commits

..
Author SHA1 Message Date
ChuckandClaude Opus 5 c9c258ac49 fix(starlark): the six review findings that were still real
CodeRabbit raised eight on this PR. Two were already fixed by later
commits (the F541 f-string prefixes, and the twenty-four CodeQL
path-injection alerts, now zero open and the last two dismissed as
false positives). These are the rest, each verified against the code as
it stands rather than as it was reviewed.

Manifest writes shared one temp file. `with_suffix('.tmp')` is a fixed
path, and five routes write the manifest while Flask serves requests
concurrently -- two writers opened `manifest.tmp`, interleaved their
json.dump output and both renamed. The rename is atomic; the content it
published was the mixture, which the next read cannot parse. mkstemp in
the destination directory gives each writer its own file.

A failed import stayed cached. Both dynamic importers insert into
sys.modules before exec_module, which they must, but left the
half-initialised module behind on failure -- every later call then took
the cache branch and raised AttributeError on the missing class instead
of retrying, so one transient failure disabled the repository or the
renderer for the life of the process. The keys stay unprefixed: the
plugin's own manager.py does `from pixlet_renderer import
PixletRenderer`, and sharing that key is what keeps one module object.

Config was applied before it was saved. A false from save_config()
answers 500, but the loaded app kept the new values, so GET config
reported settings that were never written and the plugin rendered with
them until a restart silently put the old ones back. Snapshot and
restore on the failure branch, and do not re-render.

The toggle reported success it had not earned. `_update_manifest_safe`
returns False on failure and the result was dropped, and the in-memory
flag was flipped first -- so a failed write left a toggle that reverted
itself on the next restart. Persist first, then update memory.

The toggle also could not find what the list had shown.
`_starlark_virtual_plugins` publishes the raw manifest key, while the
toggle re-slugified it through `_validate_and_sanitize_app_id`
(lowercase, every character outside [a-z0-9_] rewritten): an app stored
as `My-App` was offered as `starlark:My-App` and looked up as `my_app`,
so toggling an app the page had just drawn answered 404. Check for
traversal with `_validate_starlark_app_path` and keep the key.

`star_file` was stored absolute. Readers join it to the app's own
directory and default to a bare `<app_id>.star`, so the key had two
meanings -- and `Path.__truediv__` discards the left side when the right
is absolute, pinning the manifest to whatever PROJECT_ROOT installed it.
Store the filename.

Two store tests reached GitHub. browse and categories call
list_all_apps_cached(), which fetches the app index on a cold cache --
slow, network-dependent, rate-limitable, and since both only assert "not
a 404" a rate-limited 500 passed and hid it. The file now runs in 1.5s
rather than 17.5s.

13 new tests, all six findings covered. Whole suite: no new failures
against this branch, 3892 passed against 3879.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-07 16:12:18 -04:00
ChuckBuildsandClaude Opus 5 a98bb4d866 fix(starlark): stop three more error paths leaking exception text
CodeQL's three remaining stack-trace-exposure alerts were real, not noise
like the path ones -- these answered the caller with the exception:

  'Failed to save configuration: {e}'
  'File error during upload: {err}'
  'Failed to load app module: {err}'

An OSError there names absolute paths on the device. All three already
logged the detail; the response now says what failed and nothing more,
matching the generic Exception arm sitting directly beneath two of them.

That leaves 25 CodeQL alerts on this PR, all of a kind: 24 path-injection
where traversal is demonstrably blocked, and one command-line-injection
on a list-form subprocess with no shell. Both classes already appear on
main -- api_v3.py carries 11 path-injection and 69 stack-trace alerts
before this change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 15:44:51 -04:00
ChuckBuildsandClaude Opus 5 431daa0b59 refactor(starlark): sanitise with basename, which CodeQL can follow
Threading the checked Path through the callers was right on its own
terms but moved CodeQL 29 -> 28: relative_to() is a check it cannot
trace, so app_dir still resolved back to the URL parameter and all 24
path alerts stayed.

os.path.basename is the sanitiser its path-injection query does follow.
Applied with an equality check so this rejects rather than silently
truncates -- identical behaviour to the character test above it, same
error string, same status code.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 14:54:22 -04:00
ChuckBuildsandClaude Opus 5 c7de8be6f5 refactor(starlark): let the path check be a sanitiser, not a boolean
CodeQL reported 24 path-injection alerts across these handlers. The guard
was effective -- traversal is blocked at every entry, verified on
hardware -- but the shape was wrong in a way worth fixing rather than
dismissing: _validate_starlark_app_path returned a bool, and the caller
then rebuilt the path with `_STARLARK_APPS_DIR / app_id` using the raw
value. Validate in one place, join in another, and the two have to stay
in step by hand. A boolean is also not something CodeQL can follow, so
the value it saw reaching the filesystem was the untrusted one -- the
alerts were reporting the pattern accurately.

It now returns the resolved, checked Path, and every caller uses that
instead of re-joining. Seven join sites became zero; the only remaining
construction from app_id is inside the validator itself.

_standalone_render_starlark_app and _install_star_file validate for
themselves too. Both are reachable independently of the handlers, and a
path built from app_id should not be assembled anywhere without the
check.

Also stops three error paths returning exception text to the caller
(CodeQL's 4 information-exposure alerts): a bad config.json answered with
its absolute path and the parser's message. Logged in full, answered
generically -- the same split the composer blueprint already makes.

The pixlet subprocess is left as it is. It is list-form with no shell,
whitelists config keys against ^[a-zA-Z_][a-zA-Z0-9_]*$ and drops any
value containing shell metacharacters; "x; id", "x$(id)" and "x`id`" are
all treated as data on hardware. Nothing to fix there.

Behaviour is unchanged: same error strings for callers, same status
codes. 32 Starlark tests and the full core suite (3979) pass.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 14:44:43 -04:00
ChuckBuildsandClaude Opus 5 2a92f1ab3b fix(starlark): show installed apps with the other plugins, and let them toggle
An app installed from the store appeared nowhere and could not be enabled
or disabled. Same cause as the 404s: #253 surfaced installed apps in
/plugins/installed as `starlark:<app_id>` entries and routed `starlark:`
toggles to the Starlark manifest, and #330 removed both along with the
routes.

Without the first, the app store installs successfully into a list
nothing renders. Without the second, toggling one falls through to the
plugin_manager lookup and answers "Plugin not found" -- a Starlark app is
an entry in starlark-apps' own manifest, not a plugin in that sense.

Restored as two named helpers rather than the original inline blocks:

  _starlark_virtual_plugins()  reads the loaded plugin when there is one
                               and the on-disk manifest otherwise, so the
                               list is right before starlark-apps loads
  _toggle_starlark_app()       updates through the plugin's own
                               _update_manifest_safe when loaded, or the
                               manifest directly when not

Two changes on the original. The toggle now runs app_id through
_validate_and_sanitize_app_id first -- it reaches a filesystem manifest
and had no validation where the other Starlark routes all have it. And
_starlark_virtual_plugins swallows its own failures: the entries are
appended to the real plugin list, and a broken Starlark manifest should
cost the Starlark rows, not empty the plugins page.

6 new tests covering listing, the fields the UI keys on, toggle
persistence, the unknown-app 404, traversal rejection, and that a
Starlark failure leaves the rest of the list intact.

Full core suite: 3979 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 14:31:40 -04:00
ChuckBuildsandClaude Opus 5 2538690435 fix(starlark): restore the remaining eleven routes, so the app store works
The Pixlet button was the reported symptom; the app store is dead the
same way. #330 dropped all thirteen Starlark routes, and browse,
categories and install are what the store page is built on -- each
answering the generic 404, which from the UI is indistinguishable from
an empty store.

Restores the other eleven from 1c4d5c52^: apps list and detail, delete,
per-app config get/put, toggle, render, manual .star upload, and the
Tronbyte repository browse/categories/install. Their dependency closure
came with them (8 helpers, resolved by walking the handlers' references
rather than by eye), plus the Tuple and Type imports the old file had.

Two changes rather than a straight revert. The restored handlers used
`request.get_json()` where the file has since standardised on
`silent=True`: without it Werkzeug raises on a bodyless POST before the
handler's own `if not data` guard runs, so the caller gets a framework
error instead of the declared 400. test_api_v3_optional_body.py already
checks for exactly that and caught all three.

test_every_endpoint_the_frontend_calls_is_registered reads the URLs out
of plugins_manager.js and matches each through the URL map, so a rewrite
of this file cannot quietly drop the set again -- one assertion over the
frontend's own list is what would have caught #330.

26 tests, 17 of which fail against main. Full core suite: 3973 passed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 14:14:23 -04:00
ChuckBuildsandClaude Opus 5 76c5e367ef fix(starlark): restore the Pixlet install and status routes
"Pixlet install failed: Resource not found" is Flask's 404 handler. The
install button posts to /api/v3/starlark/install-pixlet, that route does
not exist, and app.py's generic 404 answers with

    {"status": "error", "message": "Resource not found"}

which the button prints verbatim -- naming neither the resource nor the
cause.

The routes were added by #253 and removed by #330, which rewrote
api_v3.py (3272 lines changed) and dropped all thirteen Starlark routes
with it. Nothing else moved: the frontend still calls them and
plugin-repos/starlark-apps still implements the work behind them, so
every Starlark call in the UI has been landing on the 404 handler since.

Restores the two the Pixlet flow needs -- install-pixlet, and the status
call the button reloads afterwards -- verbatim from 1c4d5c52^, plus the
three helpers they use (_get_starlark_plugin, _find_pixlet_binary,
_read_starlark_manifest) and _STARLARK_APPS_DIR.

The installer itself is fine; I checked rather than assumed. Against the
live release, the naming the script builds matches what tronbyt/pixlet
publishes, and the URL resolves:

    latest tag              v0.53.1
    asset published         pixlet_v0.53.1_linux-arm64.tar.gz
    script would request    pixlet_v0.53.1_linux-arm64.tar.gz
    HEAD                    200

So the missing route was the whole fault.

Deliberately scoped to the reported bug. The other eleven routes are
still missing -- apps CRUD, config, toggle, render, repository browse
and install, repository categories -- and the rest of the Starlark page
is still dead. #463 covers those alongside behavioural changes; this
stands alone so the install can be fixed without them.

test_starlark_pixlet_routes.py asserts both routes are registered and
answer in the shape the JS reads, including the not-yet-loaded-plugin
case a first-time user is in when they press Install. 8 of its 9 tests
fail against main.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
2026-09-07 13:47:14 -04:00
15 changed files with 1706 additions and 6559 deletions
-52
View File
@@ -1,52 +0,0 @@
"""web_interface/app.py resolves plugins_dir from config at import time.
`project_root` used to be assigned only inside the `else` branch (the
relative-path case). SchemaManager and composer_bp are wired up with
`project_root` further down the same module, so a device configured with an
*absolute* `plugin_system.plugins_directory` (a supported value -- see
CLAUDE.md) hit `UnboundLocalError` importing web_interface.app at all, taking
the whole web UI down.
app.py isn't imported directly by the test suite -- it has real side effects
(managers, background state) at import time -- so this extracts the exact
resolution block by its stable comment markers and executes it in isolation,
the same way the rest of the suite validates generated/templated code rather
than re-implementing it.
"""
from pathlib import Path
APP_PY = Path(__file__).resolve().parent.parent / "web_interface/app.py"
START_MARKER = "# Resolve plugin directory - handle both absolute and relative paths\n"
END_MARKER = "\nplugin_manager = PluginManager("
def _resolution_block() -> str:
src = APP_PY.read_text()
start = src.index(START_MARKER)
end = src.index(END_MARKER, start)
return src[start:end]
def test_resolution_block_still_matches_expected_markers():
"""If app.py is restructured enough that these markers move, the exec
below would silently test nothing -- fail loudly instead."""
block = _resolution_block()
assert "plugins_dir" in block and "os.path.isabs" in block
def test_absolute_plugins_directory_does_not_raise_unboundlocalerror():
import os
ns = {"os": os, "Path": Path, "__file__": str(APP_PY),
"plugins_dir_name": "/opt/ledmatrix-plugins"}
exec(compile(_resolution_block(), str(APP_PY), "exec"), ns) # noqa: S102
assert ns["project_root"] == APP_PY.parent.parent
assert ns["plugins_dir"] == Path("/opt/ledmatrix-plugins")
def test_relative_plugins_directory_still_resolves_under_project_root():
import os
ns = {"os": os, "Path": Path, "__file__": str(APP_PY),
"plugins_dir_name": "plugin-repos"}
exec(compile(_resolution_block(), str(APP_PY), "exec"), ns) # noqa: S102
assert ns["plugins_dir"] == ns["project_root"] / "plugin-repos"
@@ -1,64 +0,0 @@
"""composer_bp must actually be registered on the Flask app.
web_interface/blueprints/composer.py defines a full blueprint -- the
/composer/ page, /composer/api/generate, /composer/api/install, etc. -- but
nothing in web_interface/app.py imported or registered it. pages_v3 and
api_v3 both go through the same "import, wire managers, register_blueprint"
sequence; composer_bp had none of the three. The routes existed and worked
in isolation (every other composer test builds its own minimal Flask app
and registers composer_bp directly), but in the real running app every
composer URL -- including the page itself -- was a 404, and the frontend's
hardcoded '/composer/api/...' fetches (composer-app.js) had nothing to
reach.
app.py has import-time side effects (it loads config, builds managers, and
in some configurations starts background threads), so this scans the source
text rather than importing the module -- the same approach already used for
test_composer_js_contracts.py's structural checks.
"""
import re
from pathlib import Path
APP_PY = Path(__file__).resolve().parent.parent / "web_interface" / "app.py"
def _source() -> str:
return APP_PY.read_text()
def test_composer_blueprint_is_imported():
src = _source()
assert re.search(
r"from web_interface\.blueprints\.composer import composer_bp", src
), "app.py never imports composer_bp"
def test_composer_blueprint_is_registered_under_composer_prefix():
src = _source()
match = re.search(
r"app\.register_blueprint\(\s*composer_bp\s*,\s*url_prefix\s*=\s*"
r"['\"](?P<prefix>[^'\"]*)['\"]",
src,
)
assert match, "app.py never calls app.register_blueprint(composer_bp, ...)"
# composer.html and composer-app.js hardcode '/composer/api/...' and
# '/composer/api/fonts/...' -- any other prefix serves a page whose own
# fetches and @font-face all 404.
assert match.group("prefix") == "/composer"
def test_composer_blueprint_managers_are_wired_before_registration():
"""Registered with no managers set means every route 503s or crashes.
composer.py initialises config_manager/plugin_manager/plugins_dir/
project_root to None at import time specifically so app.py can set them
-- the same pattern api_v3 and pages_v3 use.
"""
src = _source()
register_match = re.search(r"app\.register_blueprint\(\s*composer_bp", src)
assert register_match, "app.py never registers composer_bp"
before_registration = src[: register_match.start()]
for attr in ("config_manager", "plugin_manager", "plugins_dir", "project_root"):
assert re.search(rf"composer_bp\.{attr}\s*=", before_registration), (
f"composer_bp.{attr} is never set before registration"
)
-323
View File
@@ -1,323 +0,0 @@
"""The composer generates Python that the plugin loader imports and executes.
/api/install writes the generated manager.py into plugins_dir and the loader
imports it, so anything the payload can splice into that source runs on the
device. The ast.parse check in _generate_plugin_files rejects only *invalid*
syntax -- an injected `import os` is perfectly valid and passed it.
Two ways in, both confirmed against the code before it was fixed:
metadata.name = a name containing a triple-quote, a newline, then
`import os; PWNED = os.getuid()`, then another triple-quote
-> closes the module docstring; the rest became module-level statements
(spelled out rather than shown literally -- writing the payload into
this docstring closes *this* file's docstring, which is the bug)
element x = '0 or __import__("os").system("id")'
-> f-string interpolated it verbatim: x=0 or __import__("os").system("id")
"""
import ast
import re
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from web_interface.blueprints import composer as C # noqa: E402
BASE_META = {"id": "test-plugin", "name": "Clock", "author": "a",
"version": "1.0.0", "description": "d"}
#: Values that terminate a Python expression and start a new statement.
EXPR_PAYLOADS = [
'0 or __import__("os").system("id")',
'0);import os;os.system("id");(',
'__import__("subprocess").run(["id"])',
"0 if False else exec('x=1')",
"1e999", "nan", "0x41", "0__0",
]
#: Values that close a string literal in the generated source.
LITERAL_PAYLOADS = [
'Clock"""\nimport os; PWNED = os.getuid()\n"""',
"Clock'''\nimport os\n'''",
'Clock" + __import__("os").system("id") + "',
"Clock\\", "Clock\nimport os",
]
def _payload(**over):
# dataModel.configVars is the key _generate_plugin_files reads; "config_vars"
# was never looked at, so anything passed through it tested nothing.
p = {"metadata": dict(BASE_META), "elements": [],
"dataModel": {"configVars": over.pop("config_vars", [])}}
p["metadata"].update(over.pop("metadata", {}))
p.update(over)
return p
def _generated(payload):
return C._generate_plugin_files(payload)["manager.py"]
def _module_level_code(src):
"""Statements at module level that are not the docstring/imports/classes."""
tree = ast.parse(src)
out = []
for node in tree.body:
if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.ImportFrom, ast.Import)):
continue
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant):
continue # the docstring
out.append(ast.unparse(node))
return out
@pytest.mark.parametrize("payload", LITERAL_PAYLOADS)
def test_a_name_that_breaks_out_of_a_literal_is_refused(payload):
with pytest.raises(C.ComposerInputError):
_generated(_payload(metadata={"name": payload}))
#: Types with a drawing branch in manager.py.j2. An injection test using any
#: other type proves nothing: _preprocess_elements drops it, so its values
#: never reach the generated source and every assertion passes trivially.
#: This test previously used "line", which has never had a branch.
RENDERED_GEOMETRY_CASES = [
("rectangle", {"x": 0, "y": 0, "width": 10, "height": 8}),
("arc", {"x": 0, "y": 0, "width": 24, "height": 24}),
("ellipse", {"x": 0, "y": 0, "width": 24, "height": 12}),
("rounded_rectangle", {"x": 0, "y": 0, "width": 24, "height": 10}),
("gauge", {"x": 0, "y": 0, "width": 32, "height": 32}),
]
@pytest.mark.parametrize("etype,base", RENDERED_GEOMETRY_CASES)
@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
@pytest.mark.parametrize("field", ["x", "y", "width", "height"])
def test_a_non_numeric_geometry_value_cannot_reach_the_source(etype, base, evil, field):
"""width/height were interpolated raw into the generated source.
p['x2_expr'] = f"({x_expr}) + {w}" with w straight off the payload, so a
rectangle with width='0 or __import__("os").system("id")' produced
[0, 0, (0) + 0 or __import__("os").system("id"), (0) + 8],
in a manager.py that /api/install writes to disk and the loader imports.
"""
el = {"type": etype, "id": "e1", **base}
el[field] = evil
src = _generated(_payload(elements=[el]))
assert "__import__" not in src, f"{etype}.{field}={evil!r} reached the generated source"
assert "os.system" not in src
assert not _module_level_code(src), \
f"{etype}.{field}={evil!r} produced module-level statements: {_module_level_code(src)}"
def test_every_injection_case_uses_a_type_that_actually_renders():
"""Guards against the whole suite quietly going vacuous again.
An element type with no template branch is dropped before generation, so
an injection test written against one asserts nothing and still passes.
"""
used = {etype for etype, _ in RENDERED_GEOMETRY_CASES}
missing = used - set(C._RENDERABLE_ELEMENT_TYPES)
assert not missing, f"injection tests use non-rendering types: {sorted(missing)}"
@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
@pytest.mark.parametrize("channel", ["r", "g", "b"])
def test_a_non_numeric_colour_channel_cannot_reach_the_source(evil, channel):
el = {"type": "text", "id": "t1", "x": 0, "y": 0, "text": "hi",
"font": "press_start", "r": 255, "g": 255, "b": 255}
el[channel] = evil
src = _generated(_payload(elements=[el]))
assert "__import__" not in src and "os.system" not in src
assert not _module_level_code(src)
def test_colour_channels_are_clamped_to_a_byte():
el = {"type": "text", "id": "t1", "x": 0, "y": 0, "text": "hi",
"font": "press_start", "r": 99999, "g": -5, "b": 128}
src = _generated(_payload(elements=[el]))
assert "(255, 0, 128)" in src, "channels were not clamped to 0-255"
#: fillR/fillG/fillB (and outR/G/B, bgR/G/B, trackR/G/B) go through
#: _as_fill_filter -> _as_rgb_filter, a separate path from _rgb_expr above.
#: It used raw int() until it was found to raise ValueError on a non-numeric
#: channel instead of clamping like every other coerced value in this module.
@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
@pytest.mark.parametrize("channel", ["fillR", "fillG", "fillB"])
def test_a_non_numeric_fill_channel_cannot_reach_the_source(evil, channel):
el = {"type": "rectangle", "id": "r1", "x": 0, "y": 0, "width": 10, "height": 8,
"fillR": 0, "fillG": 0, "fillB": 128}
el[channel] = evil
src = _generated(_payload(elements=[el]))
assert "__import__" not in src and "os.system" not in src
assert not _module_level_code(src)
def test_fill_channels_are_clamped_to_a_byte():
el = {"type": "rectangle", "id": "r1", "x": 0, "y": 0, "width": 10, "height": 8,
"fillR": 99999, "fillG": -5, "fillB": 128}
src = _generated(_payload(elements=[el]))
assert "(255, 0, 128)" in src, "fill channels were not clamped to 0-255"
def test_the_generated_module_still_has_no_top_level_statements():
"""The clean case: a normal payload produces only imports and a class."""
el = {"type": "text", "id": "t1", "x": 4, "y": 4, "text": "hi",
"font": "press_start", "r": 1, "g": 2, "b": 3}
src = _generated(_payload(elements=[el]))
assert not _module_level_code(src)
assert "(1, 2, 3)" in src
# --- config variable keys ---------------------------------------------------
def _with_key(key):
return {"metadata": dict(BASE_META), "elements": [],
"dataModel": {"configVars": [{"key": key, "type": "string",
"default": "x", "label": "L"}]}}
@pytest.mark.parametrize("key", ["class", "def", "import", "None", "True",
"lambda", "pass", "match", "case"])
def test_a_keyword_config_key_is_named_in_the_error(key):
"""ast.parse already rejected these, but as an unhelpful line number.
"Generated code has a syntax error: invalid syntax (line 17)" tells the
user nothing about which field to fix.
"""
with pytest.raises(C.ComposerInputError) as exc:
_generated(_with_key(key))
assert key in str(exc.value) and "keyword" in str(exc.value).lower()
@pytest.mark.parametrize("key", ["config", "logger", "display_manager",
"cache_manager", "plugin_id", "enabled",
"self", "update", "display"])
def test_a_reserved_attribute_config_key_is_refused(key):
"""These generate *valid* Python that silently clobbers plugin state.
The worst is `config`: the assignment lands right after super().__init__(),
so `self.config = config.get("config", "x")` replaces the plugin's config
dict with a string and every later self.config.get(...) fails at runtime.
"""
with pytest.raises(C.ComposerInputError) as exc:
_generated(_with_key(key))
assert key in str(exc.value) and "reserved" in str(exc.value).lower()
@pytest.mark.parametrize("key", ["brightness", "my_var", "_private", "x1",
"update_interval_seconds"])
def test_ordinary_config_keys_are_still_accepted(key):
src = _generated(_with_key(key))
assert f"self.{key} = config.get(" in src
def test_the_generated_config_assignment_does_not_precede_super_init():
"""Guards the reasoning behind the reserved list, not just the list."""
src = _generated(_with_key("brightness"))
body = src.splitlines()
super_at = next(i for i, line in enumerate(body) if "super().__init__(" in line)
assign_at = next(i for i, line in enumerate(body)
if "self.brightness = config.get(" in line)
assert assign_at > super_at, (
"config vars are assigned before super().__init__(); the reserved-name "
"list assumes they land after it")
# --- optional keys ----------------------------------------------------------
@pytest.mark.parametrize("el_type,missing", [
("text", "text"), ("text", "text2"), ("clock", "format"),
])
def test_an_element_missing_an_optional_key_does_not_500(el_type, missing):
"""`p` is a copy of the raw element, so an absent key stays absent.
The defaults were applied to locals only, so manager.py.j2 rendered
`{{ el.text | tojson }}` over a jinja2.Undefined and tojson raised
TypeError -- which no handler catches, making a missing key a 500 rather
than a validation error or a sensible default.
"""
el = {"type": el_type, "id": "e1", "x": 0, "y": 0, "font": "press_start"}
src = _generated(_payload(elements=[el]))
ast.parse(src) # must still be valid Python
assert "Undefined" not in src
def test_a_clock_without_a_format_uses_the_documented_default():
el = {"type": "clock", "id": "c1", "x": 0, "y": 0, "font": "press_start"}
src = _generated(_payload(elements=[el]))
assert '"%H:%M"' in src, "the %H:%M default did not reach the generated source"
#: (element type, channel key, base element) for colour channels that were
#: interpolated raw rather than through _rgb_expr/_safe_int. Prefixed channels
#: (emptyR/G/B, labelR/G/B) were the ones the original r/g/b test never reached.
RAW_COLOUR_CASES = [
("progress_bar", "r", {"x": 0, "y": 0}),
("progress_bar", "g", {"x": 0, "y": 0}),
("pips", "b", {"x": 0, "y": 0}),
("pips", "emptyR", {"x": 0, "y": 0}),
("pips", "emptyG", {"x": 0, "y": 0}),
("sparkline", "r", {"x": 0, "y": 0}),
("gauge", "labelR", {"x": 0, "y": 0, "width": 32, "height": 32}),
("gauge", "labelB", {"x": 0, "y": 0, "width": 32, "height": 32}),
]
@pytest.mark.parametrize("etype,channel,base", RAW_COLOUR_CASES)
@pytest.mark.parametrize("evil", EXPR_PAYLOADS)
def test_a_prefixed_colour_channel_cannot_reach_the_source(etype, channel, base, evil):
"""Five tuples were built with f"({el.get('r', 100)}, ...)" -- no coercion.
The pre-existing colour test only covered r/g/b on a text element, so the
prefixed channels and the four other types were never exercised.
"""
el = {"type": etype, "id": "e1", **base}
el[channel] = evil
src = _generated(_payload(elements=[el]))
assert "__import__" not in src, f"{etype}.{channel}={evil!r} reached the source"
assert "os.system" not in src
assert not _module_level_code(src)
@pytest.mark.parametrize("value", [float("inf"), float("-inf"), float("nan")])
@pytest.mark.parametrize("field", ["x", "y", "width", "height"])
def test_a_non_finite_dimension_does_not_escape_as_an_unhandled_error(field, value):
"""json.loads accepts Infinity/NaN and Flask passes them through, so a
payload can hand _safe_int a non-finite float. int(inf) raises
OverflowError -- neither ValueError nor ComposerInputError -- so it escaped
both handlers and surfaced as a 500 with a traceback instead of a 422."""
el = {"type": "rectangle", "id": "r1", "x": 0, "y": 0, "width": 10, "height": 8}
el[field] = value
src = _generated(_payload(elements=[el])) # must not raise
# A non-finite value must be replaced by the default, not spelled into the
# source. Word-boundary match: "info" in self.logger.info contains "inf".
assert not re.search(r"\b(inf|nan|Infinity|NaN)\b", src), \
f"{field}={value!r} leaked a non-finite literal into the source"
assert not _module_level_code(src)
@pytest.mark.parametrize("bad_id", [
'x = __import__("os").system("id") #',
"x\nimport os\n_y",
"x[0]",
"",
"a" * 200,
])
def test_a_marquee_id_cannot_become_code(bad_id):
"""data_key is spliced UNQUOTED into variable names
(_{{ data_key }}_text = ...), so a non-identifier id landed in the source
as code. ast.parse caught it, but the caller then got an opaque
"Generated code has a syntax error" rather than being told the id is bad."""
el = {"type": "marquee", "id": bad_id, "x": 0, "y": 0, "text": "hi"}
src = _generated(_payload(elements=[el])) # must not raise
assert "__import__(" not in src
assert "os.system(" not in src
assert not _module_level_code(src)
-124
View File
@@ -1,124 +0,0 @@
"""An element the template cannot draw must not produce an empty `if` block.
manager.py.j2 wraps each element in `if width >= N:` (breakpoint) and/or
`if int(time.time() * 2) % 2:` (blink), and the body comes from the per-type
branches. A type with no branch contributed nothing, so the wrapper opened a
block with no statements in it. ast.parse in _generate_plugin_files then
failed and the caller was told only:
Generated code has a syntax error: expected an indented block after
'if' statement on line 49
which names a line of generated source the user never sees. Confirmed against
the code before the fix with a `group` element carrying minWidth.
Two defences, both covered here: _preprocess_elements drops types the template
has no branch for, and the template emits a `pass` fallback so a type added to
the canvas before its branch exists degrades to a no-op instead of a broken
plugin.
"""
import ast
import re
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from web_interface.blueprints import composer as C # noqa: E402
TEMPLATE = (Path(__file__).resolve().parent.parent
/ "web_interface/templates/v3/composer/manager.py.j2")
BASE_META = {"id": "test-plugin", "name": "Clock", "author": "a",
"version": "1.0.0", "description": "d"}
def generate(element):
return C._generate_plugin_files({
"metadata": BASE_META,
"elements": [element],
"dataModel": {"configVars": []},
})
@pytest.mark.parametrize("wrapper", [
{"minWidth": 64}, # breakpoint block
{"blink": True}, # blink block
{"minWidth": 64, "blink": True}, # both, nested
])
@pytest.mark.parametrize("etype", ["group", "widget_9000", "section"])
def test_undrawable_element_does_not_break_generation(etype, wrapper):
element = {"type": etype, "x": 0, "y": 0, "color": "#ffffff", **wrapper}
files = generate(element) # must not raise ComposerInputError
assert "manager.py" in files
def test_drawable_element_still_renders_inside_a_breakpoint():
files = generate({"type": "text", "text": "hi", "x": 0, "y": 0,
"minWidth": 64, "color": "#ffffff"})
src = files["manager.py"]
assert "if width >= 64:" in src
assert "draw_text" in src
def test_renderable_types_match_the_template_branches():
"""The constant and the template must agree.
A type listed in the constant with no branch emits an empty block (the bug
above); a type with a branch but missing from the constant is silently
dropped from every generated plugin. Neither is visible without this check.
"""
branches = set(re.findall(r"el\.type == '([a-z_]+)'", TEMPLATE.read_text()))
assert branches == set(C._RENDERABLE_ELEMENT_TYPES)
def test_template_closes_the_branch_chain_with_a_fallback():
"""Belt and braces: even if the constant drifts, no empty block escapes."""
text = TEMPLATE.read_text()
assert "{% else %}" in text
assert "pass # element type" in text
@pytest.mark.parametrize("wrapper", [
{"minWidth": 64},
{"blink": True},
{"minWidth": 64, "blink": True},
])
def test_dynamic_text_with_non_config_binding_does_not_break_generation(wrapper):
"""dynamic_text only renders a body for binding.source == 'config'.
_preprocess_elements accepts any string as binding.source (it just
defaults a missing one to 'config'), so a 'live' or 'sensor' source --
anything a client sends that isn't literally 'config' -- hit the same
empty-if-block bug as an undrawable element type, just one level deeper:
the branch is taken, but its own inner `if` produced nothing.
"""
element = {"type": "dynamic_text", "x": 0, "y": 0, "color": "#ffffff",
"binding": {"source": "live", "key": "temperature"}, **wrapper}
files = generate(element) # must not raise ComposerInputError
assert "manager.py" in files
assert "non-config dynamic_text binding draws nothing" in files["manager.py"]
def test_binding_source_cannot_break_out_of_the_comment_it_lands_in():
"""binding.source used to be interpolated straight into a Python comment
(`pass # dynamic_text binding_source "{{ el.binding_source }}" draws
nothing`) with no escaping. A source string carrying a newline closed the
comment, and text on the following line(s), indented to match, became a
real statement in the generated plugin -- CWE-94, and live once
composer_bp was registered (confirmed: this exact payload produces a
manager.py containing a clean, ast.parse-valid `import os` against the
pre-fix template). The comment is now a fixed literal that never
interpolates the value at all.
"""
payload = "foo\n import os\n os.system('id') # "
element = {"type": "dynamic_text", "x": 0, "y": 0, "color": "#ffffff",
"binding": {"source": payload, "key": "temperature"}}
files = generate(element)
src = files["manager.py"]
assert "import os" not in src
assert "os.system" not in src
assert "non-config dynamic_text binding draws nothing" in src
ast.parse(src) # belt and braces: generate() already enforces this
-263
View File
@@ -1,263 +0,0 @@
"""Structural checks on the composer's JavaScript.
There is no JS test runner in this repo, but three defects here are structural
enough to assert on the parse tree, and each was a real bug:
- Stroke widths inside _drawElement were left in canvas pixels while the
geometry around them scaled by SCALE, so at SCALE>1 every outline rendered
thinner than one LED pixel and the preview stopped matching the panel.
- The `line` branch drew raw el.x0/el.y0/el.x1/el.y1, ignoring the anchor that
every other element type honours, so setting xAnchor moved everything except
lines -- and getBoundingBox had the same omission, leaving the hit box behind.
- Four methods mutated exactly what _snapshot() serialises (metadata,
currentPreset) without calling it. _snapshot is the only caller of
_debouncedAutosave, so those changes were lost on reload and could not be
undone.
"""
import re
from pathlib import Path
import pytest
tree_sitter = pytest.importorskip("tree_sitter")
tree_sitter_javascript = pytest.importorskip("tree_sitter_javascript")
JS_DIR = Path(__file__).resolve().parent.parent / "web_interface/static/v3/js/composer"
CANVAS = JS_DIR / "composer-canvas.js"
APP = JS_DIR / "composer-app.js"
def _function_source(path: Path, name: str) -> str:
"""Return the source of a top-level function declaration by name."""
src = path.read_bytes()
lang = tree_sitter.Language(tree_sitter_javascript.language())
tree = tree_sitter.Parser(lang).parse(src)
found = []
def walk(node):
if node.type == "function_declaration":
ident = node.child_by_field_name("name")
if ident is not None and src[ident.start_byte:ident.end_byte].decode() == name:
found.append(src[node.start_byte:node.end_byte].decode())
for c in node.children:
walk(c)
walk(tree.root_node)
assert found, f"{name} not found in {path.name}"
return found[0]
def _method_source(path: Path, name: str) -> str:
"""Return the source of a top-level object method by name."""
src = path.read_bytes()
lang = tree_sitter.Language(tree_sitter_javascript.language())
tree = tree_sitter.Parser(lang).parse(src)
found = []
def walk(node):
if node.type == "method_definition":
ident = node.child_by_field_name("name")
if ident is not None and src[ident.start_byte:ident.end_byte].decode() == name:
found.append(src[node.start_byte:node.end_byte].decode())
for c in node.children:
walk(c)
walk(tree.root_node)
assert found, f"{name} not found in {path.name}"
return found[0]
def test_both_files_parse():
lang = tree_sitter.Language(tree_sitter_javascript.language())
parser = tree_sitter.Parser(lang)
for path in (CANVAS, APP):
tree = parser.parse(path.read_bytes())
errors = []
def walk(node):
if node.type == "ERROR" or node.is_missing:
errors.append(node.start_point[0] + 1)
for c in node.children:
walk(c)
walk(tree.root_node)
assert not errors, f"{path.name} has parse errors at lines {errors}"
def test_element_strokes_scale_with_scale():
"""No bare `ctx.lineWidth = 1` inside _drawElement.
Selection handles and the grid are drawn in canvas pixels deliberately and
live in other functions, so this is scoped to the element drawing routine.
"""
body = _function_source(CANVAS, "_drawElement")
offenders = re.findall(r"ctx\.lineWidth\s*=\s*1\s*;", body)
assert not offenders, f"{len(offenders)} unscaled stroke width(s) in _drawElement"
def test_line_branch_applies_the_anchor_offset():
"""Scoped to _drawElement.
getBoundingBox has its own `case 'line': {` and appears first in the file,
so searching the whole text found *that* branch -- this assertion passed
with the draw branch's anchor offset removed. Verified: stripping it and
re-running gave 11/11 green.
"""
body = _function_source(CANVAS, "_drawElement")
line_branch = body[body.index("case 'line': {"):]
line_branch = line_branch[:line_branch.index("case 'divider'")]
assert "ax - el.x0" in line_branch and "ay - el.y0" in line_branch, \
"line drawing ignores xAnchor/yAnchor"
assert "moveTo(el.x0 * s" not in line_branch, \
"line still drawn from unanchored endpoints"
def test_line_bounding_box_applies_the_anchor_offset():
"""The companion to the above: scoped to getBoundingBox specifically, so
the two tests cannot both be satisfied by the same branch."""
body = _function_source(CANVAS, "getBoundingBox")
box = body[body.index("case 'line'"):]
box = box[:box.index("case 'divider'")]
assert "ax - el.x0" in box, "line bounding box ignores the anchor"
@pytest.mark.parametrize("method", [
"onBgColorChange", # mutates metadata.bgColor
"setCustomSize", # mutates currentPreset / MATRIX_W / MATRIX_H
"changePreset", # mutates currentPreset / MATRIX_W / MATRIX_H
"applyPresetLabel", # same, for sizes not in DISPLAY_PRESETS
"onColorChange", # the one that was already fixed — keeps it fixed
])
def test_state_mutations_take_a_snapshot(method):
body = _method_source(APP, method)
assert "_snapshot()" in body, \
f"{method} changes snapshotted state without calling _snapshot()"
assert "isDirty = true" in body, f"{method} does not mark the design dirty"
@pytest.mark.parametrize("method", ["changePreset", "applyPresetLabel"])
def test_restore_path_stays_snapshot_free(method):
"""_applyState and loadTemplate call these with {silent: true} while
restoring; snapshotting there would push restore steps onto the undo stack
and re-autosave the state just loaded."""
body = _method_source(APP, method)
assert "opts.silent" in body, f"{method} lost its silent guard"
snap = body.index("_snapshot()")
guard = body.index("!opts.silent")
assert guard < snap, f"{method} snapshots outside the !opts.silent guard"
TEMPLATE_HTML = (Path(__file__).resolve().parent.parent
/ "web_interface/templates/v3/composer.html")
#: The six toolbar buttons and the wrapper each must call.
ALIGN_BUTTONS = ["alignLeft", "alignHCenter", "alignRight",
"alignTop", "alignVCenter", "alignBottom"]
def test_alignment_buttons_use_the_anchor_clearing_path():
"""Two alignment implementations existed and the toolbar used the wrong one.
The legacy alignElement(dir) set el.x/el.y but left xAnchor/yAnchor in
place. resolveAnchor turns anchor='right' into `dim - val`, so "align left"
(el.x = 0) resolved to x = MATRIX_W -- the element jumped to the far right
edge instead. _alignElement clears the anchor first, so the stored value is
absolute, and it also updates el.x0/el.y0 so lines actually move.
"""
html = TEMPLATE_HTML.read_text()
for wrapper in ALIGN_BUTTONS:
assert f"{wrapper}()" in html, f"toolbar does not call {wrapper}()"
assert not re.search(r"[^_]alignElement\(", html), \
"toolbar still calls the legacy alignElement()"
def test_the_legacy_alignelement_is_gone():
"""Leaving it in place invites the toolbar drifting back to it."""
src = APP.read_text()
assert not re.search(r"^\s{4}alignElement\(dir\)", src, re.M), \
"legacy alignElement(dir) still defined"
def test_align_clears_the_anchor_and_moves_line_endpoints():
body = _method_source(APP, "_alignElement")
assert "xAnchor = null" in body and "yAnchor = null" in body, \
"_alignElement no longer clears the anchor, so aligning an anchored " \
"element resolves to the wrong edge"
assert "el.x0" in body and "el.y0" in body, \
"_alignElement no longer moves line endpoints"
def test_align_translates_both_line_endpoints_not_just_the_start():
"""Setting only x0 (or y0) left x1/y1 behind, so aligning a line changed
its shape instead of moving it -- e.g. a line from x0=20 to x1=50 aligned
right became x0=98, x1=50, stretching rather than translating it. Both
endpoints must move by the same delta."""
body = _method_source(APP, "_alignElement")
x_branch = body[body.index("if (axis === 'x')"):body.index("} else {")]
y_branch = body[body.index("} else {"):]
assert "el.x1" in x_branch, \
"_alignElement moves x0 but not x1 -- a line's shape changes, not its position"
assert "el.y1" in y_branch, \
"_alignElement moves y0 but not y1 -- a line's shape changes, not its position"
def test_divider_stroke_is_centered_in_led_pixels():
"""A 0.5 canvas-pixel offset (correct only at SCALE=1) was applied after
scaling instead of before it, so at SCALE>1 the stroke bled into the
preceding LED row/column instead of straddling its own."""
body = _function_source(CANVAS, "_drawElement")
divider_branch = body[body.index("case 'divider'"):]
divider_branch = divider_branch[:divider_branch.index("case 'pips'")]
assert "ay * s + 0.5" not in divider_branch, \
"divider still offsets by 0.5 canvas pixels after scaling"
assert "ax * s + 0.5" not in divider_branch, \
"divider still offsets by 0.5 canvas pixels after scaling"
assert "(ay + 0.5) * s" in divider_branch and "(ax + 0.5) * s" in divider_branch, \
"divider stroke is not centered within its LED pixel"
def test_gauge_radii_are_clamped_to_zero():
"""An imported design can carry a small gauge with a wide lineWidth --
width=1, height=1, lineWidth=3 sends a negative radius into
ctx.ellipse(), which throws IndexSizeError and aborts render() for every
element still to be drawn, not just the gauge."""
body = _function_source(CANVAS, "_drawElement")
gauge_branch = body[body.index("case 'gauge'"):]
assert re.search(r"Math\.max\(0,\s*rx\s*-\s*lwPx\s*/\s*2\)", gauge_branch), \
"gauge x-radius is not clamped to zero"
assert re.search(r"Math\.max\(0,\s*ry\s*-\s*lwPx\s*/\s*2\)", gauge_branch), \
"gauge y-radius is not clamped to zero"
assert "rx - lwPx / 2" not in re.sub(r"Math\.max\(0,\s*rx\s*-\s*lwPx\s*/\s*2\)", "", gauge_branch), \
"an unclamped gauge radius is still passed to ctx.ellipse()"
#: ELEMENT_DEFAULTS in composer-canvas.js gives exactly these types a
#: `binding` object; verified against that file rather than assumed.
BOUND_TYPES = ["dynamic_text", "progress_bar", "countdown", "pips", "sparkline", "gauge"]
def test_bound_types_constant_lists_every_bound_element_type():
"""ELEMENT_DEFAULTS in composer-canvas.js gives dynamic_text, progress_bar,
countdown, pips, sparkline and gauge a `binding` object. If a type is added
to (or removed from) that list without updating APP's BOUND_TYPES, the
checks below drift out of sync silently -- this pins the two together."""
src = APP.read_text()
match = re.search(r"const BOUND_TYPES = \[([^\]]*)\];", src)
assert match, "BOUND_TYPES constant not found in composer-app.js"
declared = {t.strip().strip("'\"") for t in match.group(1).split(",") if t.strip()}
assert declared == set(BOUND_TYPES), \
f"BOUND_TYPES {declared} does not match the bound element types {set(BOUND_TYPES)}"
@pytest.mark.parametrize("method", ["_isBound", "removeConfigVar", "_validateBeforeExport"])
def test_binding_checks_cover_every_bound_element_type(method):
"""Only dynamic_text and progress_bar were checked, so a countdown, pips,
sparkline or gauge element with an empty binding passed export validation
silently, and deleting a config var still used by one of them gave no
warning -- the plugin would read a now-missing key at runtime with no
indication why."""
body = _method_source(APP, method)
assert "BOUND_TYPES.includes(e.type)" in body, \
f"{method} does not check every bound element type via BOUND_TYPES"
assert "e.type === 'dynamic_text'" not in body, \
f"{method} still hardcodes only dynamic_text instead of BOUND_TYPES"
-225
View File
@@ -1,225 +0,0 @@
"""A composer plugin id must never resolve outside the plugins directory.
CodeQL reported sixteen high-severity py/path-injection alerts against
web_interface/blueprints/composer.py: a request-supplied plugin_id reaching
Path(plugins_dir) / plugin_id, which is then created, written to, deleted
(shutil.rmtree) and read back.
The id was already validated by an anchored regex, so every traversal payload
was in fact rejected. What was missing was the guarantee living *with* the path
building rather than in a regex several hundred lines away -- loosen that regex
later and the traversal opens silently, with nothing at the filesystem boundary
to catch it. _plugin_dir() closes that, and is the form static analysis can see.
"""
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from web_interface.blueprints import composer as C # noqa: E402
#: Anything that has ever been used to climb out of a directory.
TRAVERSAL = [
"../../etc/passwd", "..", ".", "a/../../etc", "good/../../..",
"/etc/passwd", "//etc/passwd", "a\\..\\..", "a%2f..%2f..",
"....//....//etc", "a/./../../etc", "~", "~root",
"plugin/../../../../../../etc/shadow",
]
#: Rejected for shape, not traversal -- but rejected all the same.
MALFORMED = ["", "A-upper", "1-leading-digit", "-leading-dash", "has_underscore",
"has space", "has.dot", "a" * 64, "plugin\n", "plugin\n../../etc",
"\n", "plug\x00in"]
@pytest.fixture
def plugins_dir(tmp_path, monkeypatch):
base = tmp_path / "plugin-repos"
base.mkdir()
monkeypatch.setattr(C.composer_bp, "plugins_dir", str(base), raising=False)
return base
@pytest.mark.parametrize("payload", TRAVERSAL)
def test_traversal_payloads_are_refused(plugins_dir, payload):
assert C._plugin_dir(payload) is None
@pytest.mark.parametrize("payload", MALFORMED)
def test_malformed_ids_are_refused(plugins_dir, payload):
assert C._plugin_dir(payload) is None
@pytest.mark.parametrize("payload", ["a", "my-plugin", "x9", "a" * 63])
def test_valid_ids_resolve_inside_the_base(plugins_dir, payload):
resolved = C._plugin_dir(payload)
assert resolved is not None, f"{payload!r} was rejected but is valid"
assert resolved.parent == plugins_dir.resolve(), (
f"{payload!r} resolved to {resolved}, outside {plugins_dir}")
def test_no_payload_can_escape_even_if_the_regex_is_loosened(plugins_dir, monkeypatch):
"""The containment check must stand on its own.
This is the whole point of resolving at the filesystem boundary: if the id
pattern is ever relaxed, traversal must still be impossible. Replace the
regex with one that permits slashes and dots, then re-run the payloads.
"""
import re
monkeypatch.setattr(C, "_PLUGIN_ID_RE", re.compile(r"\A[\w./\\~-]+\Z"))
import os
escaped = []
base = os.path.realpath(str(plugins_dir))
for payload in TRAVERSAL:
resolved = C._plugin_dir(payload)
if resolved is None:
continue
real = os.path.realpath(str(resolved))
if real != base and os.path.commonpath([base, real]) != base:
escaped.append((payload, real))
assert not escaped, f"these escaped the base with a loosened regex: {escaped}"
def test_a_sibling_directory_with_a_shared_prefix_is_not_inside(tmp_path, monkeypatch):
"""commonpath, not startswith.
"/x/plugins-evil" starts with "/x/plugins" but is a different directory, so
a prefix test would accept it.
"""
base = tmp_path / "plugins"
base.mkdir()
(tmp_path / "plugins-evil").mkdir()
monkeypatch.setattr(C.composer_bp, "plugins_dir", str(base), raising=False)
import re
# Neutralise the two layers in front so this exercises the containment
# check itself; otherwise secure_filename rejects the payload first and a
# startswith regression would go unnoticed here.
monkeypatch.setattr(C, "_PLUGIN_ID_RE", re.compile(r"\A[\w./\\~-]+\Z"))
monkeypatch.setattr(C, "secure_filename", lambda v: v)
assert C._plugin_dir("../plugins-evil") is None
def test_containment_still_holds_if_the_sanitiser_is_defeated(plugins_dir, monkeypatch):
"""Each layer is tested on its own, not just the stack.
secure_filename's equality guard rejects every traversal payload before the
containment check sees it, so removing containment does not fail the other
tests -- which would make it look load-bearing when it is not. Neutralise
the regex *and* the sanitiser, and the realpath/commonpath check must still
refuse everything on its own.
"""
import re
monkeypatch.setattr(C, "_PLUGIN_ID_RE", re.compile(r"\A[\w./\\~-]+\Z"))
monkeypatch.setattr(C, "secure_filename", lambda v: v)
import os
base = os.path.realpath(str(plugins_dir))
escaped = []
for payload in TRAVERSAL:
resolved = C._plugin_dir(payload)
if resolved is None:
continue
real = os.path.realpath(str(resolved))
# Inside the base is fine -- "...." and "~" are ordinary directory
# names on Linux, so they are not escapes. What must never happen is
# landing outside the base, or on the base itself: install() rmtrees
# its target, so the plugins root resolving to a "plugin" would wipe
# every installed plugin.
if real == base or os.path.commonpath([base, real]) != base:
escaped.append((payload, real))
assert not escaped, f"containment alone let these through: {escaped}"
def test_secure_filename_never_rewrites_an_accepted_id(plugins_dir):
"""The sanitiser must be a no-op on everything the regex accepts.
If secure_filename ever altered an accepted id, _plugin_dir would resolve
to a *different* plugin's directory than the caller asked for -- a silent
redirect, which is worse than a refusal. The guard turns that into a
refusal; this proves the guard never has to fire in practice.
"""
import random
from werkzeug.utils import secure_filename
random.seed(1)
alphabet = "abcdefghijklmnopqrstuvwxyz0123456789-"
altered = []
for _ in range(2000):
n = random.randint(1, 63)
cand = random.choice("abcdefghijklmnopqrstuvwxyz") + "".join(
random.choice(alphabet) for _ in range(n - 1))
if C._PLUGIN_ID_RE.match(cand) and secure_filename(cand) != cand:
altered.append((cand, secure_filename(cand)))
assert not altered, f"secure_filename rewrote accepted ids: {altered[:5]}"
def test_a_trailing_newline_is_not_a_valid_id():
r"""Python's `$` also matches before a trailing newline, so the original
`^...$` accepted "myplugin\n" and would have created a directory whose
name ends in one. \Z does not."""
assert C._PLUGIN_ID_RE.match("myplugin") is not None
assert C._PLUGIN_ID_RE.match("myplugin\n") is None
# --- font serving -----------------------------------------------------------
FONT_TRAVERSAL = [
"../../../etc/passwd", "../config/config.json", "..%2f..%2fetc%2fpasswd",
"PressStart2P-Regular.ttf/../../../etc/passwd", "/etc/passwd", "",
"PressStart2P-Regular.TTF", # case differs -> not the allowlisted name
"PressStart2P-Regular.ttf ", # trailing space
]
def test_serve_font_refuses_a_file_that_exists_but_is_not_allowlisted(monkeypatch, tmp_path):
"""The allowlist must be what refuses it, not a missing file.
Asserting 404 on traversal payloads proves nothing here: Flask's router
will not match a path segment containing '/', and everything else 404s
simply because no such file exists. Put a real, readable file next to the
fonts and confirm it is still refused -- that is the allowlist working.
"""
fonts = tmp_path / "assets" / "fonts"
fonts.mkdir(parents=True)
(fonts / "id_rsa.ttf").write_bytes(b"PRIVATE KEY")
monkeypatch.setattr(C.composer_bp, "project_root", str(tmp_path), raising=False)
app = __import__("flask").Flask(__name__)
app.register_blueprint(C.composer_bp)
with app.test_client() as client:
resp = client.get("/api/fonts/id_rsa.ttf")
assert resp.status_code == 404, (
"a readable non-allowlisted file was served; the allowlist is not gating")
assert b"PRIVATE KEY" not in resp.data
@pytest.mark.parametrize("payload", FONT_TRAVERSAL)
def test_serve_font_refuses_anything_not_allowlisted(payload, monkeypatch, tmp_path):
"""The name reaching the filesystem must come from the allowlist constant.
_ALLOWED_FONTS gates this endpoint, so nothing here was ever exploitable.
Building the path from the matched constant rather than the request value
is what makes that provable -- and it is why CodeQL reported two
high-severity py/path-injection alerts on an endpoint that was already
safe.
"""
monkeypatch.setattr(C.composer_bp, "project_root", str(tmp_path), raising=False)
app = C.composer_bp.name and __import__("flask").Flask(__name__)
app.register_blueprint(C.composer_bp)
with app.test_client() as client:
resp = client.get(f"/api/fonts/{payload}")
assert resp.status_code in (404, 405, 308), (
f"{payload!r} was not refused (status {resp.status_code})")
def test_serve_font_still_serves_each_allowlisted_font(monkeypatch, tmp_path):
fonts = tmp_path / "assets" / "fonts"
fonts.mkdir(parents=True)
monkeypatch.setattr(C.composer_bp, "project_root", str(tmp_path), raising=False)
app = __import__("flask").Flask(__name__)
app.register_blueprint(C.composer_bp)
for name in C._ALLOWED_FONTS:
(fonts / name).write_bytes(b"\x00\x01ttf")
with app.test_client() as client:
resp = client.get(f"/api/fonts/{name}")
assert resp.status_code == 200, f"{name} should be served, got {resp.status_code}"
@@ -0,0 +1,507 @@
"""The Starlark routes the frontend calls must exist.
`plugins_manager.js` posts to /api/v3/starlark/install-pixlet and then reloads
/api/v3/starlark/status. Neither route existed: #330 rewrote api_v3.py and
dropped all thirteen Starlark routes that #253 had added, so both calls fell
through to Flask's 404 handler, which answers
{"status": "error", "message": "Resource not found"}
and the button reported "Pixlet install failed: Resource not found" -- a
message that names neither the resource nor the cause.
These assert the routes are registered and answer in the shape the frontend
reads, so a future rewrite of this file cannot silently drop them again.
"""
import json
import os
import sys
import tempfile
import threading
import types
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture
def client():
from web_interface.app import app
app.config['TESTING'] = True
with app.test_client() as c:
yield c
@pytest.fixture
def offline_repository():
"""Stand in for TronbyteRepository so the store routes stay offline.
The browse and categories routes call list_all_apps_cached(), which on a
cold server-side cache fetches the app index from GitHub. That would make
the suite slow, network-dependent and rate-limitable -- and because these
tests only assert "not a 404", a rate-limited 500 would pass and hide it.
"""
repo = MagicMock()
repo.list_all_apps_cached.return_value = {
'apps': [], 'categories': [], 'authors': [], 'count': 0, 'cached': True,
}
with patch('web_interface.blueprints.api_v3._get_tronbyte_repository_class',
return_value=lambda *a, **kw: repo):
yield repo
class TestRoutesAreRegistered:
"""The failure was a missing route, so check the URL map directly.
All thirteen, not just the two the Pixlet button needs: #330 dropped the
lot, and the app store page is built on repository/browse,
repository/categories and repository/install, which 404 the same way.
"""
@pytest.mark.parametrize("rule,method", [
("/api/v3/starlark/install-pixlet", "POST"),
("/api/v3/starlark/status", "GET"),
("/api/v3/starlark/apps", "GET"),
("/api/v3/starlark/upload", "POST"),
("/api/v3/starlark/repository/browse", "GET"),
("/api/v3/starlark/repository/categories", "GET"),
("/api/v3/starlark/repository/install", "POST"),
("/api/v3/starlark/apps/<app_id>", "GET"),
("/api/v3/starlark/apps/<app_id>", "DELETE"),
("/api/v3/starlark/apps/<app_id>/config", "GET"),
("/api/v3/starlark/apps/<app_id>/config", "PUT"),
("/api/v3/starlark/apps/<app_id>/toggle", "POST"),
("/api/v3/starlark/apps/<app_id>/render", "POST"),
])
def test_route_exists(self, rule, method):
from web_interface.app import app
matches = [r for r in app.url_map.iter_rules()
if r.rule == rule and method in r.methods]
assert matches, (
f"{method} {rule} is not registered; the frontend calls it and "
"would get Flask's generic 'Resource not found'")
class TestInstallPixlet:
def test_it_does_not_404(self, client):
with patch('web_interface.blueprints.api_v3.subprocess.run') as run:
run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
resp = client.post('/api/v3/starlark/install-pixlet')
assert resp.status_code != 404, "the route is still missing"
assert resp.get_json().get('message') != 'Resource not found'
def test_success_is_reported_in_the_shape_the_button_reads(self, client):
with patch('web_interface.blueprints.api_v3.subprocess.run') as run:
run.return_value = MagicMock(returncode=0, stdout="done", stderr="")
resp = client.post('/api/v3/starlark/install-pixlet')
body = resp.get_json()
assert body['status'] == 'success', body
assert 'message' in body, "the JS shows data.message on success"
def test_a_failed_download_says_why(self, client):
with patch('web_interface.blueprints.api_v3.subprocess.run') as run:
run.return_value = MagicMock(returncode=1, stdout="", stderr="no such release")
resp = client.post('/api/v3/starlark/install-pixlet')
body = resp.get_json()
assert body['status'] == 'error'
assert 'no such release' in body['message'], \
"the installer's own stderr is what tells the user what went wrong"
def test_a_timeout_is_reported_rather_than_hanging(self, client):
import subprocess as sp
with patch('web_interface.blueprints.api_v3.subprocess.run',
side_effect=sp.TimeoutExpired(cmd='x', timeout=300)):
resp = client.post('/api/v3/starlark/install-pixlet')
assert resp.get_json()['status'] == 'error'
assert 'timed out' in resp.get_json()['message'].lower()
class TestStarlarkStatus:
def test_it_does_not_404(self, client):
resp = client.get('/api/v3/starlark/status')
assert resp.status_code != 404, "the route is still missing"
assert resp.get_json().get('message') != 'Resource not found'
def test_it_reports_pixlet_availability_without_the_plugin_loaded(self, client):
# The status call runs before install too -- it must answer even when
# starlark-apps is not loaded, which is the state a user is in when
# they press the install button for the first time.
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None):
resp = client.get('/api/v3/starlark/status')
body = resp.get_json()
assert body['status'] == 'success', body
assert 'pixlet_available' in body
assert body['plugin_loaded'] is False
class TestTheInstallerScriptIsActuallyThere:
def test_download_pixlet_script_exists_and_is_executable(self):
# install_pixlet chmods and runs this; a missing file is the one error
# it reports as a 404 of its own, which would look identical to the
# bug being fixed here.
import os
from pathlib import Path
from web_interface.blueprints.api_v3 import PROJECT_ROOT
script = Path(PROJECT_ROOT) / 'scripts' / 'download_pixlet.sh'
assert script.is_file(), f"{script} is missing; install_pixlet would 404"
assert os.access(script, os.R_OK)
class TestTheAppStoreFlow:
"""Browsing and installing from the Tronbyte repository.
These are the calls the app store page makes. Each returned the generic
"Resource not found" before this change, which is indistinguishable from
an empty store.
"""
def test_browse_does_not_404(self, client, offline_repository):
resp = client.get('/api/v3/starlark/repository/browse')
assert resp.status_code != 404, "the store cannot list anything"
assert resp.get_json().get('message') != 'Resource not found'
def test_categories_does_not_404(self, client, offline_repository):
resp = client.get('/api/v3/starlark/repository/categories')
assert resp.status_code != 404
assert resp.get_json().get('message') != 'Resource not found'
def test_installed_apps_list_does_not_404(self, client):
resp = client.get('/api/v3/starlark/apps')
assert resp.status_code != 404
assert resp.get_json().get('message') != 'Resource not found'
def test_repository_install_rejects_a_missing_body_rather_than_404ing(self, client):
# A 400/422 here is the route working: it received the call and said
# what was wrong. A 404 means it was never reached at all.
resp = client.post('/api/v3/starlark/repository/install',
json={}, content_type='application/json')
assert resp.status_code != 404, "the install route is still missing"
assert resp.get_json().get('message') != 'Resource not found'
def test_upload_rejects_an_empty_post_rather_than_404ing(self, client):
resp = client.post('/api/v3/starlark/upload')
assert resp.status_code != 404
assert resp.get_json().get('message') != 'Resource not found'
class TestNoStarlarkRouteIsMissing:
"""A single check that the whole set is present.
#330 removed all thirteen at once by rewriting this file. One assertion
over the frontend's own list is what would have caught that.
"""
def test_every_endpoint_the_frontend_calls_is_registered(self):
import re
from pathlib import Path
from werkzeug.exceptions import MethodNotAllowed, NotFound
from web_interface.app import app
root = Path(__file__).resolve().parent.parent.parent
js = (root / 'web_interface' / 'static' / 'v3' / 'plugins_manager.js').read_text()
# The frontend builds some of these with template literals, e.g.
# `/api/v3/starlark/apps/${appId}/toggle`. Substitute a placeholder so
# the URL is concrete, then let Werkzeug match it the way a request
# would -- string comparison cannot see <app_id> rules.
raw = set(re.findall(r"[\'\"`](/api/v3/starlark/[^\'\"`\s]*)", js))
urls = set()
for u in raw:
u = re.sub(r"\$\{[^}]*\}", "probe", u)
urls.add(u.rstrip('/') or u)
assert urls, "found no starlark calls in the frontend -- did the file move?"
adapter = app.url_map.bind('localhost')
missing = []
for u in sorted(urls):
try:
adapter.match(u, method='GET')
except MethodNotAllowed:
pass # route exists, just not for GET -- fine
except NotFound:
missing.append(u)
assert not missing, f"the frontend calls these and they are not registered: {missing}"
class TestInstalledAppsAppearWithTheOtherPlugins:
"""An installed .star app must be manageable like any other plugin.
#253 surfaced installed apps in /plugins/installed as `starlark:<app_id>`
entries, so they could be seen and enabled/disabled from the same list as
everything else, and routed `starlark:` toggles to the Starlark manifest.
#330 removed both. The result was an app that installs successfully, then
appears nowhere and cannot be turned on or off.
"""
APPS = {'apps': {'quoteoftheday': {'name': 'A Quote A Day', 'enabled': True}}}
def test_an_installed_app_is_listed(self, client):
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest', return_value=self.APPS):
resp = client.get('/api/v3/plugins/installed')
ids = [p['id'] for p in resp.get_json()['data']['plugins']]
assert 'starlark:quoteoftheday' in ids, \
"an installed Starlark app does not appear among the plugins"
def test_the_entry_carries_what_the_ui_needs(self, client):
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest', return_value=self.APPS):
resp = client.get('/api/v3/plugins/installed')
entry = next(p for p in resp.get_json()['data']['plugins']
if p['id'] == 'starlark:quoteoftheday')
assert entry['name'] == 'A Quote A Day'
assert entry['enabled'] is True
assert entry['is_starlark_app'] is True, "the UI keys its Starlark handling off this"
assert entry['category'] == 'Starlark App'
def test_a_starlark_failure_does_not_empty_the_plugin_list(self, client):
# The virtual entries are appended to the real ones; a broken manifest
# must cost the Starlark rows, not everybody else's.
with patch('web_interface.blueprints.api_v3._get_starlark_plugin',
side_effect=RuntimeError('boom')):
resp = client.get('/api/v3/plugins/installed')
assert resp.status_code == 200
assert resp.get_json()['status'] == 'success'
def test_toggling_an_app_does_not_report_plugin_not_found(self, client):
written = {}
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest',
return_value={'apps': {'quoteoftheday': {'enabled': True}}}), \
patch('web_interface.blueprints.api_v3._write_starlark_manifest',
side_effect=lambda m: written.update(m) or True):
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:quoteoftheday', 'enabled': False})
body = resp.get_json()
assert body['status'] == 'success', body
assert body['enabled'] is False
assert written['apps']['quoteoftheday']['enabled'] is False, \
"the manifest was not actually updated"
def test_toggling_an_unknown_app_says_so(self, client):
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest',
return_value={'apps': {}}):
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:nope', 'enabled': True})
assert resp.status_code == 404
assert 'nope' in resp.get_json()['message']
def test_a_traversal_app_id_is_rejected_before_touching_the_manifest(self, client):
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:../../etc/passwd', 'enabled': True})
assert resp.status_code == 400
assert 'traversal' in resp.get_json()['message']
def test_an_app_id_the_listing_published_can_be_toggled(self, client):
"""The id here is the one _starlark_virtual_plugins publishes.
It used to be re-slugified on the way in -- lowercased, with every
character outside [a-z0-9_] replaced -- so 'My-App' was listed as
'starlark:My-App' and looked up as 'my_app', and toggling an app the
page had just drawn answered 404.
"""
manifest = {'apps': {'My-App': {'name': 'My App', 'enabled': False}}}
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest',
return_value=manifest), \
patch('web_interface.blueprints.api_v3._write_starlark_manifest',
return_value=True) as write:
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:My-App', 'enabled': True})
assert resp.status_code == 200, resp.get_json()
assert write.called, "the toggle never reached the manifest"
assert manifest['apps']['My-App']['enabled'] is True
def test_a_failed_manifest_write_is_not_reported_as_success(self, client):
"""The toggle answered 200 while the change was never persisted."""
manifest = {'apps': {'demo': {'name': 'Demo', 'enabled': False}}}
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=None), \
patch('web_interface.blueprints.api_v3._read_starlark_manifest',
return_value=manifest), \
patch('web_interface.blueprints.api_v3._write_starlark_manifest',
return_value=False):
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:demo', 'enabled': True})
assert resp.status_code == 500
def test_a_loaded_app_is_not_flipped_when_the_manifest_write_fails(self, client):
"""Persist first, then update memory -- otherwise the UI shows a
toggle that silently reverts on the next restart."""
app = MagicMock()
app.manifest = {'enabled': False}
plugin = MagicMock()
plugin.apps = {'demo': app}
plugin._update_manifest_safe.return_value = False
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=plugin):
resp = client.post('/api/v3/plugins/toggle',
json={'plugin_id': 'starlark:demo', 'enabled': True})
assert resp.status_code == 500
assert app.manifest['enabled'] is False, "in-memory state changed without being saved"
class TestTheManifestSurvivesConcurrentWriters:
"""Flask serves requests concurrently and five routes write this file.
The atomic-write pattern used one fixed temp name, `manifest.tmp`, shared
by every writer: two of them opened it, interleaved their json.dump output,
and both renamed. The rename is atomic; the content it published was the
mixture, which the next read could not parse.
"""
@pytest.fixture
def starlark_dir(self, tmp_path, monkeypatch):
from web_interface.blueprints import api_v3 as module
apps_dir = tmp_path / "starlark-apps"
apps_dir.mkdir()
monkeypatch.setattr(module, '_STARLARK_APPS_DIR', apps_dir)
monkeypatch.setattr(module, '_STARLARK_MANIFEST_FILE', apps_dir / 'manifest.json')
return apps_dir
def test_each_writer_gets_its_own_temp_file(self, starlark_dir):
from web_interface.blueprints import api_v3 as module
seen = []
real_mkstemp = tempfile.mkstemp
def recording_mkstemp(*args, **kwargs):
fd, name = real_mkstemp(*args, **kwargs)
seen.append(name)
return fd, name
with patch.object(module.tempfile, 'mkstemp', side_effect=recording_mkstemp):
for i in range(5):
assert module._write_starlark_manifest({'apps': {f'app{i}': {}}})
assert len(set(seen)) == 5, f"writers shared a temp file: {seen}"
def test_concurrent_writes_leave_readable_json(self, starlark_dir):
from web_interface.blueprints import api_v3 as module
manifests = [{'apps': {f'app{i}': {'name': 'x' * 400}}} for i in range(8)]
errors = []
def write(m):
try:
module._write_starlark_manifest(m)
except Exception as exc: # noqa: BLE001 - surfaced by the assert below
errors.append(exc)
threads = [threading.Thread(target=write, args=(m,)) for m in manifests]
for t in threads:
t.start()
for t in threads:
t.join()
assert not errors
loaded = json.loads((starlark_dir / 'manifest.json').read_text())
assert loaded in manifests, "the published manifest was a mix of two writers"
def test_no_temp_files_are_left_behind(self, starlark_dir):
from web_interface.blueprints import api_v3 as module
module._write_starlark_manifest({'apps': {}})
assert list(starlark_dir.glob('*.tmp')) == []
def test_the_installed_star_file_is_recorded_by_name(self, starlark_dir, tmp_path):
"""Readers join this to the app's own directory and default to a bare
'<app_id>.star'. An absolute value gave the key two meanings and, since
Path.__truediv__ discards the left side when the right is absolute,
pinned the manifest to the PROJECT_ROOT it was installed under."""
from web_interface.blueprints import api_v3 as module
source = tmp_path / "source.star"
source.write_text("# app")
with patch.object(module, '_get_pixlet_renderer_class',
side_effect=ImportError("no pixlet here")):
assert module._install_star_file('demo', str(source), {'name': 'Demo'})
entry = json.loads((starlark_dir / 'manifest.json').read_text())['apps']['demo']
assert entry['star_file'] == 'demo.star'
assert not os.path.isabs(entry['star_file'])
class TestATransientImportFailureIsNotPermanent:
"""Both importers insert into sys.modules before executing the module.
That order is required -- a module has to be findable while it runs -- but
a failure left the half-initialised object cached, so every later call took
the cache branch and raised AttributeError on the missing class instead of
retrying. One transient failure disabled the repository or the renderer for
the life of the process.
"""
@pytest.mark.parametrize("getter,key", [
('_get_tronbyte_repository_class', 'tronbyte_repository'),
('_get_pixlet_renderer_class', 'pixlet_renderer'),
])
def test_a_failed_import_leaves_no_entry_behind(self, getter, key):
from web_interface.blueprints import api_v3 as module
original = sys.modules.pop(key, None)
try:
with patch('importlib.util.module_from_spec') as from_spec:
from_spec.return_value = types.ModuleType(key)
with patch('importlib.util.spec_from_file_location') as spec_from:
spec = MagicMock()
spec.loader.exec_module.side_effect = RuntimeError("network down")
spec_from.return_value = spec
with pytest.raises(RuntimeError):
getattr(module, getter)()
assert key not in sys.modules, "a half-initialised module stayed cached"
finally:
if original is not None:
sys.modules[key] = original
else:
sys.modules.pop(key, None)
class TestConfigIsNotAppliedUntilItIsSaved:
"""save_config() returning False answers 500, but the loaded app kept the
new values -- so GET config reported settings that were never written and
the plugin rendered with them until a restart silently reverted them."""
def _plugin_with_app(self, save_ok):
app = MagicMock()
app.config = {'city': 'Philadelphia'}
app.manifest = {'render_interval': 300, 'display_duration': 15}
app.save_config.return_value = save_ok
plugin = MagicMock()
plugin.apps = {'demo': app}
plugin._update_manifest_safe.return_value = True
return plugin, app
def test_a_failed_save_leaves_the_config_untouched(self, client):
plugin, app = self._plugin_with_app(save_ok=False)
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=plugin):
resp = client.put('/api/v3/starlark/apps/demo/config',
json={'city': 'Pittsburgh'})
assert resp.status_code == 500
assert app.config == {'city': 'Philadelphia'}
def test_a_failed_save_leaves_the_timing_untouched(self, client):
plugin, app = self._plugin_with_app(save_ok=False)
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=plugin):
resp = client.put('/api/v3/starlark/apps/demo/config',
json={'render_interval': 60})
assert resp.status_code == 500
assert app.manifest['render_interval'] == 300
def test_a_failed_save_does_not_re_render_with_values_it_did_not_keep(self, client):
plugin, _ = self._plugin_with_app(save_ok=False)
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=plugin):
client.put('/api/v3/starlark/apps/demo/config', json={'city': 'Pittsburgh'})
plugin._render_app.assert_not_called()
def test_a_successful_save_still_applies(self, client):
plugin, app = self._plugin_with_app(save_ok=True)
with patch('web_interface.blueprints.api_v3._get_starlark_plugin', return_value=plugin):
resp = client.put('/api/v3/starlark/apps/demo/config',
json={'city': 'Pittsburgh', 'render_interval': 60})
assert resp.status_code == 200
assert app.config['city'] == 'Pittsburgh'
assert app.manifest['render_interval'] == 60
plugin._render_app.assert_called_once()
+1 -8
View File
@@ -83,11 +83,11 @@ plugin_system_config = config.get('plugin_system', {})
plugins_dir_name = plugin_system_config.get('plugins_directory', 'plugin-repos') plugins_dir_name = plugin_system_config.get('plugins_directory', 'plugin-repos')
# Resolve plugin directory - handle both absolute and relative paths # Resolve plugin directory - handle both absolute and relative paths
project_root = Path(__file__).parent.parent
if os.path.isabs(plugins_dir_name): if os.path.isabs(plugins_dir_name):
plugins_dir = Path(plugins_dir_name) plugins_dir = Path(plugins_dir_name)
else: else:
# If relative, resolve relative to the project root (LEDMatrix directory) # If relative, resolve relative to the project root (LEDMatrix directory)
project_root = Path(__file__).parent.parent
plugins_dir = project_root / plugins_dir_name plugins_dir = project_root / plugins_dir_name
plugin_manager = PluginManager( plugin_manager = PluginManager(
@@ -157,7 +157,6 @@ _health_monitor_initialized = False
# Register blueprints # Register blueprints
from web_interface.blueprints.pages_v3 import pages_v3 from web_interface.blueprints.pages_v3 import pages_v3
from web_interface.blueprints.api_v3 import api_v3 from web_interface.blueprints.api_v3 import api_v3
from web_interface.blueprints.composer import composer_bp
# Initialize managers in blueprints # Initialize managers in blueprints
pages_v3.config_manager = config_manager pages_v3.config_manager = config_manager
@@ -174,11 +173,6 @@ api_v3.operation_queue = operation_queue
api_v3.plugin_state_manager = plugin_state_manager api_v3.plugin_state_manager = plugin_state_manager
api_v3.operation_history = operation_history api_v3.operation_history = operation_history
api_v3.health_monitor = health_monitor api_v3.health_monitor = health_monitor
composer_bp.config_manager = config_manager
composer_bp.plugin_manager = plugin_manager
composer_bp.plugins_dir = plugins_dir
composer_bp.project_root = project_root
# Initialize cache manager for API endpoints # Initialize cache manager for API endpoints
from src.cache_manager import CacheManager from src.cache_manager import CacheManager
api_v3.cache_manager = CacheManager() api_v3.cache_manager = CacheManager()
@@ -206,7 +200,6 @@ except Exception as _hm_err: # pragma: no cover - defensive startup guard
app.register_blueprint(pages_v3, url_prefix='') app.register_blueprint(pages_v3, url_prefix='')
app.register_blueprint(pages_v3, url_prefix='/v3', name='pages_v3_legacy') app.register_blueprint(pages_v3, url_prefix='/v3', name='pages_v3_legacy')
app.register_blueprint(api_v3, url_prefix='/api/v3') app.register_blueprint(api_v3, url_prefix='/api/v3')
app.register_blueprint(composer_bp, url_prefix='/composer')
# Route to serve plugin asset files (registered on main app, not blueprint, for /assets/... path) # Route to serve plugin asset files (registered on main app, not blueprint, for /assets/... path)
@app.route('/assets/plugins/<plugin_id>/uploads/<path:filename>', methods=['GET']) @app.route('/assets/plugins/<plugin_id>/uploads/<path:filename>', methods=['GET'])
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
-3
View File
@@ -426,9 +426,6 @@ a, button, input, select, textarea {
.md\:hidden { display: none; } .md\:hidden { display: none; }
.md\:block { display: block; } .md\:block { display: block; }
.md\:w-auto { width: auto; } .md\:w-auto { width: auto; }
/* composer.html labels its toolbar buttons `hidden md:inline`, so without
this the label is hidden at every width and the buttons stay icon-only. */
.md\:inline { display: inline; }
} }
@media (min-width: 1024px) { @media (min-width: 1024px) {
File diff suppressed because it is too large Load Diff
@@ -1,817 +0,0 @@
/**
* ComposerCanvas — stateless LED matrix canvas renderer.
*
* Coordinate system: LED pixels (integers). All drawing multiplies by SCALE.
* PIL draw.text(x,y) is top-left; canvas fillText(x,y) is baseline.
* → Canvas text cy = (actualY + fontSizePx) * SCALE
*
* Anchors: element x/y are offsets from their anchor point:
* xAnchor=null/'left' → x is fixed offset from left
* xAnchor='center' → x is offset from width/2
* xAnchor='right' → x is offset inward from right edge
* yAnchor follows the same pattern with 'top'/'middle'/'bottom'
*
* Breakpoints: elements with minWidth > currentMatrixW are rendered at 25% opacity.
*
* Resize handles: drawn on selected rectangles; 8 handles (corners + edge mids).
*/
/* eslint-disable security/detect-object-injection --
Every hit in this file is either an array index (this.elements[idx],
rawVals[i]) or a lookup on a module-private map keyed by an internal
element type (_HANDLE_CURSORS, ELEMENT_DEFAULTS, previewValues). None of
them takes an attacker-supplied property name, so none can reach a
prototype. Disabled per file rather than eight times inline. */
window.ComposerCanvas = (() => {
'use strict';
let _canvas = null;
let _ctx = null;
let _showGrid = true;
//: Element types the canvas draws resize handles for. Exported because the
//: editor has to gate its resize and hover behaviour on exactly this list --
//: the two had drifted, so handles appeared on five shapes that could not
//: actually be resized.
const RESIZABLE_TYPES = ['rectangle', 'rounded_rectangle', 'ellipse', 'arc',
'gauge', 'sparkline'];
const DISPLAY_PRESETS = [
{ label: '64×32', w: 64, h: 32 },
{ label: '128×32', w: 128, h: 32 },
{ label: '128×64', w: 128, h: 64 },
{ label: '256×32', w: 256, h: 32 },
{ label: '256×64', w: 256, h: 64 },
];
const FONT_MAP = {
press_start: { family: "'PressStart2P', monospace", sizePx: 8, charW: 8 },
four_by_six: { family: 'monospace', sizePx: 6, charW: 4 },
five_by_seven: { family: 'monospace', sizePx: 7, charW: 5 },
};
const ELEMENT_DEFAULTS = {
text: {
text: 'Hello', font: 'press_start',
r: 255, g: 255, b: 255,
text2: '', lineSpacing: 2, textAlign: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
dynamic_text: {
binding: { source: 'config', key: '', format: null },
font: 'press_start', textAlign: 'left',
r: 255, g: 200, b: 100,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
clock: {
format: '%H:%M', font: 'press_start',
r: 100, g: 255, b: 100,
format2: '', lineSpacing: 2, textAlign: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
rectangle: {
width: 20, height: 8,
fillR: 0, fillG: 0, fillB: 128, hasFill: true,
outR: 255, outG: 255, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
ellipse: {
width: 24, height: 12,
fillR: 0, fillG: 100, fillB: 200, hasFill: true,
outR: 100, outG: 180, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
arc: {
width: 24, height: 24,
startAngle: 0, endAngle: 270, lineWidth: 2,
r: 255, g: 200, b: 0,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
pixel: {
r: 255, g: 255, b: 255,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
rounded_rectangle: {
width: 24, height: 10, borderRadius: 3,
fillR: 0, fillG: 80, fillB: 180, hasFill: true,
outR: 120, outG: 180, outB: 255, hasOutline: true,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
line: {
x0: 0, y0: 16, x1: 63, y1: 16,
r: 180, g: 180, b: 180, lineWidth: 1,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
divider: {
orientation: 'horizontal', y: 16, x: 64,
r: 100, g: 100, b: 100,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
progress_bar: {
barWidth: 60, barHeight: 6,
binding: { source: 'config', key: '', format: null },
r: 80, g: 200, b: 80,
bgR: 30, bgG: 30, bgB: 30, hasBg: true,
outR: 100, outG: 100, outB: 100, hasOutline: true,
previewPct: 65,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
countdown: {
binding: { source: 'config', key: '', format: null },
countdownFormat: 'dh',
font: 'four_by_six', textAlign: 'left',
r: 255, g: 180, b: 0,
previewText: '42d 3h',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
marquee: {
text: 'Scrolling text', font: 'press_start',
r: 255, g: 255, b: 255,
scrollSpeed: 1, gap: 16, direction: 'left',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
section: {
label: 'Section',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
pips: {
count: 5, filled: 3, pipSize: 4, pipSpacing: 2,
r: 255, g: 200, b: 0,
emptyR: 50, emptyG: 50, emptyB: 50, showEmpty: true,
binding: { source: 'config', key: '', format: null },
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
sparkline: {
width: 40, height: 12,
barCount: 8, barSpacing: 1,
r: 80, g: 200, b: 120,
bgR: 30, bgG: 30, bgB: 30, hasBg: false,
binding: { source: 'config', key: '', format: null },
previewData: '0.3,0.6,0.4,0.8,0.5,0.9,0.7,0.85',
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
gauge: {
width: 32, height: 32,
startAngle: 135, endAngle: 45, lineWidth: 3,
binding: { source: 'config', key: '', format: null },
r: 80, g: 220, b: 80,
trackR: 40, trackG: 40, trackB: 40, hasTrack: true,
showLabel: true, font: 'four_by_six', labelR: 200, labelG: 200, labelB: 200,
previewPct: 65,
xAnchor: null, yAnchor: null, minWidth: 0, locked: false, blink: false, visible: true,
},
};
// ── Anchor resolution ────────────────────────────────────────────────
function resolveAnchor(val, anchor, dim) {
if (!anchor || anchor === 'left' || anchor === 'top') return val;
if (anchor === 'center' || anchor === 'middle') return Math.floor(dim / 2) + val;
if (anchor === 'right' || anchor === 'bottom') return dim - val;
return val;
}
function computeActualPos(el, matrixW, matrixH) {
const ax = resolveAnchor(el.x ?? el.x0 ?? 0, el.xAnchor, matrixW);
const ay = resolveAnchor(el.y ?? el.y0 ?? 0, el.yAnchor, matrixH);
return { x: ax, y: ay };
}
// ── Bounding box (LED pixel space) ──────────────────────────────────
function getBoundingBox(el, matrixW, matrixH) {
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
switch (el.type) {
case 'text': {
const t1 = el.text || '', t2 = el.text2 || '';
const w = Math.max(t1.length, t2.length) * finfo.charW;
const h = t2 ? finfo.sizePx * 2 + (el.lineSpacing ?? 2) : finfo.sizePx;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h };
}
case 'dynamic_text': {
const key = el.binding?.key || '?';
const w = (`{${key}}`).length * finfo.charW;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h: finfo.sizePx };
}
case 'clock': {
const t1 = el.format || '%H:%M', t2 = el.format2 || '';
const w = Math.max(t1.length, t2.length) * finfo.charW;
const h = t2 ? finfo.sizePx * 2 + (el.lineSpacing ?? 2) : finfo.sizePx;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h };
}
case 'countdown': {
const pt = el.previewText || '--d --h';
const w = pt.length * finfo.charW;
const bx = el.textAlign === 'center' ? ax - w / 2 : el.textAlign === 'right' ? ax - w : ax;
return { x: bx, y: ay, w, h: finfo.sizePx };
}
case 'rectangle':
case 'rounded_rectangle':
case 'ellipse':
case 'arc':
return { x: ax, y: ay, w: el.width, h: el.height };
case 'pixel':
return { x: ax, y: ay, w: 1, h: 1 };
case 'line': {
// Same anchor offset the draw path applies, or the hit box sits where
// the line used to be.
const dx = ax - el.x0, dy = ay - el.y0;
return {
x: Math.min(el.x0, el.x1) + dx, y: Math.min(el.y0, el.y1) + dy,
w: Math.max(1, Math.abs(el.x1 - el.x0)),
h: Math.max(1, Math.abs(el.y1 - el.y0)),
};
}
case 'divider':
return el.orientation === 'horizontal'
? { x: 0, y: ay, w: matrixW, h: 1 }
: { x: ax, y: 0, w: 1, h: matrixH };
case 'progress_bar':
return { x: ax, y: ay, w: el.barWidth ?? 60, h: el.barHeight ?? 6 };
case 'marquee': {
const mfinfo = FONT_MAP[el.font] || FONT_MAP.press_start;
return { x: 0, y: ay, w: matrixW, h: mfinfo.sizePx };
}
case 'gauge':
return { x: ax, y: ay, w: el.width ?? 32, h: el.height ?? 32 };
case 'sparkline':
return { x: ax, y: ay, w: el.width ?? 40, h: el.height ?? 12 };
case 'pips': {
const pc = el.count ?? 5, ps = el.pipSize ?? 4, pg = el.pipSpacing ?? 2;
return { x: ax, y: ay, w: pc * ps + (pc - 1) * pg, h: ps };
}
case 'section': {
// Was 0x0, so the element was unselectable except through the 3px
// hit-test padding and drew nothing at all -- a user adding one from
// the palette saw an empty canvas.
// Same font resolution as the draw case below, or the box will not
// match the glyphs: getBoundingBox's shared `finfo` falls back to
// press_start, and a section has no font of its own.
const sinfo = FONT_MAP[el.font] || FONT_MAP.four_by_six;
const label = el.label || 'Section';
return { x: ax, y: ay, w: label.length * sinfo.charW, h: sinfo.sizePx };
}
default:
return { x: ax, y: ay, w: 4, h: 4 };
}
}
// ── Resize handle support ─────────────────────────────────────────────
// Returns 8 handle points for a rectangle in LED pixel space
function _getRectHandles(el, matrixW, matrixH) {
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const w = el.width, h = el.height;
const cx = ax + w / 2, cy = ay + h / 2;
return {
nw: { x: ax, y: ay },
n: { x: cx, y: ay },
ne: { x: ax + w, y: ay },
w: { x: ax, y: cy },
e: { x: ax + w, y: cy },
sw: { x: ax, y: ay + h },
s: { x: cx, y: ay + h },
se: { x: ax + w, y: ay + h },
};
}
// Returns the handle direction under LED-space point (lx, ly), or null
function getResizeHandle(el, lx, ly, matrixW, matrixH) {
if (!RESIZABLE_TYPES.includes(el.type)) return null;
const handles = _getRectHandles(el, matrixW, matrixH);
const PAD = 4;
for (const [dir, pt] of Object.entries(handles)) {
if (Math.abs(lx - pt.x) <= PAD && Math.abs(ly - pt.y) <= PAD) return dir;
}
return null;
}
const _HANDLE_CURSORS = {
nw: 'nw-resize', n: 'n-resize', ne: 'ne-resize',
w: 'w-resize', e: 'e-resize',
sw: 'sw-resize', s: 's-resize', se: 'se-resize',
};
function getCursorForHandle(handle) {
return _HANDLE_CURSORS[handle] || 'crosshair';
}
// ── Hit test ─────────────────────────────────────────────────────────
function hitTest(el, lx, ly, matrixW, matrixH) {
const PAD = 3;
const bb = getBoundingBox(el, matrixW, matrixH);
return (
lx >= bb.x - PAD && lx <= bb.x + bb.w + PAD &&
ly >= bb.y - PAD && ly <= bb.y + bb.h + PAD
);
}
// ── Draw a single element ─────────────────────────────────────────────
function _drawElement(ctx, el, SCALE, matrixW, matrixH, opts = {}) {
const s = SCALE;
const { x: ax, y: ay } = computeActualPos(el, matrixW, matrixH);
const belowBreakpoint = el.minWidth > 0 && matrixW < el.minWidth;
const hidden = el.visible === false;
ctx.save();
if (hidden) ctx.globalAlpha = 0.12;
else if (belowBreakpoint) ctx.globalAlpha = 0.25;
// Blink animation: when blinkOff, fully hide blinking elements
if (el.blink) {
if (opts.blinkOff) { ctx.restore(); return; }
ctx.globalAlpha *= 0.55;
}
// Helper: compute draw X for text alignment
// biome-ignore lint/correctness/useQwikValidLexicalScope: not Qwik -- this is an Alpine.js component, and the rule is about Qwik's $() serialization boundary, which does not exist here.
const _textX = (text, finfo) => {
const tw = text.length * finfo.charW * s;
if (el.textAlign === 'center') return ax * s - tw / 2;
if (el.textAlign === 'right') return ax * s - tw;
return ax * s;
};
try {
switch (el.type) {
case 'section': {
// A design-time label: it marks a region for the author and is not
// emitted into the generated plugin. There was no case here at all,
// so adding "Section Label" from the palette drew nothing and left
// the user with an apparently broken control.
const sfinfo = FONT_MAP[el.font] || FONT_MAP.four_by_six;
ctx.font = `${sfinfo.sizePx * s}px ${sfinfo.family}`;
ctx.fillStyle = `rgba(${el.r ?? 120},${el.g ?? 120},${el.b ?? 120},0.85)`;
ctx.textBaseline = 'top';
ctx.fillText(el.label || 'Section', ax * s, ay * s);
break;
}
case 'text':
case 'dynamic_text':
case 'clock': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const key = el.binding?.key || '?';
const pv = opts.previewValues?.[key];
// Substitute {variable} tokens in text using previewValues
// biome-ignore lint/correctness/useQwikValidLexicalScope: not Qwik -- this is an Alpine.js component, and the rule is about Qwik's $() serialization boundary, which does not exist here.
const _subVars = str => (str || '').replace(/\{(\w+)\}/g, (_, k) => {
const v = opts.previewValues?.[k];
return v !== undefined && v !== '' ? String(v) : `{${k}}`;
});
const displayText =
el.type === 'text' ? _subVars(el.text || '')
: el.type === 'clock' ? (el.format || '%H:%M')
: (pv !== undefined && pv !== '' ? String(pv) : `{${key}}`);
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillText(displayText, _textX(displayText, finfo), (ay + finfo.sizePx) * s);
// Second line (text and clock)
if (el.type === 'text' && el.text2) {
const t2 = _subVars(el.text2);
const y2 = ay + finfo.sizePx + (el.lineSpacing ?? 2);
ctx.fillText(t2, _textX(t2, finfo), (y2 + finfo.sizePx) * s);
}
if (el.type === 'clock' && el.format2) {
const y2 = ay + finfo.sizePx + (el.lineSpacing ?? 2);
ctx.fillText(el.format2, _textX(el.format2, finfo), (y2 + finfo.sizePx) * s);
}
break;
}
case 'countdown': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const t = el.previewText || '--d --h';
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillText(t, _textX(t, finfo), (ay + finfo.sizePx) * s);
break;
}
case 'rectangle': {
const rx = ax * s, ry = ay * s;
const rw = el.width * s, rh = el.height * s;
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fillRect(rx, ry, rw, rh);
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
// 1 LED pixel, not 1 canvas pixel: at SCALE>1 an unscaled stroke
// renders thinner than the geometry it outlines.
ctx.lineWidth = s;
ctx.strokeRect(rx, ry, rw, rh);
}
break;
}
case 'ellipse': {
const cx = (ax + el.width / 2) * s;
const cy = (ay + el.height / 2) * s;
const lwPx = s; // 1 LED pixel
// Inset by half the stroke, which straddles the path, so the outline
// stays within the element's bounds.
const rx = Math.max(0, (el.width / 2) * s - (el.hasOutline ? lwPx / 2 : 0));
const ry = Math.max(0, (el.height / 2) * s - (el.hasOutline ? lwPx / 2 : 0));
ctx.beginPath();
ctx.ellipse(cx, cy, rx, ry, 0, 0, Math.PI * 2);
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fill();
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
ctx.lineWidth = lwPx;
ctx.stroke();
}
break;
}
case 'arc': {
const cx = (ax + el.width / 2) * s;
const cy = (ay + el.height / 2) * s;
const lwPx = Math.max(1, el.lineWidth || 2) * s;
const rx = Math.max(0, (el.width / 2) * s - lwPx / 2);
const ry = Math.max(0, (el.height / 2) * s - lwPx / 2);
// PIL: 0°=right, clockwise. Canvas: same with anticlockwise=false
const startRad = (el.startAngle ?? 0) * Math.PI / 180;
const endRad = (el.endAngle ?? 270) * Math.PI / 180;
ctx.beginPath();
ctx.ellipse(cx, cy, rx, ry, 0, startRad, endRad, false);
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = lwPx;
ctx.stroke();
break;
}
case 'pixel': {
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillRect(ax * s, ay * s, s, s);
break;
}
case 'rounded_rectangle': {
const rx = ax * s, ry = ay * s;
const rw = el.width * s, rh = el.height * s;
const rad = Math.min((el.borderRadius ?? 3) * s, rw / 2, rh / 2);
ctx.beginPath();
ctx.roundRect(rx, ry, rw, rh, rad);
if (el.hasFill) {
ctx.fillStyle = `rgb(${el.fillR},${el.fillG},${el.fillB})`;
ctx.fill();
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR},${el.outG},${el.outB})`;
ctx.lineWidth = s;
ctx.stroke();
}
break;
}
case 'line': {
// ax/ay resolve from el.x0 for a line, so this is the anchor offset.
// Without it a line stayed put while every other type moved.
const dx = ax - el.x0, dy = ay - el.y0;
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = Math.max(1, el.lineWidth || 1) * s;
ctx.beginPath();
ctx.moveTo((el.x0 + dx) * s, (el.y0 + dy) * s);
ctx.lineTo((el.x1 + dx) * s, (el.y1 + dy) * s);
ctx.stroke();
break;
}
case 'divider': {
const isH = (el.orientation || 'horizontal') === 'horizontal';
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = s;
ctx.beginPath();
if (isH) {
ctx.moveTo(0, (ay + 0.5) * s);
ctx.lineTo(_canvas.width, (ay + 0.5) * s);
} else {
ctx.moveTo((ax + 0.5) * s, 0);
ctx.lineTo((ax + 0.5) * s, _canvas.height);
}
ctx.stroke();
break;
}
case 'pips': {
const pipCount = Math.max(1, el.count ?? 5);
const pvPips = opts.previewValues?.[el.binding?.key];
const filledN = pvPips !== undefined
? Math.max(0, Math.min(pipCount, Math.round(parseFloat(pvPips) || 0)))
: Math.max(0, Math.min(pipCount, el.filled ?? 3));
const ps = Math.max(1, el.pipSize ?? 4);
const pg = Math.max(0, el.pipSpacing ?? 2);
for (let i = 0; i < pipCount; i++) {
const isFilled = i < filledN;
if (!isFilled && !el.showEmpty) continue;
ctx.fillStyle = isFilled
? `rgb(${el.r},${el.g},${el.b})`
: `rgb(${el.emptyR ?? 50},${el.emptyG ?? 50},${el.emptyB ?? 50})`;
ctx.fillRect((ax + i * (ps + pg)) * s, ay * s, ps * s, ps * s);
}
break;
}
case 'sparkline': {
const slW = el.width ?? 40, slH = el.height ?? 12;
const count = Math.max(1, el.barCount ?? 8);
const spacing = el.barSpacing ?? 1;
const barW = Math.max(1, Math.floor((slW - spacing * (count - 1)) / count));
const rawVals = (el.previewData || '').split(',')
.map(v => parseFloat(v.trim())).filter(n => !isNaN(n));
while (rawVals.length < count) rawVals.push(0);
const maxV = Math.max(...rawVals.slice(0, count), 0.001);
const rx = ax * s, ry = ay * s;
if (el.hasBg) {
ctx.fillStyle = `rgb(${el.bgR ?? 30},${el.bgG ?? 30},${el.bgB ?? 30})`;
ctx.fillRect(rx, ry, slW * s, slH * s);
}
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
for (let i = 0; i < count; i++) {
const norm = Math.max(0, Math.min(1, rawVals[i] / maxV));
const barH = Math.max(1, Math.round(slH * norm));
const bx = rx + (barW + spacing) * i * s;
const by = ry + (slH - barH) * s;
ctx.fillRect(bx, by, barW * s, barH * s);
}
break;
}
case 'gauge': {
const gw = (el.width ?? 32), gh = (el.height ?? 32);
const cx = (ax + gw / 2) * s, cy = (ay + gh / 2) * s;
const rx = (gw / 2) * s, ry = (gh / 2) * s;
const lw = Math.max(1, (el.lineWidth ?? 3));
// rx/ry are canvas pixels ((gw/2)*s) but lw is LED pixels, so
// insetting by lw/2 under-corrected by the scale factor while the
// stroke was drawn at lw*s -- the arc spilled outside the element's
// reported bounding box at any SCALE > 1, and the preview stopped
// matching the generated PIL output.
const lwPx = lw * s;
const startDeg = el.startAngle ?? 135;
const endDeg = el.endAngle ?? 45;
// Arc sweep: from startDeg clockwise to endDeg (PIL convention)
const totalSweep = ((endDeg - startDeg) + 360) % 360 || 360;
const pvGauge = opts.previewValues?.[el.binding?.key];
const pct = pvGauge !== undefined
? Math.max(0, Math.min(100, parseFloat(pvGauge) || 0)) / 100
: Math.max(0, Math.min(100, el.previewPct ?? 65)) / 100;
const fillSweep = totalSweep * pct;
// A small imported gauge with a wide lineWidth would otherwise send
// a negative radius into ctx.ellipse(), which throws IndexSizeError
// and aborts the whole render() call.
const arx = Math.max(0, rx - lwPx / 2), ary = Math.max(0, ry - lwPx / 2);
// biome-ignore lint/correctness/useQwikValidLexicalScope: not Qwik -- this is an Alpine.js component, and the rule is about Qwik's $() serialization boundary, which does not exist here.
const toRad = deg => (deg - 90) * Math.PI / 180; // canvas 0=top, PIL 0=right → offset -90
// Track arc
if (el.hasTrack !== false) {
ctx.beginPath();
ctx.ellipse(cx, cy, arx, ary, 0, toRad(startDeg), toRad(startDeg + totalSweep), false);
ctx.strokeStyle = `rgb(${el.trackR ?? 40},${el.trackG ?? 40},${el.trackB ?? 40})`;
ctx.lineWidth = lwPx;
ctx.stroke();
}
// Fill arc
if (pct > 0) {
ctx.beginPath();
ctx.ellipse(cx, cy, arx, ary, 0, toRad(startDeg), toRad(startDeg + fillSweep), false);
ctx.strokeStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.lineWidth = lwPx;
ctx.stroke();
}
// Centre label
if (el.showLabel) {
const gfinfo = FONT_MAP[el.font || 'four_by_six'] || FONT_MAP.four_by_six;
const labelText = Math.round(pct * 100) + '%';
ctx.font = `${gfinfo.sizePx * s}px ${gfinfo.family}`;
ctx.fillStyle = `rgb(${el.labelR ?? 200},${el.labelG ?? 200},${el.labelB ?? 200})`;
const ltw = ctx.measureText(labelText).width;
ctx.fillText(labelText, cx - ltw / 2, cy + (gfinfo.sizePx * s) / 2);
}
break;
}
case 'marquee': {
const finfo = FONT_MAP[el.font] || FONT_MAP.press_start;
const text = el.text || 'Scrolling text';
const tw = text.length * finfo.charW * s;
const gap = (el.gap ?? 16) * s;
const totalW = tw + gap;
const tick = opts.animTick ?? 0;
const speed = (el.scrollSpeed ?? 1) * 2;
const scrolled = (tick * speed) % totalW;
// left: text enters from right; right: text enters from left
const startX = el.direction === 'right'
? scrolled - tw
: matrixW * s - scrolled;
ctx.font = `${finfo.sizePx * s}px ${finfo.family}`;
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
// Clip to canvas width so text doesn't bleed outside
ctx.save();
ctx.beginPath();
ctx.rect(0, ay * s - 1, matrixW * s, (finfo.sizePx + 2) * s);
ctx.clip();
for (let i = -1; i <= 2; i++) {
ctx.fillText(text, startX + i * totalW, (ay + finfo.sizePx) * s);
}
ctx.restore();
break;
}
case 'progress_bar': {
const bw = el.barWidth ?? 60, bh = el.barHeight ?? 6;
const pvPb = opts.previewValues?.[el.binding?.key];
const pct = pvPb !== undefined
? Math.max(0, Math.min(100, parseFloat(pvPb) || 0)) / 100
: Math.max(0, Math.min(100, el.previewPct ?? 65)) / 100;
const rx = ax * s, ry = ay * s;
if (el.hasBg) {
ctx.fillStyle = `rgb(${el.bgR ?? 30},${el.bgG ?? 30},${el.bgB ?? 30})`;
ctx.fillRect(rx, ry, bw * s, bh * s);
}
const fillW = Math.max(0, Math.round(bw * pct));
if (fillW > 0) {
ctx.fillStyle = `rgb(${el.r},${el.g},${el.b})`;
ctx.fillRect(rx, ry, fillW * s, bh * s);
}
if (el.hasOutline) {
ctx.strokeStyle = `rgb(${el.outR ?? 100},${el.outG ?? 100},${el.outB ?? 100})`;
ctx.lineWidth = s;
ctx.strokeRect(rx, ry, bw * s, bh * s);
}
break;
}
}
if (belowBreakpoint) {
ctx.globalAlpha = 0.6;
const bb = getBoundingBox(el, matrixW, matrixH);
ctx.font = `${Math.max(8, s * 2)}px monospace`;
ctx.fillStyle = '#facc15';
ctx.fillText(`≥${el.minWidth}px`, bb.x * s, (bb.y + 4) * s);
}
} finally {
ctx.restore();
}
}
// ── Selection indicator ──────────────────────────────────────────────
function _drawSelection(ctx, el, SCALE, matrixW, matrixH) {
const bb = getBoundingBox(el, matrixW, matrixH);
const PAD = 2, s = SCALE;
const rx = bb.x * s - PAD, ry = bb.y * s - PAD;
const rw = bb.w * s + PAD * 2, rh = bb.h * s + PAD * 2;
ctx.save();
ctx.strokeStyle = '#3b82f6';
ctx.lineWidth = 1;
ctx.setLineDash([3, 2]);
ctx.strokeRect(rx, ry, rw, rh);
ctx.setLineDash([]);
if (el.xAnchor || el.yAnchor) {
ctx.font = `${Math.max(7, s)}px sans-serif`;
ctx.fillStyle = '#a78bfa';
const anchorText = [
el.xAnchor ? `x:${el.xAnchor[0]}` : '',
el.yAnchor ? `y:${el.yAnchor[0]}` : '',
].filter(Boolean).join(' ');
if (anchorText) ctx.fillText(anchorText, rx + 1, ry - 2);
}
// Resize handles: on rect, rounded rect, ellipse
if (RESIZABLE_TYPES.includes(el.type)) {
const handles = _getRectHandles(el, matrixW, matrixH);
const HS = 5;
ctx.fillStyle = 'white';
ctx.strokeStyle = '#2563eb';
ctx.lineWidth = 1;
for (const pt of Object.values(handles)) {
const hx = pt.x * s - HS / 2;
const hy = pt.y * s - HS / 2;
ctx.fillRect(hx, hy, HS, HS);
ctx.strokeRect(hx, hy, HS, HS);
}
} else {
// Corner dots for non-rectangle elements
ctx.fillStyle = '#3b82f6';
const HS = 4;
for (const [hx, hy] of [
[rx - HS / 2, ry - HS / 2], [rx + rw - HS / 2, ry - HS / 2],
[rx - HS / 2, ry + rh - HS / 2], [rx + rw - HS / 2, ry + rh - HS / 2],
]) ctx.fillRect(hx, hy, HS, HS);
}
ctx.restore();
}
// ── Dimension tooltip while dragging ─────────────────────────────────
function drawDragTooltip(ctx, el, SCALE, matrixW, matrixH) {
const bb = getBoundingBox(el, matrixW, matrixH);
const label = el.type === 'rectangle'
? `${el.width}×${el.height}`
: `${bb.x},${bb.y}`;
const s = SCALE;
ctx.save();
ctx.font = `${Math.max(9, s * 1.5)}px monospace`;
const tw = ctx.measureText(label).width;
const tx = bb.x * s, ty = (bb.y - 2) * s;
ctx.fillStyle = 'rgba(0,0,0,0.7)';
ctx.fillRect(tx - 2, ty - 10, tw + 4, 12);
ctx.fillStyle = 'white';
ctx.fillText(label, tx, ty);
ctx.restore();
}
// ── Public API ───────────────────────────────────────────────────────
function init(canvasEl) {
_canvas = canvasEl;
_ctx = canvasEl.getContext('2d');
}
function setGrid(show) { _showGrid = show; }
function updateCanvasSize(matrixW, matrixH, SCALE) {
if (!_canvas) return;
_canvas.width = matrixW * SCALE;
_canvas.height = matrixH * SCALE;
}
function render(elements, selectedId, matrixW, matrixH, SCALE, opts = {}) {
if (!_ctx) return;
const cW = matrixW * SCALE, cH = matrixH * SCALE;
const bg = opts.bgColor;
_ctx.fillStyle = bg ? `rgb(${bg.r},${bg.g},${bg.b})` : '#000';
_ctx.fillRect(0, 0, cW, cH);
if (_showGrid) {
_ctx.strokeStyle = 'rgba(255,255,255,0.07)';
_ctx.lineWidth = 0.5;
for (let x = SCALE; x < cW; x += SCALE) {
_ctx.beginPath(); _ctx.moveTo(x, 0); _ctx.lineTo(x, cH); _ctx.stroke();
}
for (let y = SCALE; y < cH; y += SCALE) {
_ctx.beginPath(); _ctx.moveTo(0, y); _ctx.lineTo(cW, y); _ctx.stroke();
}
}
for (const el of elements) _drawElement(_ctx, el, SCALE, matrixW, matrixH, opts);
if (opts.showRuler) {
_ctx.save();
_ctx.fillStyle = 'rgba(255,255,255,0.08)';
_ctx.fillRect(0, 0, cW, SCALE); // top strip
_ctx.fillRect(0, 0, SCALE, cH); // left strip
_ctx.strokeStyle = 'rgba(255,255,255,0.5)';
_ctx.fillStyle = 'rgba(255,255,255,0.6)';
_ctx.font = `${Math.max(5, SCALE - 1)}px monospace`;
const step = SCALE >= 4 ? 8 : 16;
for (let px = 0; px <= matrixW; px += step) {
const cx = px * SCALE;
const major = px % 32 === 0;
_ctx.lineWidth = 0.5;
_ctx.beginPath(); _ctx.moveTo(cx, 0); _ctx.lineTo(cx, major ? SCALE : SCALE * 0.5); _ctx.stroke();
if (major && px > 0 && px < matrixW - 4) _ctx.fillText(String(px), cx + 1, SCALE - 1);
}
for (let py = 0; py <= matrixH; py += step) {
const cy = py * SCALE;
const major = py % 32 === 0;
_ctx.beginPath(); _ctx.moveTo(0, cy); _ctx.lineTo(major ? SCALE : SCALE * 0.5, cy); _ctx.stroke();
if (major && py > 0 && py < matrixH - 4) _ctx.fillText(String(py), 1, cy + SCALE - 1);
}
_ctx.restore();
}
if (opts.showGuides) {
_ctx.save();
_ctx.strokeStyle = 'rgba(255,60,60,0.45)';
_ctx.lineWidth = 1;
_ctx.setLineDash([4, 3]);
const mx = Math.floor(cW / 2) + 0.5;
const my = Math.floor(cH / 2) + 0.5;
_ctx.beginPath(); _ctx.moveTo(mx, 0); _ctx.lineTo(mx, cH); _ctx.stroke();
_ctx.beginPath(); _ctx.moveTo(0, my); _ctx.lineTo(cW, my); _ctx.stroke();
_ctx.setLineDash([]);
_ctx.restore();
}
const sel = selectedId != null ? elements.find(e => e.id === selectedId) : null;
if (sel) {
_drawSelection(_ctx, sel, SCALE, matrixW, matrixH);
if (opts.showTooltip) drawDragTooltip(_ctx, sel, SCALE, matrixW, matrixH);
}
}
return {
init, render, setGrid, updateCanvasSize,
hitTest, getBoundingBox, computeActualPos, resolveAnchor,
getResizeHandle, getCursorForHandle,
ELEMENT_DEFAULTS, FONT_MAP, DISPLAY_PRESETS, RESIZABLE_TYPES,
};
})();
File diff suppressed because it is too large Load Diff
@@ -1,302 +0,0 @@
"""
{{ plugin_name }} — LEDMatrix Plugin
Generated by LEDMatrix Plugin Composer on {{ generated_date }}
Extension points:
update() → add HTTP/MQTT data-fetching logic here
_get_display_values() → map fetched data to display strings
display() → add new elements or adapt layout per display size
"""
from src.plugin_system.base_plugin import BasePlugin
{% if has_clock or has_countdown %}
from datetime import datetime
{% endif %}
{% if has_blink %}
import time
{% endif %}
{% if has_text_template %}
from collections import defaultdict
{% endif %}
class {{ class_name }}(BasePlugin):
def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):
super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
{% for var in config_vars %}
self.{{ var.key }} = config.get({{ var.key | tojson }}, {{ var.default | tojson }})
{% endfor %}
# Live data cache — populated by update(); always {} in static layouts
self._data = {}
def update(self):
"""Fetch and refresh display data.
For dynamic plugins: fetch from APIs/MQTT here and store in self._data.
_get_display_values() will read self._data to produce display strings.
"""
# --- Data sources (add fetch logic here for dynamic plugins) ---
pass
def _get_display_values(self):
"""Map config variables and live data to display-ready strings.
This is the single extension point for v2 data sources:
add self._data lookups here once update() populates them.
"""
return {
{% for var in config_vars %}
{{ var.key | tojson }}: str(self.{{ var.key }}),
{% endfor %}
}
def display(self, force_clear=False):
try:
{% if has_text_template %}
values = defaultdict(str, self._get_display_values())
{% else %}
values = self._get_display_values()
{% endif %}
if force_clear:
self.display_manager.clear()
width = self.display_manager.width
height = self.display_manager.height
{% if bg_color %}
self.display_manager.draw.rectangle([0, 0, width, height], fill={{ bg_color }})
{% endif %}
# ── Elements (rendered bottom to top) ──────────────────────────
{% for el in elements %}
{% set p = " " if el.min_width > 0 else " " %}
{% set pi = (p + " ") if el.blink else p %}
{% if el.min_width > 0 %}
if width >= {{ el.min_width }}: # breakpoint: {{ el.min_width }}px+ displays only
{% endif %}
{% if el.blink %}
{{ p }}if int(time.time() * 2) % 2:
{% endif %}
{% if el.type == 'text' %}
{{ pi }}self.display_manager.draw_text(
{% if el.text_is_template %}
{{ pi }} {{ el.text | tojson }}.format_map(values),
{% else %}
{{ pi }} {{ el.text | tojson }},
{% endif %}
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% if el.text2 %}
{{ pi }}self.display_manager.draw_text(
{% if el.text_is_template %}
{{ pi }} {{ el.text2 | tojson }}.format_map(values),
{% else %}
{{ pi }} {{ el.text2 | tojson }},
{% endif %}
{{ pi }} x={{ el.x2_expr }}, y={{ el.y2_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'dynamic_text' %}
{% if el.binding_source == 'config' %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} values.get({{ el.binding_key | tojson }}, ''),
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% else %}
{{ pi }}pass # non-config dynamic_text binding draws nothing
{% endif %}
{% elif el.type == 'clock' %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} datetime.now().strftime({{ el.format | tojson }}),
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% if el.format2 %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} datetime.now().strftime({{ el.format2 | tojson }}),
{{ pi }} x={{ el.x2_expr }}, y={{ el.y2_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'countdown' %}
{{ pi }}_cd_target = float(values.get({{ el.binding_key | tojson }}, 0) or 0)
{{ pi }}_cd_secs = max(0.0, _cd_target - datetime.now().timestamp())
{% if el.countdown_format == 'dhms' %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h, _cd_rem = divmod(_cd_rem, 3600)
{{ pi }}_cd_m, _cd_s = divmod(_cd_rem, 60)
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h:02d}:{_cd_m:02d}:{_cd_s:02d}'
{% elif el.countdown_format == 'hms' %}
{{ pi }}_cd_h, _cd_rem = divmod(int(_cd_secs), 3600)
{{ pi }}_cd_m, _cd_s = divmod(_cd_rem, 60)
{{ pi }}_cd_str = f'{_cd_h}h {_cd_m:02d}:{_cd_s:02d}'
{% elif el.countdown_format == 'dhm' %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h, _cd_m = divmod(_cd_rem // 60, 60)
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h:02d}h {_cd_m:02d}m'
{% else %}
{{ pi }}_cd_d, _cd_rem = divmod(int(_cd_secs), 86400)
{{ pi }}_cd_h = _cd_rem // 3600
{{ pi }}_cd_str = f'{_cd_d}d {_cd_h}h'
{% endif %}
{{ pi }}self.display_manager.draw_text(
{{ pi }} _cd_str,
{{ pi }} x={{ el.x_expr }}, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% elif el.type == 'rectangle' %}
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type == 'arc' %}
{{ pi }}self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.end_angle }},
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% elif el.type == 'ellipse' %}
{{ pi }}self.display_manager.draw.ellipse(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type == 'pixel' %}
{{ pi }}self.display_manager.draw.point(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}],
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }})
{% elif el.type == 'rounded_rectangle' %}
{{ pi }}self.display_manager.draw.rounded_rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} radius={{ el.border_radius }},
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{% elif el.type in ('line', 'divider') %}
{{ pi }}self.display_manager.draw.line(
{{ pi }} [{{ el.x0_expr }}, {{ el.y0_expr }}, {{ el.x1_expr }}, {{ el.y1_expr }}],
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% elif el.type == 'pips' %}
{{ pi }}_pip_filled = max(0, min({{ el.pip_count }}, int(float(values.get({{ el.binding_key | tojson }}, 0) or 0))))
{{ pi }}for _pip_i in range({{ el.pip_count }}):
{{ pi }} _pip_x = ({{ el.x_expr }}) + _pip_i * ({{ el.pip_size }} + {{ el.pip_spacing }})
{{ pi }} _pip_color = {{ el.fill_tuple }} if _pip_i < _pip_filled else {{ el.empty_tuple }}
{% if not el.show_empty %}
{{ pi }} if _pip_i >= _pip_filled:
{{ pi }} continue
{% endif %}
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_pip_x, {{ el.y_expr }}, _pip_x + {{ el.pip_size }} - 1, ({{ el.y_expr }}) + {{ el.pip_size }} - 1],
{{ pi }} fill=_pip_color,
{{ pi }} )
{% elif el.type == 'sparkline' %}
{{ pi }}_sl_raw = str(values.get({{ el.binding_key | tojson }}, '') or '')
{{ pi }}_sl_vals = [float(v.strip()) for v in _sl_raw.split(',') if v.strip()][:{{ el.bar_count }}]
{{ pi }}_sl_vals += [0.0] * max(0, {{ el.bar_count }} - len(_sl_vals))
{{ pi }}_sl_max = max(_sl_vals) if any(_sl_vals) else 1.0
{{ pi }}_sl_bw = max(1, ({{ el.bar_width_px }} - {{ el.bar_spacing }} * ({{ el.bar_count }} - 1)) // {{ el.bar_count }})
{% if el.bg_tuple != 'None' %}
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, ({{ el.x_expr }}) + {{ el.bar_width_px }}, ({{ el.y_expr }}) + {{ el.bar_height_px }}],
{{ pi }} fill={{ el.bg_tuple }},
{{ pi }})
{% endif %}
{{ pi }}for _sl_i, _sl_v in enumerate(_sl_vals):
{{ pi }} _sl_norm = max(0.0, min(1.0, _sl_v / (_sl_max or 1)))
{{ pi }} _sl_bh = max(1, round({{ el.bar_height_px }} * _sl_norm))
{{ pi }} _sl_bx = ({{ el.x_expr }}) + (_sl_bw + {{ el.bar_spacing }}) * _sl_i
{{ pi }} _sl_by = ({{ el.y_expr }}) + {{ el.bar_height_px }} - _sl_bh
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_sl_bx, _sl_by, _sl_bx + _sl_bw - 1, _sl_by + _sl_bh - 1],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} )
{% elif el.type == 'gauge' %}
{{ pi }}_gv = max(0.0, min(100.0, float(values.get({{ el.binding_key | tojson }}, 0) or 0)))
{{ pi }}_g_total = (({{ el.end_angle }} - {{ el.start_angle }}) % 360) or 360
{{ pi }}_g_sweep = _g_total * _gv / 100.0
{% if el.track_tuple != 'None' %}
{{ pi }}self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.start_angle }} + _g_total,
{{ pi }} fill={{ el.track_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }})
{% endif %}
{{ pi }}if _g_sweep > 0:
{{ pi }} self.display_manager.draw.arc(
{{ pi }} [{{ el.x_expr }}, {{ el.y_expr }}, {{ el.x2_expr }}, {{ el.y2_expr }}],
{{ pi }} start={{ el.start_angle }}, end={{ el.start_angle }} + _g_sweep,
{{ pi }} fill={{ el.rgb_tuple }},
{{ pi }} width={{ el.line_width }},
{{ pi }} )
{% if el.show_label %}
{{ pi }}_g_cx = ({{ el.x_expr }}) + ({{ el.x2_expr }} - ({{ el.x_expr }})) // 2
{{ pi }}_g_cy = ({{ el.y_expr }}) + ({{ el.y2_expr }} - ({{ el.y_expr }})) // 2
{{ pi }}self.display_manager.draw_text(
{{ pi }} f'{int(_gv)}%',
{{ pi }} x=_g_cx, y=_g_cy,
{{ pi }} color={{ el.label_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% endif %}
{% elif el.type == 'marquee' %}
{{ pi }}_{{ el.data_key }}_text = {{ el.text | tojson }}
{{ pi }}_{{ el.data_key }}_tw = len(_{{ el.data_key }}_text) * {{ el.char_w }}
{{ pi }}_{{ el.data_key }}_x = int(self._data.get({{ el.data_key | tojson }}, width))
{% if el.direction == 'right' %}
{{ pi }}_{{ el.data_key }}_x += {{ el.scroll_speed }}
{{ pi }}if _{{ el.data_key }}_x > width:
{{ pi }} _{{ el.data_key }}_x = -(_{{ el.data_key }}_tw + {{ el.gap }})
{% else %}
{{ pi }}_{{ el.data_key }}_x -= {{ el.scroll_speed }}
{{ pi }}if _{{ el.data_key }}_x < -(_{{ el.data_key }}_tw + {{ el.gap }}):
{{ pi }} _{{ el.data_key }}_x = width
{% endif %}
{{ pi }}self._data[{{ el.data_key | tojson }}] = _{{ el.data_key }}_x
{{ pi }}self.display_manager.draw_text(
{{ pi }} _{{ el.data_key }}_text,
{{ pi }} x=_{{ el.data_key }}_x, y={{ el.y_expr }},
{{ pi }} color={{ el.rgb_tuple }},
{{ pi }} font=self.display_manager.{{ el.font_attr }},
{{ pi }})
{% elif el.type == 'progress_bar' %}
{{ pi }}_pb_x = {{ el.x_expr }}
{{ pi }}_pb_y = {{ el.y_expr }}
{{ pi }}_pb_pct = max(0.0, min(100.0, float(values.get({{ el.binding_key | tojson }}, 0) or 0))) / 100.0
{{ pi }}_pb_fill_w = int({{ el.bar_width }} * _pb_pct)
{{ pi }}self.display_manager.draw.rectangle(
{{ pi }} [_pb_x, _pb_y, _pb_x + {{ el.bar_width }}, _pb_y + {{ el.bar_height }}],
{{ pi }} fill={{ el.bg_tuple }},
{{ pi }} outline={{ el.outline_tuple }},
{{ pi }})
{{ pi }}if _pb_fill_w > 0:
{{ pi }} self.display_manager.draw.rectangle(
{{ pi }} [_pb_x, _pb_y, _pb_x + _pb_fill_w, _pb_y + {{ el.bar_height }}],
{{ pi }} fill={{ el.fill_tuple }},
{{ pi }} )
{% else %}
{{ pi }}pass # element type "{{ el.type }}" draws nothing
{% endif %}
{% endfor %}
# ── End elements ───────────────────────────────────────────────
self.display_manager.update_display()
except Exception as e:
self.logger.error('Display error: %s', e, exc_info=True)