nessi
1ada59d3c7
Add minimum write privileges section covering VM.Audit and VM.Config.Network requirements for firewall orchestration, document NexaFabric comment marker approach for safe rule replacement, clarify that only VM/LXC-level rules with concrete workload targets are supported for live apply while security groups remain preview-only, add firewall interface checkbox requirement for enforcement, document
87 lines
2.7 KiB
Python
87 lines
2.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] = []
|
|
|
|
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, _: str, **__: object) -> FakeResponse:
|
|
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 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 = []
|
|
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.posted_payloads == [
|
|
{
|
|
"type": "in",
|
|
"action": "ACCEPT",
|
|
"enable": 1,
|
|
"proto": "tcp",
|
|
"dport": "443",
|
|
"comment": "NexaFabric policy=policy-1 version=2 rule=1 target=web",
|
|
}
|
|
]
|