nessi
da60155710
Add cluster_firewall_options_url and node_firewall_options_url helpers to build firewall options endpoint paths, implement config_interface_status to parse network interface firewall flags from VM/LXC config with bridge/firewall/raw fields, add firewall_enforcement_status to check enable status across datacenter/node/guest levels and validate interface firewall flags with warning collection for disabled settings, extend apply_rules response with enforcement_
110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
import pytest
|
|
|
|
from app.services.providers import proxmox
|
|
from app.services.providers.base import ProviderConnection
|
|
from app.services.providers.proxmox import ProxmoxProvider
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, data: object) -> None:
|
|
self._data = data
|
|
|
|
def json(self) -> dict:
|
|
return {"data": self._data}
|
|
|
|
def raise_for_status(self) -> None:
|
|
return None
|
|
|
|
|
|
class FakeAsyncClient:
|
|
deleted_urls: list[str] = []
|
|
posted_payloads: list[dict] = []
|
|
put_urls: list[str] = []
|
|
put_payloads: list[dict] = []
|
|
|
|
def __init__(self, **_: object) -> None:
|
|
return None
|
|
|
|
async def __aenter__(self) -> "FakeAsyncClient":
|
|
return self
|
|
|
|
async def __aexit__(self, *_: object) -> None:
|
|
return None
|
|
|
|
async def get(self, url: str, **__: object) -> FakeResponse:
|
|
if url.endswith("/config"):
|
|
return FakeResponse({"net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0,firewall=0", "name": "web"})
|
|
if url.endswith("/cluster/firewall/options"):
|
|
return FakeResponse({"enable": 1})
|
|
if url.endswith("/firewall/options"):
|
|
return FakeResponse({"enable": 1})
|
|
return FakeResponse(
|
|
[
|
|
{"pos": 0, "comment": "manual rule"},
|
|
{"pos": 1, "comment": "NexaFabric policy=policy-1 version=1 rule=1 target=web"},
|
|
]
|
|
)
|
|
|
|
async def delete(self, url: str, **__: object) -> FakeResponse:
|
|
self.deleted_urls.append(url)
|
|
return FakeResponse(None)
|
|
|
|
async def put(self, url: str, data: dict, **__: object) -> FakeResponse:
|
|
self.put_urls.append(url)
|
|
self.put_payloads.append(data)
|
|
return FakeResponse(None)
|
|
|
|
async def post(self, _: str, data: dict, **__: object) -> FakeResponse:
|
|
self.posted_payloads.append(data)
|
|
return FakeResponse({"pos": 1})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_rules_replaces_only_marked_nexafabric_rules(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
FakeAsyncClient.deleted_urls = []
|
|
FakeAsyncClient.posted_payloads = []
|
|
FakeAsyncClient.put_urls = []
|
|
FakeAsyncClient.put_payloads = []
|
|
monkeypatch.setattr(proxmox.httpx, "AsyncClient", FakeAsyncClient)
|
|
|
|
result = await ProxmoxProvider().apply_rules(
|
|
ProviderConnection(api_url="https://pve.example:8006", token="user@pve!token=secret", read_only=False),
|
|
[
|
|
{
|
|
"policy_id": "policy-1",
|
|
"audit_only": False,
|
|
"provider_target": {"node": "pve1", "kind": "qemu", "vmid": "100"},
|
|
"provider_rule": {
|
|
"type": "in",
|
|
"action": "ACCEPT",
|
|
"enable": 1,
|
|
"proto": "tcp",
|
|
"dport": "443",
|
|
"comment": "NexaFabric policy=policy-1 version=2 rule=1 target=web",
|
|
},
|
|
}
|
|
],
|
|
)
|
|
|
|
assert result["applied"] is True
|
|
assert result["rules_deleted"] == 1
|
|
assert FakeAsyncClient.deleted_urls == ["https://pve.example:8006/api2/json/nodes/pve1/qemu/100/firewall/rules/1"]
|
|
assert FakeAsyncClient.put_urls == [
|
|
"https://pve.example:8006/api2/json/nodes/pve1/qemu/100/firewall/options",
|
|
"https://pve.example:8006/api2/json/nodes/pve1/qemu/100/config",
|
|
]
|
|
assert FakeAsyncClient.put_payloads == [
|
|
{"enable": 1},
|
|
{"net0": "virtio=AA:BB:CC:DD:EE:FF,bridge=vmbr0,firewall=1"},
|
|
]
|
|
assert FakeAsyncClient.posted_payloads == [
|
|
{
|
|
"type": "in",
|
|
"action": "ACCEPT",
|
|
"enable": 1,
|
|
"proto": "tcp",
|
|
"dport": "443",
|
|
"comment": "NexaFabric policy=policy-1 version=2 rule=1 target=web",
|
|
}
|
|
]
|