mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-13 22:58:06 +00:00
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
This commit is contained in:
@@ -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()) == []
|
||||||
@@ -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"
|
||||||
@@ -8181,7 +8181,16 @@ def backup_restore():
|
|||||||
try:
|
try:
|
||||||
opts_dict = json.loads(options_raw)
|
opts_dict = json.loads(options_raw)
|
||||||
except json.JSONDecodeError:
|
except json.JSONDecodeError:
|
||||||
opts_dict = {}
|
opts_dict = None
|
||||||
|
if not isinstance(opts_dict, dict):
|
||||||
|
# Every option defaults to True, so falling back to {} on a
|
||||||
|
# parse failure would silently perform a FULL restore —
|
||||||
|
# secrets and all — for a caller who asked for a narrow one
|
||||||
|
# and mis-serialized it. Refuse instead of guessing.
|
||||||
|
return jsonify({
|
||||||
|
'status': 'error',
|
||||||
|
'message': 'Invalid options: expected a JSON object',
|
||||||
|
}), 400
|
||||||
options = RestoreOptions(
|
options = RestoreOptions(
|
||||||
restore_config=bool(opts_dict.get('restore_config', True)),
|
restore_config=bool(opts_dict.get('restore_config', True)),
|
||||||
restore_secrets=bool(opts_dict.get('restore_secrets', True)),
|
restore_secrets=bool(opts_dict.get('restore_secrets', True)),
|
||||||
|
|||||||
Reference in New Issue
Block a user