nessi
baa0d24eb4
Add network: prefix support in endpoint_values to resolve network names to IPAM subnet CIDRs for policy matching, extend workload_provider_target to accept vmid: prefix and bare workload names as fallback resolution methods, implement dashboard_top_talkers to aggregate traffic flows by workload with interface_traffic fallback when flows unavailable, add
98 lines
3.2 KiB
Python
98 lines
3.2 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, _: 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 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"]
|
|
assert FakeAsyncClient.put_payloads == [{"enable": 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",
|
|
}
|
|
]
|