Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-01 10:22:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

i18n: refine Slovak linked nodes and localize validation errors

Browse Source
This commit is contained in:
VAIO73
2026-10-20 06:46:15 +00:00
parent 65f47ec646
commit 9ffd154a2a
9 changed files with 379 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AppImage/scripts/flask_monitor_links_routes.py
+13 -6
View File
@@ -7,6 +7,13 @@ from jwt_middleware import require_auth, require_admin_scope
monitor_links_bp = Blueprint("monitor_links", __name__)
def link_error(exc):
payload = {"error": str(exc), "code": exc.code}
if exc.params:
payload["params"] = exc.params
return jsonify(payload), 400
@monitor_links_bp.after_request
def no_cache(response):
response.headers["Cache-Control"] = "no-store"
@@ -27,9 +34,9 @@ def add_link():
monitor_links.refresh()
return jsonify(success=True, key=key)
except monitor_links.LinkError as exc:
return jsonify(error=str(exc)), 400
return link_error(exc)
except OSError:
return jsonify(error="Could not save the node list."), 500
return jsonify(error="Could not save the node list.", code="monitor_links_save_failed"), 500
@monitor_links_bp.route("/api/monitor-links/<path:key>", methods=["PATCH"])
@@ -40,9 +47,9 @@ def update_link(key):
monitor_links.refresh()
return jsonify(success=True)
except monitor_links.LinkError as exc:
return jsonify(error=str(exc)), 400
return link_error(exc)
except OSError:
return jsonify(error="Could not save the node list."), 500
return jsonify(error="Could not save the node list.", code="monitor_links_save_failed"), 500
@monitor_links_bp.route("/api/monitor-links/<path:key>", methods=["DELETE"])
@@ -53,6 +60,6 @@ def delete_link(key):
monitor_links.refresh()
return jsonify(success=True)
except monitor_links.LinkError as exc:
return jsonify(error=str(exc)), 400
return link_error(exc)
except OSError:
return jsonify(error="Could not save the node list."), 500
return jsonify(error="Could not save the node list.", code="monitor_links_save_failed"), 500
AppImage/scripts/monitor_links.py
+15 -12
View File
@@ -39,7 +39,10 @@ _refresh_lock = threading.Lock()
class LinkError(ValueError):
pass
def __init__(self, message, *, code="monitor_links_save_failed", params=None):
super().__init__(message)
self.code = code
self.params = params
def _load():
@@ -63,19 +66,19 @@ def _save(data):
def normalize_url(value):
"""Return the origin of an http(s) address, or raise LinkError."""
if not isinstance(value, str) or not value.strip():
raise LinkError("Enter the address of the Monitor.")
raise LinkError("Enter the address of the Monitor.", code="monitor_link_address_required")
text = value.strip()
if "://" not in text:
text = "https://" + text
parsed = urllib.parse.urlsplit(text)
if parsed.scheme not in ("http", "https") or not parsed.hostname:
raise LinkError("The address must start with http:// or https:// and name a host.")
raise LinkError("The address must start with http:// or https:// and name a host.", code="monitor_link_invalid_address")
if parsed.username or parsed.password:
raise LinkError("The address must not contain a user or password.")
raise LinkError("The address must not contain a user or password.", code="monitor_link_address_credentials")
try:
port = parsed.port
except ValueError as exc:
raise LinkError("The port of the address is not valid.") from exc
raise LinkError("The port of the address is not valid.", code="monitor_link_invalid_port") from exc
host = parsed.hostname
if ":" in host:
host = f"[{host}]"
@@ -301,16 +304,16 @@ def start_prewarmer():
def add_link(payload):
if not isinstance(payload, dict):
raise LinkError("Invalid request.")
raise LinkError("Invalid request.", code="monitor_link_invalid_request")
name = str(payload.get("name") or "").strip()[:64]
url = normalize_url(payload.get("url"))
product = payload.get("product") if payload.get("product") in PRODUCTS else "pve"
with _lock:
data = _load()
if len(data["manual"]) >= MAX_MANUAL:
raise LinkError(f"At most {MAX_MANUAL} nodes can be added.")
raise LinkError(f"At most {MAX_MANUAL} nodes can be added.", code="monitor_link_limit", params={"count": MAX_MANUAL})
if any(row.get("url") == url for row in data["manual"]):
raise LinkError("This address is already in the list.")
raise LinkError("This address is already in the list.", code="monitor_link_duplicate")
identifier = secrets.token_hex(6)
data["manual"].append({"id": identifier, "name": name or urllib.parse.urlsplit(url).hostname,
"url": url, "product": product})
@@ -320,13 +323,13 @@ def add_link(payload):
def update_link(key, payload):
if not isinstance(payload, dict):
raise LinkError("Invalid request.")
raise LinkError("Invalid request.", code="monitor_link_invalid_request")
with _lock:
data = _load()
if key.startswith("manual:"):
row = next((r for r in data["manual"] if f"manual:{r.get('id')}" == key), None)
if row is None:
raise LinkError("Node not found.")
raise LinkError("Node not found.", code="monitor_link_not_found")
if "name" in payload:
row["name"] = str(payload["name"] or "").strip()[:64] or row["name"]
if payload.get("url"):
@@ -345,7 +348,7 @@ def update_link(key, payload):
if not override:
data["overrides"].pop(key, None)
else:
raise LinkError("Node not found.")
raise LinkError("Node not found.", code="monitor_link_not_found")
_save(data)
@@ -354,6 +357,6 @@ def delete_link(key):
data = _load()
remaining = [r for r in data["manual"] if f"manual:{r.get('id')}" != key]
if len(remaining) == len(data["manual"]):
raise LinkError("Only nodes added by hand can be removed; detected ones can be hidden.")
raise LinkError("Only nodes added by hand can be removed; detected ones can be hidden.", code="monitor_link_detected_remove")
data["manual"] = remaining
_save(data)
AppImage/scripts/tests/test_monitor_links_i18n.py
+138
View File
@@ -0,0 +1,138 @@
"""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()