docs: document Proxmox write permissions and firewall apply scope, add live apply implementation with rule resolution and provider integration
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
This commit is contained in:
@@ -0,0 +1,86 @@
|
||||
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",
|
||||
}
|
||||
]
|
||||
Reference in New Issue
Block a user