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"") 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()