rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-08-22 02:48:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cover the next tier of untested modules and endpoints, and fix the 43 bugs that surfaced (#459)

Browse Source 
* test(sync): cover the display sync protocol, and fix what that surfaced

DisplaySyncManager had no tests at all — it appeared in the suite only as
a MagicMock() stand-in, so none of its framing, handshake, or socket
handling was ever exercised. Writing that coverage surfaced three bugs.

Both receive loops caught the generic Exception and immediately retried.
A socket left in a bad state raises on every call, so the thread spun at
100% CPU logging the same line; the reverted-code run of the new
regression test takes 24 seconds where the fixed one takes 0.2. Both now
back off briefly before retrying.

The follower dispatched on `data[:8] == _RAW_MAGIC or len(data) > 512`.
That size threshold is not part of either wire format: a control message
over 512 bytes — a hello_ack carrying a long incompatibility error, for
instance — went to the image decoder and was dropped, and a raw frame
under 512 bytes went to the JSON parser. Both formats are already
self-describing, so dispatch on the magic prefix and treat a JSON parse
failure as the legacy unmarked PNG, with the shared frame bookkeeping
factored into _handle_received_frame().

_oversized_frame_warned was created on first use through
getattr(self, ..., False) rather than in __init__, alone among the
instance attributes.

75 tests: role parsing, the hello compatibility matrix, watchdog
timeouts, both receive loops, the TCP image server's length and
dimension caps and decompression-bomb guard, status shape per role, and
one end-to-end loopback handshake so the wire format is exercised for
real and not only against mocks.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(logos): cover LogoHelper, and stop bad downloads poisoning the cache

Nothing in test/ referenced logo_helper.py, so its caching, resizing and
download-fallback logic was entirely unexercised. Two bugs surfaced.

_download_logo wrote response.content to disk with no size limit and no
check that the bytes were an image. A logo URL is remote input, so the
response chose how much went into the assets directory; worse, an
undecodable one stayed there, and because load_logo() only reports the
decode failure and returns None, every later call re-read the same
corrupt file. The download path never retried, so a single bad response
made a logo permanently blank rather than falling back to the
placeholder. Cap the response, verify it decodes, and delete it if not,
which lets the existing fallback in load_logo_with_download do its job.

get_cache_stats() divided by self.cache_size with no guard, so a helper
built with cache_size=0 raised ZeroDivisionError from what is only a
stats call.

37 tests: size-qualified cache keys, LRU eviction and refresh, the four
load_logo_with_download paths, download permissions and timeout,
placeholder generation, and the abbreviation normalizer — including a
test pinning its deliberate divergence from
LogoDownloader.normalize_abbreviation, since logo filenames on existing
installs depend on both behaviors staying put.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(web): cover the error and response builders, and stop dropping empty values

errors.py and error_handler.py's response builders had no direct tests,
though every API response passes through them. Two bugs surfaced.

WebInterfaceError set suggested_fixes with `or`, so a caller passing []
to mean "I have no suggestions for this one" got the default list
instead. Only None should fall back.

create_success_response gated `data` on `is not None` but `message` and
`metadata` on truthiness, so an explicitly-passed "" or {} vanished from
the response while 0 and False survived — the response shape depended on
the value. api_helpers.success_response() then re-gated metadata the same
way, which is the path every api_v3 endpoint actually calls, so fixing
only the inner function would have changed nothing observable. Both now
use `is not None`.

That wrapper also merged request timing into the caller's own metadata
dict in place. A caller reusing a dict across requests would accumulate
previous responses' timings; it now copies before adding.

79 tests: category inference for every error code, mapped vs fallback
suggestions, the JSON shape including which keys are omitted when empty,
exception-to-code inference, and the success/error builders end to end.
Two behaviours are pinned as deliberate rather than fixed: an empty
context stays out of the response body, and from_exception's `message`
is the fixed per-code string, never the raw exception text.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(web): cover the input validators, and close three holes in them

validators.py had tests for dedup_unique_arrays only; the other eight
functions were untested. Three bugs surfaced.

validate_image_url checked for '..' only inside its relative-path
branch, so http://host/../secret passed validation while /../secret was
rejected — the traversal check now runs before the branch split, which
is where a safety check on the whole URL belongs.

validate_file_upload lowercased the uploaded filename's extension but
compared it against the caller's list verbatim, so allowed_extensions of
['.TTF'] rejected every valid .ttf file. Both sides are lowercased now.
The one in-tree caller passes lowercase already, so this only widens what
future callers can hand it.

validate_numeric_range accepted True and False, because bool subclasses
int; a boolean then compared as 1 or 0 against the range and validated
cleanly. Excluded explicitly, matching how base_plugin.py already handles
the same trap for display_duration.

84 tests. Two behaviours are pinned rather than changed:
sanitize_plugin_config deliberately does not HTML-escape strings, since
escaping at this layer would store the escaped form in config.json — the
docstring said "prevent injection", which read as a promise it does not
keep, and now says what it actually does. validate_font_awesome_class's
second 'fa-' check is unreachable behind its own regex; harmless, so
characterized rather than removed.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover wifi and registry endpoints, and fix bodyless POSTs

The /wifi/* routes drive the host's real networking and the registry
routes reach GitHub, and neither had endpoint-level tests. Covering them
surfaced a bug affecting six endpoints.

Six handlers read their body as `request.get_json() or {}`. The `or {}`
says every field is optional and a missing body should fall back to
defaults — but get_json() without silent=True raises UnsupportedMediaType
when there is no JSON Content-Type, and it raises before `or {}` is ever
evaluated. Each handler's catch-all then reported that as a 500. So
POSTing with no body — what curl sends by default, and what a fetch()
without options sends — failed on /plugins/store/refresh,
/display/on-demand/start, /plugins/config/reset,
/plugins/of-the-day/json/delete, /plugins/{id}/limits and
/plugins/authenticate/spotify. The shipped UI always sends a JSON object,
which is why this stayed hidden.

All six now use silent=True. test_api_v3_optional_body.py covers the
affected endpoints and adds a source check, since the combination of
`or <default>` with a non-silent read is self-contradictory wherever it
appears and is easier to catch by inspection than by exercising each
endpoint by hand.

Also adds test/_api_v3_test_helpers.py: the blueprint holds its managers
on a module-level singleton rather than in Flask app state, so a test
that mocks them leaks into every later test unless the originals are
restored. The existing _make_client() does this for unittest classes;
this is the pytest-fixture equivalent, for the five suites still to come.

69 endpoint tests: connect/disconnect/AP/radio including the string-aware
boolean coercion these endpoints deliberately use, the radio's
lockout-refusal path, registry refresh and fetch-from-URL, and a guard
that WiFiManager is never constructed for real.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover the music auth endpoints, and always clean up the wrapper

The Spotify step-2 handler writes a Python wrapper script to a temp file
with the user's redirect URL embedded in its source, then executes it.
That is the most dangerous shape in the blueprint and had no tests.

The wrapper was deleted in the success/failure branch and again in the
TimeoutExpired handler. Any other failure from subprocess.run — no
interpreter, a fork failure, an interrupted call — reached neither, and
left a world-readable temp file containing the user's redirect URL on
disk. Cleanup moves to a finally block, which is what "delete this
whatever happens" should have been from the start.

The injection tests are the point of this file. Eight adversarial
redirect URLs (embedded quotes, backslashes, newlines, triple quotes, a
full `"; import os; os.system("id"); "`) are each pushed through the
endpoint and the generated wrapper is parsed with ast: it must still be
valid Python, the URL must still be a single string literal bound to
redirect_url, and no os.system call may appear anywhere in the tree.
json.dumps holds up, but nothing was checking that it does.

40 tests. Also pins that the two endpoints are not symmetrical despite
the matching names — only Spotify has a two-step flow and a wrapper; YTM
runs its script directly — so a later change does not "restore" a parity
that was never there.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover the credentials upload, and stop it hoarding secrets

The endpoint that receives the user's Google OAuth credentials file had
no tests. Two bugs surfaced.

The OAuth-shape check ran inside `except Exception: pass`. A JSON
document that parses but is not an object — a bare 42, true, null, a
list — makes `'installed' not in creds_data` raise TypeError, which the
bare except swallowed, and the file was then written out as
credentials.json regardless. The check now decides the outcome instead
of being advisory, so anything not credentials-shaped is refused up
front rather than failing later inside the calendar plugin.

Every overwrite copies the old file to credentials.json.backup.<ts> and
nothing removed them, so a user who re-uploaded ten times had ten
complete sets of OAuth client credentials sitting in the plugin
directory, indefinitely. Keep the newest five. Pruning is housekeeping,
so a backup that cannot be removed logs and leaves the upload alone.

27 tests: size and extension limits, malformed JSON, the shape check,
0600 permissions on the written file, backup-on-overwrite, and pruning
including the repeated-upload case that stays bounded.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover the install endpoints, and make 14 dead guards reachable

/plugins/install and /plugins/install-from-url were tested only at the
PluginStoreManager layer, so the route logic — the queue-versus-direct
branch, schema invalidation, discovery, state and history recording — was
unexercised.

Covering them surfaced the wider form of the body-parsing bug fixed for
the `or {}` handlers in the previous commit. Fourteen handlers read
`data = request.get_json()` and immediately guard with `if not data:
return 400, 'No data provided'`. That guard cannot run: get_json()
without silent=True raises UnsupportedMediaType for a request with no
JSON body, so the catch-all answered 500 "an error occurred; see logs
for details" where the handler plainly meant to answer 400 and say
which field was missing. Every one of these endpoints told a caller who
simply forgot the body to go read the server logs.

All fourteen now use silent=True, so the guard each author already wrote
is the one that runs. This covers /config/raw/main and /config/raw/secrets
among them, whose own bodyless case had the same shape.

The two remaining bare reads are left alone: neither declares what a
missing body should do, so there is no stated intent to honour.

31 install tests plus 17 body tests. The install pair is checked against
each other rather than only individually — the same install logic is
written twice, once in the queue callback and once in the fallback, so
the tests assert both produce identical schema, discovery, state and
history effects. They agree today; the one difference is the success
message wording, which is characterized rather than changed.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover the raw config write endpoints

/config/raw/main and /config/raw/secrets write whatever JSON they are
given straight to config.json and config_secrets.json, bypassing the
secret-separation path the rest of the config surface goes through. Given
how carefully that surface keeps secrets out of config.json, the pair
that skips it was worth pinning precisely. Backed by a real
ConfigManager over tmp_path, so the assertions are against files on disk.

20 tests covering both routes: what lands in which file, that a raw
secrets write never touches config.json and vice versa, the GitHub token
reload, the uninitialized-manager and empty-body branches, and the
ConfigError path that carries config_path through to the response.

The bypass itself is pinned as intentional rather than changed — these
back the raw JSON editor, so writing the body verbatim is the feature.
The test says so explicitly, because the failure mode is someone later
routing plugin config through here as a convenience and silently losing
secret separation.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(api): cover backup restore and path containment, and fix restore scope

Restore is the most destructive thing the web interface can do — it
overwrites config, secrets, WiFi settings and fonts, then reinstalls
plugins — and neither it nor the file routes beside it had tests.

A malformed `options` field fell back to {}. Every RestoreOptions flag
defaults to True, so a caller who asked for a narrow restore and
mis-serialized the request got a full one instead, secrets included, and
was told it succeeded. Valid JSON that is not an object was worse:
`"null"` or `"[1,2]"` reached .get() on a non-dict and raised, so the
request died as a generic 500. Both are now refused with a 400 that says
what was wrong, and restore_backup is never reached.

The other file routes take a filename straight out of the URL and turn it
into a path — one to read, one to unlink. _safe_backup_path is the only
thing keeping those inside the export directory, and it was untested. No
bypass was found; the thirteen traversal shapes are pinned so a later
loosening of that pattern has to argue with something. The delete route's
by-name enumeration is covered too, including that a directory sharing a
backup's name is not removed.

84 tests. Two behaviours are pinned as intentional: a failed plugin
reinstall turns the whole restore into an error even though file
restoration succeeded, and omitting `options` entirely still means
restore everything — that is the documented default, and it is only the
mis-serialized case that was wrong.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* ci: raise coverage floor to 52%

Measured 54.45% after the Tier 1 and Tier 2 suites, up from 50%. Keeping
the same two points of headroom the 45 -> 48 ratchet used.

The modules this branch set out to cover: sync_manager 0 -> 97%,
logo_helper 0 -> 98%, errors and error_handler 0 -> 100%, validators
0 -> 97%. api_v3 moved less in percentage terms because it is 4,341
statements, but the endpoints covered are the destructive and
credential-handling ones.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(sync): probe for a free port on loopback, not every interface

CodeQL flagged the ephemeral-port probe in the handshake test for
binding to all interfaces. The probe only needs a free port number, so
loopback is both sufficient and correct — a test should not open a port
to the network to discover one.

The manager under test still binds to all interfaces, which is
deliberate and already marked nosec: a follower has to receive the
leader's UDP broadcast.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* fix: bound the logo download, and stop malformed input reading as a fault

Review findings on the coverage branch.

The download size cap I added checked len(response.content), which has
already buffered the whole body -- it stopped the bytes reaching disk but
not memory, which was the point. A server that omits Content-Length and
never stops sending would still exhaust the process. Stream it instead,
counting as it arrives, into a sibling .part file that is replaced over
the target only once it decodes. A transfer that dies midway now leaves
nothing behind rather than a truncated logo for load_logo() to cache.

The follower's control-message handler caught three exception types, but
two reachable UDP payloads raise others: a bare JSON scalar makes
msg.get() raise AttributeError, and an "sx" carrying a non-numeric x
raises ValueError or TypeError from float(). Those escaped to the outer
handler, skipping the legacy-PNG fallback and -- since this branch added
a backoff there -- charging one malformed packet a 0.1s stall on the
receive path. The legacy-PNG path also decoded without the dimension cap
its TCP counterpart applies, so a crafted 65KB frame could force a large
allocation on the render thread; both paths now share one constant.

Three repo_url handlers called .strip() on client input without checking
it was a string, so {"repo_url": 12345} answered 500. The credentials
upload parsed the same file twice, the second time inside a bare except
that a preceding parse had already made unreachable. And both raw-config
handlers kept a json.JSONDecodeError arm that get_json(silent=True) had
turned into dead code, collapsing "sent something unparseable" into "sent
nothing" -- they now say which.

Two of the new tests were not testing what they claimed. The pruning
round-trip wrote ten backups inside one second, so all ten landed on the
same int(time.time()) filename and overwrote each other; it never reached
the limit it asserted. And the sync clock helper patched attributes on the
stdlib time module, freezing time process-wide for every daemon thread
earlier tests had left running.

Full suite: 3352 passed, coverage 54%.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(sync): probe broadcast by sending, not by listening

The broadcast check added in the previous commit bound INADDR_ANY to
receive its own probe datagram, and the free-port probe did the same to
pick a port. CodeQL flagged both, correctly: a test suite has no reason
to open a socket the whole network can reach.

Sending is enough for what the probe is actually for. An environment
that refuses broadcast raises on sendto, which is the case that occurs
in sandboxes and is the one worth skipping over; confirming delivery
would have required the listening socket. A network that accepts the
send and silently drops it still reaches the assertion, exactly as it
did before either commit. The port probe binds loopback -- it only needs
a number, and the manager's own bind is the one that has to succeed, with
the retry loop already covering a port taken elsewhere.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* fix: keep callback faults out of the frame-decode fallback

Review follow-up on the previous two commits.

Widening the control-message except tuple put the callback dispatch
inside it, so an _on_new_cycle() that raised ValueError, TypeError or
AttributeError sent a perfectly good control packet to the legacy PNG
decoder -- which reported it as an image decode error and buried the
real fault. Split the two: whether the payload parses as JSON decides
frame vs control message, a second guard covers reading the fields of an
attacker-shaped body, and the callback fires outside both. It still
cannot kill the receive thread; the loop's own handler catches it, and
now says what actually went wrong.

The logo download's temp file was a fixed "<name>.part". Two plugins
asking for the same logo at once would interleave writes into it,
publish the mixture, or delete each other's partial. mkstemp gives each
download its own name in the same directory, so os.replace stays atomic.
Its descriptor is adopted by fdopen before the request runs, since a
request that raises before the write would otherwise leak the fd --
quietly, because load_logo_with_download swallows that.

Two test fixes: the oversized-frame test replaced PIL.Image.open
process-wide, the same hazard the clock helper documents, and Ruff B007
on an unused loop variable.

Full suite: 3355 passed, coverage 54%.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

* test(sync): cover the announce loop, and reject non-finite scroll positions

Three review findings from the follower receive path.

Non-finite scroll x reached follower rendering. json.loads accepts the
bare NaN/Infinity literals and float() accepts them as strings, so
"x": NaN arrived as a real float and was stored verbatim. NaN loses
every comparison the scroll code makes, so a follower given one sits on
a position it can never advance past. It now raises through the existing
malformed-control-message guard, which logs and drops the packet and
leaves the last good position in place.

_broadcast_available() only proves the host accepts sendto() for a
broadcast; a network that accepts the send and drops the packet would
let TestRealSocketHandshake run to its five-second deadline and fail on
assertions the code did not break. The deadline now distinguishes the
two: if not one packet crossed in either direction, that is the
environment, and the test skips rather than reporting a protocol
failure.

That skip could hide a real regression in the announcing side, so
TestFollowerAnnounceLoop covers it on mock sockets, where no network is
involved and nothing can skip: hello carries this display's hardware
config and goes to the broadcast address, heartbeats follow, an empty
hardware config falls back to 32x64x1, hello is not resent before its
interval, and a send failure is swallowed rather than killing the loop.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NohXi78cwsAKtN1sCfxjUh

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Chuck
2026-08-21 11:16:52 -04:00
committed by GitHub
co-authored by Claude Opus 5
parent cf0a551f7b
commit 10e75b977f
23 changed files with 4596 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/web_interface/test_api_v3_backup_paths.py
+220
View File
@@ -0,0 +1,220 @@
"""
Path-containment tests for the backup file routes:
GET /backup/download/<filename>, DELETE /backup/<filename>, and the
listing/validation routes alongside them.
Both filename routes take user input straight from the URL and turn it
into a filesystem path, one to read and one to unlink. `_safe_backup_path`
is what stops that from reaching outside the export directory, and it had
no tests.
This is verification of existing containment, not a fix: no bypass was
found. The tests exist so that a later "just let dots through" change has
to argue with something.
"""
import io
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from flask import Flask
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from web_interface.blueprints import api_v3 as api_v3_module # noqa: E402
from web_interface.blueprints.api_v3 import api_v3 # noqa: E402
_MANAGER_ATTRS = (
'config_manager', 'plugin_manager', 'plugin_store_manager',
'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
'operation_queue', 'operation_history', 'cache_manager',
)
_SENTINEL = object()
# Anything that tries to name a file outside the export directory, or that
# is not a plain <name>.zip.
TRAVERSAL_ATTEMPTS = [
"../../etc/passwd",
"../config.json",
"..%2f..%2fetc%2fpasswd",
"....//....//etc/passwd",
"/etc/passwd",
"..\\..\\config.json",
"backup.zip/../../../etc/passwd",
".hidden.zip",
"backup.txt",
"backup.zip.exe",
"",
".",
"..",
]
@pytest.fixture
def env(tmp_path, monkeypatch):
export_dir = tmp_path / "backups"
export_dir.mkdir()
monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", export_dir)
# A file outside the export dir that a traversal would be reaching for.
secret = tmp_path / "config.json"
secret.write_text(json.dumps({"secret": "do not touch"}))
originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
for name in _MANAGER_ATTRS:
setattr(api_v3, name, MagicMock())
app = Flask(__name__)
app.config["TESTING"] = True
app.register_blueprint(api_v3, url_prefix="/api/v3")
class Env:
pass
e = Env()
e.client = app.test_client()
e.export_dir = export_dir
e.secret = secret
yield e
for name, original in originals.items():
if original is _SENTINEL:
if hasattr(api_v3, name):
delattr(api_v3, name)
else:
setattr(api_v3, name, original)
def make_backup(export_dir, name="backup-2026-01-01.zip"):
path = export_dir / name
path.write_bytes(b"PK\x03\x04fake zip")
return path
class TestSafeBackupPath:
"""The containment helper itself."""
@pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
def test_rejects_unsafe_names(self, env, filename):
assert api_v3_module._safe_backup_path(filename) is None
def test_rejects_none(self, env):
assert api_v3_module._safe_backup_path(None) is None
@pytest.mark.parametrize("filename", [
"backup.zip",
"backup-2026-01-01.zip",
"backup_2026.01.01-v2.zip",
"a.zip",
])
def test_accepts_plain_zip_names(self, env, filename):
resolved = api_v3_module._safe_backup_path(filename)
assert resolved is not None
assert resolved.parent == env.export_dir.resolve()
def test_result_is_always_inside_the_export_dir(self, env):
resolved = api_v3_module._safe_backup_path("backup.zip")
resolved.relative_to(env.export_dir.resolve()) # raises if outside
def test_overlong_name_rejected(self, env):
assert api_v3_module._safe_backup_path("a" * 250 + ".zip") is None
class TestDownload:
def test_downloads_an_existing_backup(self, env):
make_backup(env.export_dir)
response = env.client.get("/api/v3/backup/download/backup-2026-01-01.zip")
assert response.status_code == 200
assert response.data == b"PK\x03\x04fake zip"
def test_missing_file_is_a_404(self, env):
response = env.client.get("/api/v3/backup/download/never-made.zip")
assert response.status_code == 404
@pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
def test_traversal_attempts_are_refused(self, env, filename):
response = env.client.get(f"/api/v3/backup/download/{filename}")
# However the request is turned away — 404 from the containment
# check, or 308/405 from routing never matching at all — what
# matters is that no file outside the export directory is served.
assert response.status_code != 200
assert b"do not touch" not in response.data
class TestDelete:
def test_deletes_an_existing_backup(self, env):
path = make_backup(env.export_dir)
response = env.client.delete("/api/v3/backup/backup-2026-01-01.zip")
assert response.status_code == 200
assert not path.exists()
def test_missing_file_is_a_404(self, env):
response = env.client.delete("/api/v3/backup/never-made.zip")
assert response.status_code == 404
@pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
def test_traversal_attempts_delete_nothing(self, env, filename):
response = env.client.delete(f"/api/v3/backup/{filename}")
assert response.status_code != 200
assert env.secret.exists() # the file a traversal was aiming at
def test_only_the_named_backup_is_removed(self, env):
keep = make_backup(env.export_dir, "keep.zip")
drop = make_backup(env.export_dir, "drop.zip")
env.client.delete("/api/v3/backup/drop.zip")
assert keep.exists()
assert not drop.exists()
def test_directory_with_a_matching_name_is_not_removed(self, env):
# The delete loop matches by name but requires a regular file.
(env.export_dir / "sneaky.zip").mkdir()
response = env.client.delete("/api/v3/backup/sneaky.zip")
assert response.status_code == 404
assert (env.export_dir / "sneaky.zip").is_dir()
class TestList:
def test_lists_only_zip_files(self, env):
make_backup(env.export_dir, "one.zip")
(env.export_dir / "notes.txt").write_text("ignore me")
response = env.client.get("/api/v3/backup/list")
assert response.status_code == 200
names = [entry["filename"] for entry in response.get_json()["data"]]
assert names == ["one.zip"]
def test_empty_directory_lists_nothing(self, env):
response = env.client.get("/api/v3/backup/list")
assert response.get_json()["data"] == []
def test_entries_carry_size_and_timestamp(self, env):
make_backup(env.export_dir, "one.zip")
entry = env.client.get("/api/v3/backup/list").get_json()["data"][0]
assert entry["size"] == len(b"PK\x03\x04fake zip")
assert entry["created_at"]
class TestValidate:
def test_missing_file_is_a_400(self, env):
response = env.client.post("/api/v3/backup/validate", data={},
content_type="multipart/form-data")
assert response.status_code == 400
assert "No backup_file" in response.get_json()["message"]
def test_invalid_archive_is_a_400(self, env):
response = env.client.post(
"/api/v3/backup/validate",
data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
content_type="multipart/form-data")
assert response.status_code == 400
assert "Invalid or corrupted" in response.get_json()["message"]
def test_validation_does_not_leave_temp_files_in_the_export_dir(self, env):
env.client.post(
"/api/v3/backup/validate",
data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
content_type="multipart/form-data")
assert list(env.export_dir.iterdir()) == []
test/web_interface/test_api_v3_backup_restore.py
+262
View File
@@ -0,0 +1,262 @@
"""
Endpoint tests for POST /backup/restore.
Restore is the most destructive operation the web interface exposes: it
overwrites config, secrets, WiFi settings and fonts, and reinstalls
plugins. It had no tests.
restore_backup itself is mocked — this file is about what the route does
with the request and with the result, not about ZIP handling, which
belongs to backup_manager's own tests.
Regression coverage for one fixed bug: a malformed `options` field fell
back to {}, and since every RestoreOptions flag defaults to True, that
turned a mis-serialized narrow restore into a full one — secrets
included — with no indication anything had been ignored.
"""
import io
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from web_interface.blueprints.api_v3 import api_v3 # noqa: E402
URL = "/api/v3/backup/restore"
_MANAGER_ATTRS = (
'config_manager', 'plugin_manager', 'plugin_store_manager',
'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
'operation_queue', 'operation_history', 'cache_manager',
)
_SENTINEL = object()
class FakeResult:
"""Stand-in for backup_manager.RestoreResult."""
def __init__(self, success=True, restored=None, errors=None,
plugins_to_install=None):
self.success = success
self.restored = restored if restored is not None else ["config"]
self.errors = errors or []
self.plugins_to_install = plugins_to_install or []
self.plugins_installed = []
self.plugins_failed = []
def to_dict(self):
return {
"success": self.success,
"restored": self.restored,
"errors": self.errors,
"plugins_installed": self.plugins_installed,
"plugins_failed": self.plugins_failed,
}
@pytest.fixture
def client():
originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
for name in _MANAGER_ATTRS:
setattr(api_v3, name, MagicMock())
app = Flask(__name__)
app.config["TESTING"] = True
app.register_blueprint(api_v3, url_prefix="/api/v3")
yield app.test_client()
for name, original in originals.items():
if original is _SENTINEL:
if hasattr(api_v3, name):
delattr(api_v3, name)
else:
setattr(api_v3, name, original)
@pytest.fixture
def restore():
"""Patch backup_manager.restore_backup (imported inside the handler)."""
with patch("src.backup_manager.restore_backup") as mock:
mock.return_value = FakeResult()
yield mock
def post(client, options=None, filename="backup.zip", content=b"PK\x03\x04fake"):
data = {"backup_file": (io.BytesIO(content), filename)}
if options is not None:
data["options"] = options
return client.post(URL, data=data, content_type="multipart/form-data")
class TestRequestValidation:
def test_missing_file_is_a_400(self, client, restore):
response = client.post(URL, data={}, content_type="multipart/form-data")
assert response.status_code == 400
assert "No backup_file" in response.get_json()["message"]
restore.assert_not_called()
def test_absent_options_defaults_to_a_full_restore(self, client, restore):
# Documented default, not the bug: omitting options entirely means
# "restore everything".
post(client)
options = restore.call_args[0][2]
assert options.restore_config is True
assert options.restore_secrets is True
assert options.reinstall_plugins is True
def test_partial_options_are_honoured(self, client, restore):
post(client, options=json.dumps({
"restore_secrets": False, "reinstall_plugins": False}))
options = restore.call_args[0][2]
assert options.restore_secrets is False
assert options.reinstall_plugins is False
assert options.restore_config is True # unspecified stays default
@pytest.mark.parametrize("raw", ["{not json", "", "{'single': 'quotes'}"])
def test_malformed_options_are_refused(self, client, restore, raw):
# Regression: this fell back to {}, and every flag defaults to
# True, so a caller asking for a narrow restore and mis-serializing
# it got a full one — secrets overwritten — and no warning.
response = post(client, options=raw)
assert response.status_code == 400
assert "Invalid options" in response.get_json()["message"]
restore.assert_not_called()
@pytest.mark.parametrize("raw", ["[1,2,3]", '"a string"', "42", "true", "null"])
def test_options_that_are_not_an_object_are_refused(self, client, restore, raw):
response = post(client, options=raw)
assert response.status_code == 400
restore.assert_not_called()
def test_empty_object_is_accepted_as_all_defaults(self, client, restore):
assert post(client, options="{}").status_code == 200
assert restore.call_args[0][2].restore_config is True
class TestSuccess:
def test_success_returns_the_result(self, client, restore):
restore.return_value = FakeResult(success=True, restored=["config", "secrets"])
response = post(client)
assert response.status_code == 200
body = response.get_json()
assert body["status"] == "success"
assert body["data"]["restored"] == ["config", "secrets"]
def test_temp_file_is_cleaned_up(self, client, restore):
seen = {}
def capture(path, project_root, options):
seen["path"] = Path(path)
assert seen["path"].exists() # present while restoring
return FakeResult()
restore.side_effect = capture
post(client)
assert not seen["path"].exists()
def test_temp_file_cleaned_up_even_when_restore_raises(self, client, restore):
seen = {}
def blow_up(path, project_root, options):
seen["path"] = Path(path)
raise RuntimeError("corrupt archive")
restore.side_effect = blow_up
response = post(client)
assert response.status_code == 500
assert not seen["path"].exists()
class TestPluginReinstall:
def test_plugins_are_reinstalled_when_requested(self, client, restore):
restore.return_value = FakeResult(
plugins_to_install=[{"plugin_id": "clock"}, {"plugin_id": "weather"}])
api_v3.plugin_store_manager.install_plugin.return_value = True
response = post(client)
assert response.status_code == 200
assert response.get_json()["data"]["plugins_installed"] == ["clock", "weather"]
def test_reinstall_skipped_when_not_requested(self, client, restore):
restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
post(client, options=json.dumps({"reinstall_plugins": False}))
api_v3.plugin_store_manager.install_plugin.assert_not_called()
def test_entries_without_a_plugin_id_are_skipped(self, client, restore):
restore.return_value = FakeResult(plugins_to_install=[{}, {"plugin_id": "clock"}])
api_v3.plugin_store_manager.install_plugin.return_value = True
post(client)
assert api_v3.plugin_store_manager.install_plugin.call_count == 1
def test_failed_reinstall_turns_the_whole_restore_into_an_error(
self, client, restore):
# Pinned as intentional: file restoration succeeded and does not
# touch result.errors, but a user whose plugins did not come back
# should not be told the restore was a success.
restore.return_value = FakeResult(
success=True, plugins_to_install=[{"plugin_id": "clock"}])
api_v3.plugin_store_manager.install_plugin.return_value = False
response = post(client)
assert response.status_code == 500
body = response.get_json()
assert body["status"] == "error"
assert "clock" in body["message"]
def test_message_names_what_landed_and_what_did_not(self, client, restore):
restore.return_value = FakeResult(
success=True, restored=["config", "fonts"],
plugins_to_install=[{"plugin_id": "clock"}])
api_v3.plugin_store_manager.install_plugin.return_value = False
message = post(client).get_json()["message"]
assert "restored: config, fonts" in message
assert "plugins not reinstalled: clock" in message
def test_install_exception_is_recorded_without_leaking_details(
self, client, restore):
restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
api_v3.plugin_store_manager.install_plugin.side_effect = RuntimeError(
"/srv/internal/path exploded")
body = post(client).get_json()
failures = body["data"]["plugins_failed"]
assert failures[0]["plugin_id"] == "clock"
assert "/srv/internal/path" not in json.dumps(body)
def test_missing_store_manager_is_reported_per_plugin(self, client, restore):
restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
api_v3.plugin_store_manager = None
with patch("web_interface.blueprints.api_v3.plugin_store_manager", None):
body = post(client).get_json()
assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable"
class TestFailureReporting:
def test_restore_errors_produce_a_500(self, client, restore):
restore.return_value = FakeResult(
success=False, restored=[], errors=["config: permission denied"])
response = post(client)
assert response.status_code == 500
assert "permission denied" in response.get_json()["message"]
def test_partial_restore_names_both_sides(self, client, restore):
restore.return_value = FakeResult(
success=False, restored=["config"], errors=["secrets: unwritable"])
message = post(client).get_json()["message"]
assert "restored: config" in message
assert "failed: secrets: unwritable" in message
def test_failure_without_detail_still_says_something(self, client, restore):
restore.return_value = FakeResult(success=False, restored=[], errors=[])
message = post(client).get_json()["message"]
assert "Restore incomplete" in message
def test_unexpected_exception_is_a_500(self, client, restore):
restore.side_effect = RuntimeError("boom")
response = post(client)
assert response.status_code == 500
assert response.get_json()["status"] == "error"
test/web_interface/test_api_v3_config_raw.py
+204
View File
@@ -0,0 +1,204 @@
"""
Endpoint tests for POST /config/raw/main and POST /config/raw/secrets.
These write whatever JSON they are given straight to config.json and
config_secrets.json, bypassing the secret-separation path that
/config/main and the plugin-config endpoints go through. Given how much
care the rest of the config surface takes to keep secrets out of
config.json, an untested pair of endpoints that writes it verbatim is
worth pinning precisely.
Like test_api_v3_secret_roundtrip.py, these run a REAL ConfigManager over
tmp_path so the assertions are against files on disk rather than mock
calls.
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from flask import Flask
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from src.config_manager import ConfigManager # noqa: E402
from src.exceptions import ConfigError # noqa: E402
from web_interface.blueprints.api_v3 import api_v3 # noqa: E402
MAIN = "/api/v3/config/raw/main"
SECRETS = "/api/v3/config/raw/secrets"
@pytest.fixture
def env(tmp_path):
config_file = tmp_path / "config.json"
config_file.write_text(json.dumps({"timezone": "UTC"}))
secrets_file = tmp_path / "config_secrets.json"
config_manager = ConfigManager(
config_path=str(config_file), secrets_path=str(secrets_file))
config_manager.template_path = str(tmp_path / "no-template.json")
_SENTINEL = object()
attrs = ('config_manager', 'plugin_manager', 'plugin_store_manager',
'plugin_state_manager', 'saved_repositories_manager',
'schema_manager', 'operation_queue', 'operation_history',
'cache_manager')
originals = {name: getattr(api_v3, name, _SENTINEL) for name in attrs}
for name in attrs:
setattr(api_v3, name, MagicMock())
api_v3.config_manager = config_manager
app = Flask(__name__)
app.config["TESTING"] = True
app.register_blueprint(api_v3, url_prefix="/api/v3")
class Env:
pass
e = Env()
e.client = app.test_client()
e.config_manager = config_manager
e.config_file = config_file
e.secrets_file = secrets_file
yield e
for name, original in originals.items():
if original is _SENTINEL:
if hasattr(api_v3, name):
delattr(api_v3, name)
else:
setattr(api_v3, name, original)
class TestSaveRawMain:
def test_writes_the_body_to_config_json(self, env):
response = env.client.post(MAIN, json={"timezone": "America/Chicago"})
assert response.status_code == 200
assert json.loads(env.config_file.read_text()) == {"timezone": "America/Chicago"}
def test_replaces_rather_than_merges(self, env):
env.client.post(MAIN, json={"only": "this"})
assert json.loads(env.config_file.read_text()) == {"only": "this"}
def test_does_not_touch_the_secrets_file(self, env):
env.secrets_file.write_text(json.dumps({"weather": {"api_key": "k"}}))
env.client.post(MAIN, json={"timezone": "UTC"})
assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "k"}}
def test_uninitialized_manager_is_a_500(self, env):
api_v3.config_manager = None
response = env.client.post(MAIN, json={"timezone": "UTC"})
assert response.status_code == 500
assert "not initialized" in response.get_json()["message"]
def test_empty_object_is_a_400(self, env):
response = env.client.post(MAIN, json={})
assert response.status_code == 400
assert "No data provided" in response.get_json()["message"]
def test_bodyless_post_is_a_400(self, env):
response = env.client.post(MAIN)
assert response.status_code == 400
assert "No data provided" in response.get_json()["message"]
def test_malformed_json_is_a_400_in_the_app_shape(self, env):
response = env.client.post(MAIN, data="{not json",
content_type="application/json")
assert response.status_code == 400
body = response.get_json()
assert body["status"] == "error"
# A body that was sent but does not parse is a distinct mistake
# from sending none, and says so. Previously the handler's own
# json.JSONDecodeError arm was unreachable — Werkzeug raised
# first — so this collapsed into "No data provided".
assert "Invalid JSON in request body" in body["message"]
def test_config_error_is_a_500_with_context(self, env, monkeypatch):
def refuse(kind, data):
raise ConfigError("cannot write", config_path="/etc/x.json")
monkeypatch.setattr(env.config_manager, "save_raw_file_content", refuse)
response = env.client.post(MAIN, json={"timezone": "UTC"})
assert response.status_code == 500
assert "/etc/x.json" in json.dumps(response.get_json())
def test_unexpected_error_is_a_500(self, env, monkeypatch):
def boom(kind, data):
raise RuntimeError("disk on fire")
monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
response = env.client.post(MAIN, json={"timezone": "UTC"})
assert response.status_code == 500
assert response.get_json()["status"] == "error"
class TestSaveRawSecrets:
def test_writes_only_to_the_secrets_file(self, env):
response = env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
assert response.status_code == 200
assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "s3cret"}}
def test_secret_values_never_reach_config_json(self, env):
env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
assert "s3cret" not in env.config_file.read_text()
def test_existing_main_config_is_untouched(self, env):
before = env.config_file.read_text()
env.client.post(SECRETS, json={"weather": {"api_key": "k"}})
assert env.config_file.read_text() == before
def test_github_token_is_reloaded_for_the_store_manager(self, env):
store = MagicMock()
store._load_github_token.return_value = "ghp_new"
api_v3.plugin_store_manager = store
env.client.post(SECRETS, json={"github": {"token": "ghp_new"}})
store._load_github_token.assert_called_once()
assert store.github_token == "ghp_new"
def test_absent_store_manager_is_fine(self, env):
api_v3.plugin_store_manager = None
assert env.client.post(SECRETS, json={"a": 1}).status_code == 200
def test_uninitialized_manager_is_a_500(self, env):
api_v3.config_manager = None
assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
def test_empty_object_is_a_400(self, env):
assert env.client.post(SECRETS, json={}).status_code == 400
def test_bodyless_post_is_a_400(self, env):
assert env.client.post(SECRETS).status_code == 400
def test_error_is_a_500(self, env, monkeypatch):
def boom(kind, data):
raise RuntimeError("nope")
monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
class TestRawEndpointsBypassSecretSeparation:
"""Pinned behaviour, deliberately not "fixed".
These endpoints are the escape hatch for editing the config files
directly from the web UI's raw JSON editor. They write what they are
given, so a secret typed into the main-config editor lands in
config.json in plain text — unlike /config/main and the plugin-config
endpoints, which route x-secret fields into config_secrets.json.
That is the point of a raw editor, but it is a sharp edge worth
stating out loud: anyone adding a "convenience" that posts plugin
config through this endpoint would silently lose secret separation.
"""
def test_secret_shaped_keys_are_written_verbatim_to_main(self, env):
env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
on_disk = json.loads(env.config_file.read_text())
assert on_disk["weather"]["api_key"] == "PLAINTEXT-KEY"
def test_no_separation_happens_on_the_raw_path(self, env):
env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
# Nothing was moved aside into the secrets file.
assert not env.secrets_file.exists() or "PLAINTEXT-KEY" not in env.secrets_file.read_text()
test/web_interface/test_error_handler.py
+149
View File
@@ -0,0 +1,149 @@
"""
Tests for the response builders in src/web_interface/error_handler.py and
the success path in src/web_interface/api_helpers.py.
describe_exception() in the same module is already covered by
test/test_web_error_detail.py and is not duplicated here.
Regression coverage for one fixed bug: create_success_response used
truthiness for `message` and `metadata` while using `is not None` for
`data`, so an explicitly-passed "" or {} was silently dropped —
api_helpers.success_response() repeated the same gate, which is the path
every api_v3 endpoint actually calls.
"""
import pytest
from flask import Flask
from src.web_interface.api_helpers import success_response
from src.web_interface.error_handler import (
create_error_response,
create_success_response,
)
from src.web_interface.errors import ErrorCode, WebInterfaceError
@pytest.fixture
def app():
return Flask(__name__)
class TestCreateErrorResponse:
def test_returns_response_and_status_tuple(self, app):
with app.test_request_context():
response, status = create_error_response(
ErrorCode.CONFIG_SAVE_FAILED, "could not save")
assert status == 500
assert response.get_json()["message"] == "could not save"
def test_status_code_passthrough(self, app):
with app.test_request_context():
_, status = create_error_response(
ErrorCode.INVALID_INPUT, "bad", status_code=400)
assert status == 400
def test_body_matches_the_error_dataclass(self, app):
with app.test_request_context():
response, _ = create_error_response(
ErrorCode.NETWORK_ERROR, "offline",
details="connection refused", context={"url": "http://x"})
expected = WebInterfaceError(
error_code=ErrorCode.NETWORK_ERROR, message="offline",
details="connection refused", context={"url": "http://x"}).to_dict()
assert response.get_json() == expected
def test_none_context_produces_no_context_key(self, app):
with app.test_request_context():
response, _ = create_error_response(ErrorCode.SYSTEM_ERROR, "boom")
assert "context" not in response.get_json()
def test_suggested_fixes_passed_through(self, app):
with app.test_request_context():
response, _ = create_error_response(
ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=["Try again"])
assert response.get_json()["suggested_fixes"] == ["Try again"]
class TestCreateSuccessResponse:
def test_bare_success(self):
assert create_success_response() == {"status": "success"}
def test_data_included(self):
assert create_success_response(data={"a": 1})["data"] == {"a": 1}
@pytest.mark.parametrize("falsy", [0, "", False, {}, []])
def test_falsy_data_is_still_included(self, falsy):
assert create_success_response(data=falsy)["data"] == falsy
def test_none_data_omitted(self):
assert "data" not in create_success_response(data=None)
def test_message_included(self):
assert create_success_response(message="done")["message"] == "done"
def test_empty_message_is_still_included(self):
# Regression: `if message:` dropped an explicitly-passed "".
assert create_success_response(message="")["message"] == ""
def test_none_message_omitted(self):
assert "message" not in create_success_response(message=None)
def test_metadata_included(self):
assert create_success_response(metadata={"v": 1})["metadata"] == {"v": 1}
def test_empty_metadata_is_still_included(self):
# Regression: `if metadata:` dropped an explicitly-passed {}.
assert create_success_response(metadata={})["metadata"] == {}
def test_none_metadata_omitted(self):
assert "metadata" not in create_success_response(metadata=None)
class TestSuccessResponseHelper:
"""api_helpers.success_response — the wrapper every endpoint calls."""
def test_plain_response_has_no_metadata_block(self, app):
with app.test_request_context():
body = success_response(data={"a": 1}).get_json()
assert body == {"status": "success", "data": {"a": 1}}
def test_explicit_empty_metadata_survives_the_wrapper(self, app):
# Regression: the wrapper re-gated metadata on truthiness after
# create_success_response had already included it, so {} was
# dropped again on the way out.
with app.test_request_context():
body = success_response(data=None, metadata={}).get_json()
assert body["metadata"] == {}
def test_caller_metadata_preserved(self, app):
with app.test_request_context():
body = success_response(metadata={"version": "1.2"}).get_json()
assert body["metadata"]["version"] == "1.2"
def test_timing_added_when_request_has_start_time(self, app):
with app.test_request_context() as ctx:
ctx.request.start_time = 0.0
body = success_response(data={"a": 1}).get_json()
assert "response_time_ms" in body["metadata"]
def test_timing_merges_with_caller_metadata(self, app):
with app.test_request_context() as ctx:
ctx.request.start_time = 0.0
body = success_response(metadata={"version": "1.2"}).get_json()
assert body["metadata"]["version"] == "1.2"
assert "response_time_ms" in body["metadata"]
def test_caller_metadata_dict_is_not_mutated(self, app):
# The helper used to add response_time_ms straight into the dict the
# caller passed, so a module-level or reused metadata dict would
# accumulate timings from previous requests.
caller_metadata = {"version": "1.2"}
with app.test_request_context() as ctx:
ctx.request.start_time = 0.0
success_response(metadata=caller_metadata)
assert caller_metadata == {"version": "1.2"}
def test_message_passed_through(self, app):
with app.test_request_context():
body = success_response(message="saved").get_json()
assert body["message"] == "saved"
test/web_interface/test_errors.py
+208
View File
@@ -0,0 +1,208 @@
"""
Tests for src/web_interface/errors.py — the structured error type behind
every API error response (category inference, default suggestions, the
JSON shape, and exception conversion).
Pure logic; no Flask context needed.
Regression coverage for one fixed bug: suggested_fixes used `or`, so a
caller passing [] to mean "no suggestions" silently got the default list.
"""
import pytest
from src.web_interface.errors import ErrorCategory, ErrorCode, WebInterfaceError
class TestCategoryInference:
@pytest.mark.parametrize("code,expected", [
(ErrorCode.CONFIG_SAVE_FAILED, ErrorCategory.CONFIGURATION),
(ErrorCode.CONFIG_ROLLBACK_FAILED, ErrorCategory.CONFIGURATION),
(ErrorCode.PLUGIN_NOT_FOUND, ErrorCategory.PLUGIN),
(ErrorCode.PLUGIN_OPERATION_CONFLICT, ErrorCategory.PLUGIN),
(ErrorCode.VALIDATION_ERROR, ErrorCategory.VALIDATION),
(ErrorCode.SCHEMA_VALIDATION_FAILED, ErrorCategory.VALIDATION),
(ErrorCode.INVALID_INPUT, ErrorCategory.VALIDATION),
(ErrorCode.NETWORK_ERROR, ErrorCategory.NETWORK),
(ErrorCode.API_ERROR, ErrorCategory.NETWORK),
(ErrorCode.TIMEOUT, ErrorCategory.NETWORK),
(ErrorCode.PERMISSION_DENIED, ErrorCategory.PERMISSION),
(ErrorCode.FILE_PERMISSION_ERROR, ErrorCategory.PERMISSION),
(ErrorCode.SYSTEM_ERROR, ErrorCategory.SYSTEM),
(ErrorCode.SERVICE_UNAVAILABLE, ErrorCategory.SYSTEM),
(ErrorCode.UNKNOWN_ERROR, ErrorCategory.UNKNOWN),
])
def test_every_code_prefix_maps_to_its_category(self, code, expected):
assert WebInterfaceError(code, "msg").category is expected
def test_explicit_category_overrides_inference(self):
error = WebInterfaceError(
ErrorCode.CONFIG_SAVE_FAILED, "msg", category=ErrorCategory.SYSTEM)
assert error.category is ErrorCategory.SYSTEM
def test_every_error_code_gets_a_category(self):
# No code may fall through uncategorized as the enum grows.
for code in ErrorCode:
assert isinstance(WebInterfaceError(code, "msg").category, ErrorCategory)
class TestDefaultSuggestions:
def test_mapped_code_gets_specific_suggestions(self):
fixes = WebInterfaceError(ErrorCode.CONFIG_SAVE_FAILED, "msg").suggested_fixes
assert "Check available disk space" in fixes
def test_unmapped_code_gets_generic_fallback(self):
# PLUGIN_UPDATE_FAILED has no entry in suggestions_map.
fixes = WebInterfaceError(ErrorCode.PLUGIN_UPDATE_FAILED, "msg").suggested_fixes
assert fixes == ["Review error details and try again"]
def test_explicit_suggestions_win(self):
error = WebInterfaceError(
ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=["Do the thing"])
assert error.suggested_fixes == ["Do the thing"]
def test_explicit_empty_list_is_respected(self):
# Regression: `suggested_fixes or default` treated [] as "unset",
# so a caller could not express "I have no suggestions".
error = WebInterfaceError(
ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=[])
assert error.suggested_fixes == []
def test_none_still_gets_defaults(self):
error = WebInterfaceError(
ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=None)
assert len(error.suggested_fixes) > 0
class TestToDict:
def test_base_keys_always_present(self):
result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
assert result["status"] == "error"
assert result["error_code"] == "SYSTEM_ERROR"
assert result["error_category"] == "system"
assert result["message"] == "boom"
def test_details_included_when_set(self):
result = WebInterfaceError(
ErrorCode.SYSTEM_ERROR, "boom", details="disk full").to_dict()
assert result["details"] == "disk full"
def test_details_omitted_when_absent(self):
assert "details" not in WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
def test_context_included_when_non_empty(self):
result = WebInterfaceError(
ErrorCode.SYSTEM_ERROR, "boom", context={"path": "/tmp/x"}).to_dict()
assert result["context"] == {"path": "/tmp/x"}
def test_empty_context_is_omitted(self):
# Pinned as intentional, not a bug: __init__ normalizes context to
# {}, and an empty context carries no information, so it is left out
# rather than padding every error body with "context": {}.
result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom", context={}).to_dict()
assert "context" not in result
def test_empty_suggestions_omitted(self):
result = WebInterfaceError(
ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=[]).to_dict()
assert "suggested_fixes" not in result
def test_is_json_serializable(self):
import json
error = WebInterfaceError(
ErrorCode.NETWORK_ERROR, "boom",
details="timeout", context={"url": "http://x"})
assert json.loads(json.dumps(error.to_dict()))["error_code"] == "NETWORK_ERROR"
class TestFromException:
@pytest.mark.parametrize("exc_name,expected", [
("ConfigError", ErrorCode.CONFIG_LOAD_FAILED),
("PluginError", ErrorCode.PLUGIN_LOAD_FAILED),
("PermissionError", ErrorCode.PERMISSION_DENIED),
("AccessDenied", ErrorCode.PERMISSION_DENIED),
("ValidationError", ErrorCode.VALIDATION_ERROR),
("SchemaError", ErrorCode.VALIDATION_ERROR),
("NetworkError", ErrorCode.NETWORK_ERROR),
("ConnectionError", ErrorCode.NETWORK_ERROR),
("TimeoutError", ErrorCode.TIMEOUT),
("SomethingElse", ErrorCode.UNKNOWN_ERROR),
])
def test_code_inferred_from_exception_class_name(self, exc_name, expected):
exc = type(exc_name, (Exception,), {})("boom")
assert WebInterfaceError.from_exception(exc).error_code is expected
def test_explicit_code_skips_inference(self):
error = WebInterfaceError.from_exception(
ValueError("boom"), error_code=ErrorCode.PLUGIN_NOT_FOUND)
assert error.error_code is ErrorCode.PLUGIN_NOT_FOUND
def test_message_is_the_safe_one_not_the_exception_text(self):
# The raw exception text is not echoed into `message`; that field is
# a fixed, user-facing string per code.
error = WebInterfaceError.from_exception(ValueError("secret-ish detail"))
assert error.message == "An unexpected error occurred"
assert "secret-ish" not in error.message
def test_exception_type_recorded_in_context(self):
error = WebInterfaceError.from_exception(ValueError("boom"))
assert error.context["exception_type"] == "ValueError"
def test_caller_context_is_preserved_alongside_type(self):
error = WebInterfaceError.from_exception(
ValueError("boom"), context={"plugin_id": "clock"})
assert error.context["plugin_id"] == "clock"
assert error.context["exception_type"] == "ValueError"
def test_caller_supplied_exception_type_is_overwritten(self):
error = WebInterfaceError.from_exception(
ValueError("boom"), context={"exception_type": "Fake"})
assert error.context["exception_type"] == "ValueError"
def test_original_error_retained(self):
exc = ValueError("boom")
assert WebInterfaceError.from_exception(exc).original_error is exc
def test_every_code_has_a_safe_message(self):
for code in ErrorCode:
assert WebInterfaceError._safe_message(code)
class TestExceptionDetails:
def test_context_dict_is_flattened(self):
exc = ValueError("boom")
exc.context = {"config_path": "/etc/x.json", "line": 4}
details = WebInterfaceError._get_exception_details(exc)
assert "config_path: /etc/x.json" in details
assert "line: 4" in details
assert "; " in details
def test_exception_type_key_excluded(self):
exc = ValueError("boom")
exc.context = {"exception_type": "ValueError", "path": "/tmp/x"}
details = WebInterfaceError._get_exception_details(exc)
assert "exception_type" not in details
assert details == "path: /tmp/x"
def test_context_with_only_exception_type_gives_none(self):
exc = ValueError("boom")
exc.context = {"exception_type": "ValueError"}
assert WebInterfaceError._get_exception_details(exc) is None
def test_no_context_attribute_gives_none(self):
assert WebInterfaceError._get_exception_details(ValueError("boom")) is None
def test_non_dict_context_gives_none(self):
exc = ValueError("boom")
exc.context = "not a dict"
assert WebInterfaceError._get_exception_details(exc) is None
def test_empty_context_gives_none(self):
exc = ValueError("boom")
exc.context = {}
assert WebInterfaceError._get_exception_details(exc) is None
def test_details_flow_into_from_exception(self):
exc = ValueError("boom")
exc.context = {"config_path": "/etc/x.json"}
assert "config_path" in WebInterfaceError.from_exception(exc).details
test/web_interface/test_validators.py
+284
View File
@@ -0,0 +1,284 @@
"""
Tests for src/web_interface/validators.py.
dedup_unique_arrays is already covered by test_dedup_unique_arrays.py and
is not repeated here; this file covers the other eight functions, none of
which had any tests.
Regression coverage for three fixed bugs:
- validate_numeric_range accepted True/False, since bool subclasses int.
- validate_file_upload lowercased the filename's extension but not the
caller's allowed_extensions list, so ['.TTF'] rejected 'font.ttf'.
- validate_image_url only checked for '..' inside the relative-path
branch, so http://host/../secret passed validation untouched.
"""
import pytest
from src.web_interface.validators import (
escape_html,
sanitize_plugin_config,
validate_file_upload,
validate_font_awesome_class,
validate_image_url,
validate_mime_type,
validate_numeric_range,
validate_string_length,
)
class TestEscapeHtml:
def test_escapes_all_five_entities(self):
assert escape_html("""<a href="x">O'Neill & co</a>""") == (
"&lt;a href=&quot;x&quot;&gt;O&#x27;Neill &amp; co&lt;/a&gt;")
def test_ampersand_is_escaped_first_so_nothing_double_escapes(self):
# If '<' were replaced before '&', the '&' of '&lt;' would be
# escaped again into '&amp;lt;'.
assert escape_html("<") == "&lt;"
assert escape_html("&") == "&amp;"
assert escape_html("&<") == "&amp;&lt;"
def test_plain_text_unchanged(self):
assert escape_html("hello world") == "hello world"
def test_non_string_is_coerced(self):
assert escape_html(42) == "42"
assert escape_html(None) == "None"
def test_script_tag_neutralized(self):
assert "<script>" not in escape_html("<script>alert(1)</script>")
class TestValidateImageUrl:
@pytest.mark.parametrize("url", [
"javascript:alert(1)",
"JavaScript:alert(1)",
"JAVASCRIPT:alert(1)",
"data:text/html;base64,PHNjcmlwdD4=",
"vbscript:msgbox(1)",
"file:///etc/passwd",
])
def test_dangerous_protocols_rejected(self, url):
valid, error = validate_image_url(url)
assert valid is False and "protocol" in error.lower()
@pytest.mark.parametrize("url", [
"http://x/a.png?onerror=alert(1)",
"http://x/a.png#onload=alert(1)",
"http://x/onclick=alert(1).png",
])
def test_event_handlers_rejected(self, url):
valid, error = validate_image_url(url)
assert valid is False and "Event handlers" in error
@pytest.mark.parametrize("url", ["", None, 123, []])
def test_empty_or_non_string_rejected(self, url):
assert validate_image_url(url)[0] is False
def test_http_and_https_allowed(self):
assert validate_image_url("http://example.com/logo.png") == (True, None)
assert validate_image_url("https://example.com/logo.png") == (True, None)
def test_other_schemes_rejected(self):
valid, error = validate_image_url("ftp://example.com/logo.png")
assert valid is False and "http://" in error
def test_relative_path_allowed(self):
assert validate_image_url("/static/logo.png") == (True, None)
def test_protocol_relative_url_rejected(self):
assert validate_image_url("//evil.com/logo.png")[0] is False
def test_relative_traversal_rejected(self):
assert validate_image_url("/static/../../etc/passwd")[0] is False
def test_absolute_url_traversal_rejected(self):
# Regression: the '..' check used to sit inside the leading-slash
# branch, so an absolute URL skipped it entirely.
valid, error = validate_image_url("http://example.com/../secret")
assert valid is False and "traversal" in error.lower()
def test_bare_traversal_rejected(self):
assert validate_image_url("../../etc/passwd")[0] is False
class TestValidateFontAwesomeClass:
@pytest.mark.parametrize("cls", ["fa-star", "fas fa-star", "fa-solid fa-house"])
def test_valid_classes_accepted(self, cls):
assert validate_font_awesome_class(cls) == (True, None)
@pytest.mark.parametrize("cls", ["star", "glyphicon-star", ""])
def test_classes_without_fa_prefix_rejected(self, cls):
assert validate_font_awesome_class(cls)[0] is False
def test_injection_attempt_rejected(self):
assert validate_font_awesome_class('fa-star" onload="alert(1)')[0] is False
def test_angle_brackets_rejected(self):
assert validate_font_awesome_class("<script>fa-star</script>")[0] is False
def test_non_string_rejected(self):
valid, error = validate_font_awesome_class(None)
assert valid is False and "string" in error
def test_explicit_fa_check_is_unreachable_but_harmless(self):
# Characterized, not fixed: the regex already requires 'fa-', so the
# follow-up `if 'fa-' not in class_name` can never fire. Anything
# lacking 'fa-' is rejected by the pattern first, with the pattern's
# own message.
valid, error = validate_font_awesome_class("star")
assert valid is False
assert error == "Invalid Font Awesome class name format"
class TestValidateFileUpload:
def test_plain_filename_accepted(self):
assert validate_file_upload("logo.png") == (True, None)
@pytest.mark.parametrize("filename", [
"../etc/passwd", "dir/file.png", "dir\\file.png", "..\\..\\secrets",
])
def test_traversal_characters_rejected(self, filename):
valid, error = validate_file_upload(filename)
assert valid is False and "invalid characters" in error
@pytest.mark.parametrize("filename", ["", None, 123])
def test_empty_or_non_string_rejected(self, filename):
assert validate_file_upload(filename)[0] is False
def test_allowed_extension_accepted(self):
assert validate_file_upload("font.ttf", allowed_extensions=[".ttf", ".otf"]) == (True, None)
def test_disallowed_extension_rejected(self):
valid, error = validate_file_upload("evil.exe", allowed_extensions=[".ttf"])
assert valid is False and "extension" in error
def test_uppercase_filename_extension_matches(self):
assert validate_file_upload("FONT.TTF", allowed_extensions=[".ttf"]) == (True, None)
def test_uppercase_allowed_list_matches(self):
# Regression: only the filename side was lowercased, so a caller
# passing ['.TTF'] rejected every valid .ttf upload.
assert validate_file_upload("font.ttf", allowed_extensions=[".TTF"]) == (True, None)
def test_no_extension_list_skips_the_check(self):
assert validate_file_upload("anything.xyz") == (True, None)
class TestValidateMimeType:
def test_known_type_accepted(self):
assert validate_mime_type("logo.png", ["image/png"]) == (True, None)
def test_mismatched_type_rejected(self):
valid, error = validate_mime_type("logo.png", ["image/jpeg"])
assert valid is False and "not allowed" in error
def test_undeterminable_type_rejected(self):
valid, error = validate_mime_type("mystery.zzz", ["image/png"])
assert valid is False and "Could not determine" in error
def test_guess_type_failure_is_caught(self, monkeypatch):
import mimetypes
monkeypatch.setattr(mimetypes, "guess_type",
lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")))
valid, error = validate_mime_type("logo.png", ["image/png"])
assert valid is False and "Error validating MIME type" in error
class TestValidateNumericRange:
def test_value_in_range(self):
assert validate_numeric_range(5, min_val=0, max_val=10) == (True, None)
def test_boundaries_are_inclusive(self):
assert validate_numeric_range(0, min_val=0, max_val=10) == (True, None)
assert validate_numeric_range(10, min_val=0, max_val=10) == (True, None)
def test_below_minimum_rejected(self):
valid, error = validate_numeric_range(-1, min_val=0)
assert valid is False and "at least" in error
def test_above_maximum_rejected(self):
valid, error = validate_numeric_range(11, max_val=10)
assert valid is False and "at most" in error
def test_floats_accepted(self):
assert validate_numeric_range(2.5, min_val=0, max_val=10) == (True, None)
def test_no_bounds_accepts_any_number(self):
assert validate_numeric_range(-9999) == (True, None)
@pytest.mark.parametrize("value", ["5", None, [], {}])
def test_non_numeric_rejected(self, value):
valid, error = validate_numeric_range(value, min_val=0, max_val=10)
assert valid is False and error == "Value must be a number"
@pytest.mark.parametrize("value", [True, False])
def test_booleans_rejected(self, value):
# Regression: bool subclasses int, so True passed the isinstance
# check and then compared as 1 against the range.
valid, error = validate_numeric_range(value, min_val=0, max_val=10)
assert valid is False and error == "Value must be a number"
class TestValidateStringLength:
def test_within_range(self):
assert validate_string_length("hello", min_length=1, max_length=10) == (True, None)
def test_boundaries_are_inclusive(self):
assert validate_string_length("abc", min_length=3, max_length=3) == (True, None)
def test_too_short_rejected(self):
valid, error = validate_string_length("", min_length=1)
assert valid is False and "at least" in error
def test_too_long_rejected(self):
valid, error = validate_string_length("abcdef", max_length=3)
assert valid is False and "at most" in error
def test_non_string_rejected(self):
valid, error = validate_string_length(123, max_length=10)
assert valid is False and "must be a string" in error
def test_no_bounds_accepts_anything(self):
assert validate_string_length("") == (True, None)
class TestSanitizePluginConfig:
def test_valid_keys_and_scalars_kept(self):
config = {"enabled": True, "count": 3, "ratio": 1.5, "name": "clock"}
assert sanitize_plugin_config(config) == config
@pytest.mark.parametrize("key", ["has space", "has-dash", "has.dot", "has/slash", ""])
def test_invalid_key_names_dropped(self, key):
assert sanitize_plugin_config({key: "value", "good": 1}) == {"good": 1}
def test_non_string_keys_dropped(self):
assert sanitize_plugin_config({1: "a", "good": 2}) == {"good": 2}
def test_nested_dicts_recursed(self):
result = sanitize_plugin_config({"outer": {"inner": 1, "bad key": 2}})
assert result == {"outer": {"inner": 1}}
def test_list_of_scalars_preserved(self):
assert sanitize_plugin_config({"teams": ["PHI", "NYG"]})["teams"] == ["PHI", "NYG"]
def test_list_of_dicts_recursed(self):
result = sanitize_plugin_config({"items": [{"ok": 1, "bad key": 2}]})
assert result["items"] == [{"ok": 1}]
def test_unknown_value_types_dropped(self):
assert sanitize_plugin_config({"weird": {1, 2, 3}, "good": 1}) == {"good": 1}
def test_none_values_dropped(self):
assert sanitize_plugin_config({"nothing": None, "good": 1}) == {"good": 1}
def test_strings_are_not_html_escaped(self):
# Pinned, not a bug: escaping here would persist the escaped form in
# config.json. Output escaping belongs to the template layer, which
# the function's docstring now says explicitly.
payload = "<script>alert(1)</script>"
assert sanitize_plugin_config({"title": payload})["title"] == payload
def test_empty_config(self):
assert sanitize_plugin_config({}) == {}