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

fix(oci): clean up owned host firewall rules

Browse Source
This commit is contained in:
VAIO73
2026-09-27 21:44:47 +02:00
parent d45da0742d
commit ec391246a1
11 changed files with 109 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oci/tests/test_host_monitor_firewall.py
+43
View File
@@ -4,13 +4,17 @@ from pathlib import Path
import sys
import unittest
from unittest.mock import patch
from subprocess import CompletedProcess
import json
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
sys.path.insert(0, str(ROOT / "remote"))
from proxmenux_oci.installer import (InstallError, confirm_host_monitor_firewall,
host_monitor_firewall_plan)
import oci_remove
class ConfirmUI:
@@ -55,13 +59,52 @@ class HostMonitorFirewallPlanTests(unittest.TestCase):
class HostMonitorFirewallInstallerContractTests(unittest.TestCase):
def test_netdata_declares_its_host_web_port_for_the_firewall_plan(self):
catalog = json.loads((ROOT / "catalog" / "overlays" / "netdata.json").read_text(encoding="utf-8"))
profile = catalog["proxmox"]["installer_profile"]
self.assertEqual(profile["host_monitor"], "netdata")
self.assertEqual(profile["host_monitor_firewall"],
{"protocol": "tcp", "web_port": 19999})
def test_remote_installer_revalidates_and_uses_proxmox_rule_api(self):
source = (ROOT / "remote" / "install_oci.sh").read_text(encoding="utf-8")
self.assertIn("validate_host_monitor_firewall", source)
self.assertIn("The host-monitor firewall subnet changed; no firewall rule was added", source)
self.assertIn('pvesh create "/nodes/${node}/firewall/rules"', source)
self.assertIn("--dport \"$HOST_FIREWALL_PORT\" --source \"$HOST_FIREWALL_SOURCE\"", source)
self.assertIn('--enable 1 --comment "$comment"', source)
self.assertIn("only the original, separately confirmed installation", source)
self.assertIn('comment="ProxMenux OCI firewall ${INSTANCE_ID}"', source)
self.assertIn('if ! pvesh create "/nodes/${node}/firewall/rules"', source)
def test_removal_only_targets_a_uniquely_owned_firewall_rule(self):
source = (ROOT / "remote" / "oci_remove.py").read_text(encoding="utf-8")
self.assertIn("def remove_owned_host_firewall(record):", source)
self.assertIn("ProxMenux OCI firewall {installation_id}", source)
self.assertIn("len(matches) != 1", source)
self.assertIn("firewall/rules/{matches[0]['pos']}", source)
def test_removal_deletes_only_the_exact_rule_owned_by_the_installation(self):
record = {"installation_id": "123e4567-e89b-12d3-a456-426614174000",
"deployment": {"host_firewall": {"source": "192.0.2.0/24", "port": 61208}}}
rules = [{"pos": 7, "comment": "ProxMenux OCI firewall 123e4567-e89b-12d3-a456-426614174000",
"type": "in", "action": "ACCEPT", "proto": "tcp", "source": "192.0.2.0/24",
"dport": "61208"},
{"pos": 8, "comment": "manual rule", "type": "in", "action": "ACCEPT",
"proto": "tcp", "source": "192.0.2.0/24", "dport": "61208"}]
with patch("oci_remove.subprocess.run", side_effect=[
CompletedProcess([], 0, json.dumps(rules), ""), CompletedProcess([], 0, "", "")]) as run:
oci_remove.remove_owned_host_firewall(record)
self.assertIn("/firewall/rules/7", run.call_args_list[1].args[0][-1])
def test_removal_keeps_an_unowned_matching_rule(self):
record = {"installation_id": "123e4567-e89b-12d3-a456-426614174000",
"deployment": {"host_firewall": {"source": "192.0.2.0/24", "port": 61208}}}
rules = [{"pos": 8, "comment": "manual rule", "type": "in", "action": "ACCEPT",
"proto": "tcp", "source": "192.0.2.0/24", "dport": "61208"}]
with patch("oci_remove.subprocess.run", return_value=CompletedProcess([], 0, json.dumps(rules), "")) as run:
oci_remove.remove_owned_host_firewall(record)
run.assert_called_once()
def test_oci_menu_wrapper_does_not_hide_engine_errors_with_the_main_menu(self):
wrapper = (ROOT.parent / "scripts" / "oci" / "oci_manager_apps.sh").read_text(encoding="utf-8")