mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 18:32:13 +00:00
MacRimi
98ae1f3626
- Open the post-install menu from the PBS menu - List only the customizable options that apply to PBS - Offer the PBS no-subscription repository through the native repository API - Upgrade PBS with the package list shown before applying it - Remove and restore the subscription notice on the PBS web interface - Install Figurine on PBS without the Proxmox VE aliases - Offer the Helper-Scripts PBS post-install and microcode scripts - Tune firewall bridges after the systemd udev rules in the automated post-install - Add PBS repository policy tests and syntax checks - Document the PBS differences on the post-install page - Use backups in the Spanish PBS update message
151 lines
7.6 KiB
Python
151 lines
7.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
sys.path.insert(0, str(ROOT / "AppImage/scripts"))
|
|
|
|
import monitor_links as links
|
|
|
|
|
|
class MonitorLinksTests(unittest.TestCase):
|
|
def setUp(self):
|
|
temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(temporary.cleanup)
|
|
self.root = Path(temporary.name)
|
|
storage = self.root / "storage.cfg"
|
|
storage.write_text("dir: local\n\tpath /var/lib/vz\n\npbs: Lab\n\tdatastore store1\n\tserver 10.0.0.49\n\n"
|
|
"pbs: Cloud\n\tdatastore x\n\tserver pbs.example.com\n\tport 8443\n")
|
|
patches = [
|
|
mock.patch.object(links, "LINKS_FILE", self.root / "links.json"),
|
|
mock.patch.object(links, "SSL_CONFIG_FILE", self.root / "ssl.json"),
|
|
mock.patch.object(links, "STORAGE_CFG", storage),
|
|
mock.patch.object(links.pbs_manager, "get_platform", return_value={"product": "pve", "hostname": "amd"}),
|
|
mock.patch.object(links.pbs_manager, "public_servers", return_value=[]),
|
|
]
|
|
for patch in patches:
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
links._probe_cache.clear()
|
|
cluster = [{"type": "cluster", "name": "Casa"},
|
|
{"type": "node", "name": "amd", "ip": "10.0.0.55", "local": 1},
|
|
{"type": "node", "name": "constructor", "ip": "10.0.0.50", "local": 0}]
|
|
self.run_patch = mock.patch.object(links.subprocess, "run",
|
|
return_value=SimpleNamespace(returncode=0, stdout=json.dumps(cluster)))
|
|
self.run_patch.start()
|
|
self.addCleanup(self.run_patch.stop)
|
|
|
|
def test_normalize_url_keeps_only_the_origin_and_rejects_credentials(self):
|
|
self.assertEqual(links.normalize_url("192.168.0.49:8008/dashboard?x=1"), "https://192.168.0.49:8008")
|
|
self.assertEqual(links.normalize_url("http://pve.lan"), "http://pve.lan")
|
|
for bad in ("", "ftp://host", "https://user:pass@host", "https://host:99999", "javascript:alert(1)"):
|
|
with self.subTest(bad=bad), self.assertRaises(links.LinkError):
|
|
links.normalize_url(bad)
|
|
|
|
def test_discovers_other_cluster_nodes_and_pbs_without_the_local_node(self):
|
|
probed = []
|
|
with mock.patch.object(links, "_probe", side_effect=lambda url: probed.append(url) or url.endswith("10.0.0.50:8008")), \
|
|
mock.patch.object(links, "_is_private", side_effect=lambda host: host.startswith("10.")):
|
|
result = links.list_links()
|
|
by_key = {row["key"]: row for row in result["links"]}
|
|
self.assertEqual(result["self"]["name"], "amd")
|
|
self.assertNotIn("cluster:amd", by_key)
|
|
self.assertEqual(by_key["cluster:constructor"]["status"], "online")
|
|
self.assertEqual(by_key["pbs:10.0.0.49"]["native_url"], "https://10.0.0.49:8007")
|
|
self.assertEqual(by_key["pbs:10.0.0.49"]["status"], "offline")
|
|
# A public PBS is listed with its own interface but never probed unasked.
|
|
self.assertEqual(by_key["pbs:pbs.example.com"]["status"], "unknown")
|
|
self.assertEqual(by_key["pbs:pbs.example.com"]["native_url"], "https://pbs.example.com:8443")
|
|
self.assertFalse(any("example.com" in url for url in probed))
|
|
|
|
def test_pbs_host_lists_only_manual_entries(self):
|
|
links.pbs_manager.get_platform.return_value = {"product": "pbs", "hostname": "pbs-lab"}
|
|
links.add_link({"name": "amd", "url": "https://10.0.0.55:8008", "product": "pve"})
|
|
with mock.patch.object(links, "_probe", return_value=True):
|
|
result = links.list_links()
|
|
self.assertEqual([row["name"] for row in result["links"]], ["amd"])
|
|
self.links_run_not_called()
|
|
|
|
def links_run_not_called(self):
|
|
self.assertFalse(any(call.args and call.args[0][:1] == ["pvesh"] for call in links.subprocess.run.call_args_list))
|
|
|
|
def test_manual_entries_overrides_and_hidden_detected_nodes(self):
|
|
key = links.add_link({"name": "lab", "url": "10.0.0.46:8008", "product": "pve"})
|
|
with self.assertRaises(links.LinkError):
|
|
links.add_link({"url": "https://10.0.0.46:8008"})
|
|
links.update_link("cluster:constructor", {"url": "https://constructor.lan:8008"})
|
|
links.update_link("pbs:10.0.0.49", {"hidden": True})
|
|
with mock.patch.object(links, "_probe", return_value=False), \
|
|
mock.patch.object(links, "_is_private", return_value=True):
|
|
visible = {row["key"]: row for row in links.list_links()["links"]}
|
|
everything = {row["key"]: row for row in links.list_links(include_hidden=True)["links"]}
|
|
self.assertEqual(visible[key]["url"], "https://10.0.0.46:8008")
|
|
self.assertEqual(visible["cluster:constructor"]["url"], "https://constructor.lan:8008")
|
|
self.assertNotIn("pbs:10.0.0.49", visible)
|
|
self.assertTrue(everything["pbs:10.0.0.49"]["hidden"])
|
|
with self.assertRaises(links.LinkError):
|
|
links.delete_link("cluster:constructor")
|
|
links.delete_link(key)
|
|
self.assertEqual(json.loads((self.root / "links.json").read_text())["manual"], [])
|
|
self.assertEqual((self.root / "links.json").stat().st_mode & 0o777, 0o600)
|
|
|
|
def test_cached_list_is_served_without_new_probes_and_renewed_when_stale(self):
|
|
links._snapshot.update(at=0.0, data=None)
|
|
self.addCleanup(links._snapshot.update, at=0.0, data=None)
|
|
links.update_link("pbs:10.0.0.49", {"hidden": True})
|
|
with mock.patch.object(links, "_probe", return_value=True) as probe, \
|
|
mock.patch.object(links, "_is_private", return_value=True):
|
|
first = links.cached_links()
|
|
calls = probe.call_count
|
|
again = links.cached_links()
|
|
self.assertEqual(probe.call_count, calls)
|
|
self.assertEqual(first, again)
|
|
self.assertNotIn("pbs:10.0.0.49", [row["key"] for row in first["links"]])
|
|
self.assertIn("pbs:10.0.0.49", [row["key"] for row in links.cached_links(include_hidden=True)["links"]])
|
|
links._snapshot["at"] -= links.PROBE_TTL + 1
|
|
with mock.patch.object(links.threading, "Thread") as thread:
|
|
stale = links.cached_links()
|
|
self.assertEqual(stale, first)
|
|
thread.assert_called_once()
|
|
|
|
def test_unknown_keys_are_rejected(self):
|
|
for key in ("other:x", "cluster:../../etc", "manual:nope"):
|
|
with self.subTest(key=key), self.assertRaises(links.LinkError):
|
|
links.update_link(key, {"hidden": True})
|
|
|
|
def test_probe_sends_no_credentials_and_needs_a_monitor_answer(self):
|
|
seen = []
|
|
|
|
class Response:
|
|
status = 200
|
|
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def read(self, _size):
|
|
return self.body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
return False
|
|
|
|
def urlopen(request, timeout, context):
|
|
seen.append(request)
|
|
return Response(b'{"auth_enabled": true}' if "monitor" in request.full_url else b"<html>")
|
|
|
|
with mock.patch.object(links.urllib.request, "urlopen", side_effect=urlopen):
|
|
self.assertTrue(links._probe("https://monitor:8008"))
|
|
self.assertFalse(links._probe("https://other:8008"))
|
|
self.assertTrue(all(request.full_url.endswith("/api/auth/status") for request in seen))
|
|
self.assertTrue(all("Authorization" not in request.headers for request in seen))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|