feat: complete group ACL permission matrix

This commit is contained in:
Sergey Antropoff
2026-07-13 01:40:49 +03:00
parent 5934c567b5
commit 7721ba87c5
8 changed files with 133 additions and 10 deletions
+26 -3
View File
@@ -44,6 +44,8 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None:
verify_ssl=False,
)
assert readonly_api.nodes.get()
assert readonly_api.nodes("pve1").status.get()["status"] == "online"
assert readonly_api.nodes("pve1").qemu("101").config.get()["vmid"] == 101
with pytest.raises(ResourceException) as denied:
readonly_api.nodes("pve1").qemu("101").status.start.post()
assert denied.value.status_code == 403
@@ -68,13 +70,34 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None:
ephemeral.nodes.get()
assert removed.value.status_code == 401
storage_api = ProxmoxAPI(
os.environ["PROXMOXER_HOST"],
port=int(os.getenv("PROXMOXER_PORT", "8007")),
user="storage@pve",
token_name=os.getenv("PROXMOXER_STORAGE_TOKEN_NAME", "storage"),
token_value=os.getenv("PROXMOXER_STORAGE_TOKEN_SECRET", "storage-secret"),
verify_ssl=False,
)
for vmid in ("101", "999999"):
with pytest.raises(ResourceException) as hidden:
storage_api.nodes("pve1").qemu(vmid).config.get()
assert hidden.value.status_code == 403
if os.getenv("PROXMOXER_MUTATION_TEST") == "1":
status = proxmox.nodes("pve1").qemu("101").status.current.get()
operator_api = ProxmoxAPI(
os.environ["PROXMOXER_HOST"],
port=int(os.getenv("PROXMOXER_PORT", "8007")),
user="operator@pve",
token_name=os.getenv("PROXMOXER_OPERATOR_TOKEN_NAME", "operator"),
token_value=os.getenv("PROXMOXER_OPERATOR_TOKEN_SECRET", "operator-secret"),
verify_ssl=False,
)
status = operator_api.nodes("pve1").qemu("101").status.current.get()
operation = "start" if status["status"] == "stopped" else "stop"
endpoint = proxmox.nodes("pve1").qemu("101").status(operation)
endpoint = operator_api.nodes("pve1").qemu("101").status(operation)
upid = endpoint.post()
for _attempt in range(100):
task = proxmox.nodes("pve1").tasks(upid).status.get()
task = operator_api.nodes("pve1").tasks(upid).status.get()
if task["status"] == "stopped":
break
Event().wait(0.05)