mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 18:32:13 +00:00
139 lines
8.3 KiB
Python
139 lines
8.3 KiB
Python
"""Exercise the shipped routes and persistence; localization is additive."""
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from flask import Flask
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0, str(ROOT / "AppImage/scripts"))
|
|
import monitor_links as links
|
|
|
|
|
|
class MonitorLinksI18nTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.path = Path(temporary.name) / "links.json"
|
|
self.previous = {"manual": [{"id": "saved", "name": "Saved", "url": "https://saved.test", "product": "pve"}], "overrides": {}}
|
|
self.path.write_text(json.dumps(self.previous))
|
|
for patch in [mock.patch.object(links, "LINKS_FILE", self.path),
|
|
mock.patch.object(links, "refresh"), mock.patch.object(links, "_probe")]:
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
auth_provider = types.SimpleNamespace(
|
|
load_auth_config=lambda: {"enabled": True},
|
|
verify_token=lambda token: "fixture" if token in ("admin", "reader") else None,
|
|
verify_token_full=lambda token: ("fixture", "full_admin" if token == "admin" else "read_only")
|
|
if token in ("admin", "reader") else (None, None))
|
|
middleware_spec = importlib.util.spec_from_file_location("fixture_jwt_middleware", ROOT / "AppImage/scripts/jwt_middleware.py")
|
|
middleware = importlib.util.module_from_spec(middleware_spec)
|
|
# Execute the shipped decorators; only credential verification is faked.
|
|
with mock.patch.dict(sys.modules, {"auth_manager": auth_provider}):
|
|
middleware_spec.loader.exec_module(middleware)
|
|
spec = importlib.util.spec_from_file_location("fixture_monitor_links_routes", ROOT / "AppImage/scripts/flask_monitor_links_routes.py")
|
|
routes = importlib.util.module_from_spec(spec)
|
|
# Import isolation keeps the real JWT module and other suites untouched.
|
|
with mock.patch.dict(sys.modules, {"jwt_middleware": middleware}):
|
|
spec.loader.exec_module(routes)
|
|
app = Flask(__name__)
|
|
app.register_blueprint(routes.monitor_links_bp)
|
|
self.client = app.test_client()
|
|
|
|
def send(self, method, path="", payload=None, role="admin"):
|
|
return self.client.open("/api/monitor-links" + path, method=method,
|
|
json=payload, headers={"Authorization": "Bearer " + role} if role else {})
|
|
|
|
def rejected(self, method, path, payload, message, code, params=None):
|
|
before = self.path.read_bytes()
|
|
with mock.patch.object(links, "_save") as save:
|
|
response = self.send(method, path, payload)
|
|
self.assertEqual(response.status_code, 400)
|
|
expected = {"error": message, "code": code}
|
|
if params:
|
|
expected["params"] = params
|
|
self.assertEqual(response.json, expected)
|
|
self.assertEqual(response.headers["Cache-Control"], "no-store")
|
|
save.assert_not_called()
|
|
links.refresh.assert_not_called()
|
|
links._probe.assert_not_called()
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
|
|
def test_validation_errors_preserve_english_status_and_have_stable_codes_without_effects(self):
|
|
cases = [
|
|
("POST", "", None, "Invalid request.", "monitor_link_invalid_request"),
|
|
("PATCH", "/manual:saved", [], "Invalid request.", "monitor_link_invalid_request"),
|
|
("POST", "", {"url": " "}, "Enter the address of the Monitor.", "monitor_link_address_required"),
|
|
("POST", "", {"url": "ftp://example.test"}, "The address must start with http:// or https:// and name a host.", "monitor_link_invalid_address"),
|
|
("POST", "", {"url": "https://user:pass@example.test"}, "The address must not contain a user or password.", "monitor_link_address_credentials"),
|
|
("PATCH", "/manual:saved", {"name": "Unsaved", "url": "https://example.test:99999"}, "The port of the address is not valid.", "monitor_link_invalid_port"),
|
|
("PATCH", "/cluster:node", {"url": "ftp://example.test"}, "The address must start with http:// or https:// and name a host.", "monitor_link_invalid_address"),
|
|
("POST", "", {"url": "https://saved.test/path"}, "This address is already in the list.", "monitor_link_duplicate"),
|
|
("PATCH", "/manual:absent", {"url": "https://new.test"}, "Node not found.", "monitor_link_not_found"),
|
|
("PATCH", "/other:node", {}, "Node not found.", "monitor_link_not_found"),
|
|
("DELETE", "/cluster:node", None, "Only nodes added by hand can be removed; detected ones can be hidden.", "monitor_link_detected_remove"),
|
|
]
|
|
for case in cases:
|
|
with self.subTest(case=case):
|
|
self.rejected(*case)
|
|
|
|
def test_limit_uses_actual_count_and_preserves_previous_file(self):
|
|
with mock.patch.object(links, "MAX_MANUAL", 1):
|
|
self.rejected("POST", "", {"url": "https://new.test"}, "At most 1 nodes can be added.",
|
|
"monitor_link_limit", {"count": 1})
|
|
|
|
def test_write_failure_is_localizable_without_leaking_filesystem_details(self):
|
|
for method, path, payload in [("POST", "", {"url": "https://new.test"}),
|
|
("PATCH", "/manual:saved", {"name": "Unsaved"}),
|
|
("DELETE", "/manual:saved", None)]:
|
|
with self.subTest(method=method), mock.patch.object(links.os, "replace", side_effect=OSError("private path details")):
|
|
before = self.path.read_bytes()
|
|
response = self.send(method, path, payload)
|
|
self.assertEqual(response.status_code, 500)
|
|
self.assertEqual(response.json, {"error": "Could not save the node list.", "code": "monitor_links_save_failed"})
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
links.refresh.assert_not_called()
|
|
links._probe.assert_not_called()
|
|
|
|
def test_successful_crud_and_detected_hide_reset_keep_existing_contract(self):
|
|
response = self.send("POST", payload={"name": "New", "url": "new.test:8008/path", "product": "pbs"})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(set(response.json), {"success", "key"})
|
|
key = response.json["key"]
|
|
self.assertTrue(key.startswith("manual:"))
|
|
row = links._load()["manual"][-1]
|
|
self.assertEqual((row["url"], row["product"]), ("https://new.test:8008", "pbs"))
|
|
self.assertEqual(self.send("PATCH", "/" + key, {"url": "http://changed.test"}).json, {"success": True})
|
|
self.assertEqual(links._load()["manual"][-1]["url"], "http://changed.test")
|
|
self.assertEqual(self.send("PATCH", "/cluster:node", {"url": "https://override.test", "hidden": True}).json, {"success": True})
|
|
self.assertEqual(self.send("PATCH", "/cluster:node", {"url": "", "hidden": False}).json, {"success": True})
|
|
self.assertEqual(links._load()["overrides"], {"cluster:node": {"hidden": False}})
|
|
self.assertEqual(self.send("DELETE", "/" + key).json, {"success": True})
|
|
self.assertEqual(links._load()["manual"], self.previous["manual"])
|
|
self.assertEqual(self.path.stat().st_mode & 0o777, 0o600)
|
|
self.assertEqual(links.refresh.call_count, 5)
|
|
links._probe.assert_not_called()
|
|
|
|
def test_existing_read_and_admin_decorators_still_guard_routes_before_effects(self):
|
|
with mock.patch.object(links, "cached_links", return_value={"self": {}, "links": []}) as cached, \
|
|
mock.patch.object(links, "_save") as save:
|
|
for method, path in [("POST", ""), ("PATCH", "/manual:saved"), ("DELETE", "/manual:saved")]:
|
|
for role, status in [(None, 401), ("reader", 403)]:
|
|
self.assertEqual(self.send(method, path, {"url": "https://new.test"}, role).status_code, status)
|
|
self.assertEqual(self.send("GET", role=None).status_code, 401)
|
|
self.assertEqual(self.send("GET", "?all=1", role="reader").status_code, 200)
|
|
cached.assert_called_once_with(include_hidden=True)
|
|
save.assert_not_called()
|
|
links.refresh.assert_not_called()
|
|
links._probe.assert_not_called()
|
|
self.assertEqual(json.loads(self.path.read_text()), self.previous)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|