Sergey Antropoff
0773f721ea
- Scale small/large/big seeds (3×50 / 10×1000 / 20×2000) with proportional backups, snapshots, HA, replication, Ceph capacity, and OSD totals (10 / 100 / 500) plus matching node disks and crush/pg metadata - Enrich handler responses for apt, certificates, qemu/lxc status, storage, SDN, metrics export, and related cluster/node dumps - Flatten nested body_example fields into PARAMS and sync the request body via dotted paths (oVirt-style) - Restyle DATA controls as size cards with full-width Reset to minimal / Refresh stats; unload reloads the minimal cluster
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""Deterministic seed profile tests."""
|
|
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
from app.simulation.seed import (
|
|
big_profile,
|
|
build_profile,
|
|
clear_simulation_state,
|
|
cluster_domain_metadata,
|
|
default_node_ops_for_seed,
|
|
enrich_guest_state,
|
|
enrich_storage_state,
|
|
lab_profile,
|
|
large_profile,
|
|
small_profile,
|
|
stable_id,
|
|
)
|
|
|
|
|
|
def test_lab_profile_matches_required_logical_shape() -> None:
|
|
first = lab_profile()
|
|
second = lab_profile()
|
|
|
|
assert first == second
|
|
state = first.logical_state()
|
|
assert state == second.logical_state()
|
|
assert state["nodes"] == [{"name": "pve01", "status": "online"}]
|
|
resources = state["resources"]
|
|
assert isinstance(resources, list)
|
|
assert [resource["kind"] for resource in resources].count("qemu") == 2
|
|
assert [resource["kind"] for resource in resources].count("lxc") == 1
|
|
assert [resource["kind"] for resource in resources].count("storage") == 3
|
|
assert [resource["kind"] for resource in resources].count("ceph-osd") == 3
|
|
assert [resource["kind"] for resource in resources].count("ha") == 1
|
|
tasks = state["tasks"]
|
|
assert isinstance(tasks, list)
|
|
assert len(tasks) == 2
|
|
|
|
|
|
def test_sized_cluster_profiles() -> None:
|
|
small = small_profile()
|
|
assert len(small.nodes) == 3
|
|
assert sum(resource.kind in {"qemu", "lxc"} for resource in small.resources) == 50
|
|
assert sum(resource.kind == "ceph-osd" for resource in small.resources) == 10
|
|
assert sum(resource.kind == "ha" for resource in small.resources) == 3
|
|
assert len(small.tasks) == 12
|
|
|
|
large = large_profile()
|
|
assert len(large.nodes) == 10
|
|
assert sum(resource.kind in {"qemu", "lxc"} for resource in large.resources) == 1_000
|
|
assert sum(resource.kind == "ceph-osd" for resource in large.resources) == 100
|
|
assert sum(resource.kind == "pool" for resource in large.resources) == 2
|
|
assert len(large.tasks) == 50
|
|
|
|
big = big_profile()
|
|
assert len(big.nodes) == 20
|
|
assert sum(resource.kind in {"qemu", "lxc"} for resource in big.resources) == 2_000
|
|
assert sum(resource.kind == "ceph-osd" for resource in big.resources) == 500
|
|
assert sum(resource.kind == "pool" for resource in big.resources) == 3
|
|
assert len(big.tasks) == 100
|
|
assert build_profile("big") == big
|
|
|
|
|
|
def test_sized_profiles_spread_osds_across_nodes() -> None:
|
|
for profile_name, expected in (("small", 10), ("large", 100), ("big", 500)):
|
|
profile = build_profile(profile_name)
|
|
names = {node.id: node.name for node in profile.nodes}
|
|
counts: dict[str, int] = {name: 0 for name in names.values()}
|
|
for resource in profile.resources:
|
|
if resource.kind == "ceph-osd":
|
|
counts[names[resource.node_id]] += 1
|
|
assert sum(counts.values()) == expected
|
|
assert max(counts.values()) - min(counts.values()) <= 1
|
|
|
|
|
|
def test_medium_and_fault_profiles_are_deterministic() -> None:
|
|
medium = build_profile("medium")
|
|
assert len(medium.nodes) == 3
|
|
assert sum(resource.kind in {"qemu", "lxc"} for resource in medium.resources) == 50
|
|
assert build_profile("ha-demo") == build_profile("ha-demo")
|
|
broken = build_profile("broken-storage")
|
|
assert any(resource.state.get("status") == "offline" for resource in broken.resources)
|
|
|
|
|
|
def test_large_profile_is_configurable_and_stable() -> None:
|
|
first = large_profile(node_count=4, resource_count=1_000)
|
|
second = large_profile(node_count=4, resource_count=1_000)
|
|
assert first == second
|
|
assert len(first.nodes) == 4
|
|
guests = [r for r in first.resources if r.kind in {"qemu", "lxc"}]
|
|
assert len(guests) == 1_000
|
|
assert any(r.kind == "storage" and r.external_id == "local" for r in first.resources)
|
|
assert any(r.kind == "ceph-osd" for r in first.resources)
|
|
|
|
|
|
def test_profile_validation() -> None:
|
|
with pytest.raises(ValueError, match="unknown seed profile"):
|
|
build_profile("missing")
|
|
with pytest.raises(ValueError, match="positive"):
|
|
large_profile(node_count=0, resource_count=1)
|
|
|
|
|
|
def test_demo_cluster_profile_shape() -> None:
|
|
profile = build_profile("demo-cluster")
|
|
assert profile.name == "demo-cluster"
|
|
assert len(profile.nodes) == 20
|
|
assert sum(resource.kind == "qemu" for resource in profile.resources) == 850
|
|
assert sum(resource.kind == "lxc" for resource in profile.resources) == 150
|
|
assert sum(resource.kind == "ceph-osd" for resource in profile.resources) == 300
|
|
assert sum(resource.kind == "storage" for resource in profile.resources) >= 62
|
|
assert len(profile.tasks) == 250
|
|
external_ids = {
|
|
resource.external_id for resource in profile.resources if resource.kind in {"qemu", "lxc"}
|
|
}
|
|
assert len(external_ids) == 1000
|
|
|
|
|
|
def test_demo_cluster_spreads_guests_evenly_across_nodes() -> None:
|
|
profile = build_profile("demo-cluster")
|
|
names = {node.id: node.name for node in profile.nodes}
|
|
|
|
def counts(kind: str) -> list[int]:
|
|
counter: dict[str, int] = {name: 0 for name in names.values()}
|
|
for resource in profile.resources:
|
|
if resource.kind == kind:
|
|
counter[names[resource.node_id]] += 1
|
|
return list(counter.values())
|
|
|
|
for kind, expected_total in (("qemu", 850), ("lxc", 150), ("ceph-osd", 300)):
|
|
values = counts(kind)
|
|
assert sum(values) == expected_total
|
|
assert max(values) - min(values) <= 1
|
|
|
|
guest_counts = counts("qemu")
|
|
guest_counts = [a + b for a, b in zip(guest_counts, counts("lxc"), strict=True)]
|
|
assert max(guest_counts) - min(guest_counts) <= 2
|
|
|
|
|
|
def test_minimal_profile() -> None:
|
|
profile = build_profile("minimal")
|
|
assert len(profile.nodes) == 1
|
|
assert not any(resource.kind in {"qemu", "lxc"} for resource in profile.resources)
|
|
|
|
|
|
def test_stable_ids_are_namespaced_and_repeatable() -> None:
|
|
assert stable_id("qemu:100") == stable_id("qemu:100")
|
|
assert stable_id("qemu:100") != stable_id("qemu:101")
|
|
|
|
|
|
def test_cluster_domain_metadata_seeds_list_domains() -> None:
|
|
meta = cast(dict[str, Any], cluster_domain_metadata(lab_profile()))
|
|
assert meta["firewall"]["scopes"]["cluster"]["rules"]
|
|
assert meta["firewall"]["scopes"]["cluster"]["aliases"]
|
|
assert meta["firewall"]["scopes"]["cluster"]["ipset"]
|
|
assert meta["firewall"]["scopes"]["cluster"]["groups"]
|
|
assert meta["firewall"]["scopes"]["node:pve01"]["rules"]
|
|
assert meta["firewall"]["scopes"]["qemu:pve01:100"]["rules"]
|
|
assert meta["firewall"]["scopes"]["lxc:pve01:200"]["rules"]
|
|
assert meta["firewall_macros"]
|
|
assert meta["sdn"]["zones"]
|
|
assert meta["sdn"]["vnets"]
|
|
assert meta["sdn"]["controllers"]
|
|
assert meta["sdn"]["ipams"]
|
|
assert meta["sdn"]["dns"]
|
|
assert meta["notifications"]["endpoints"]["smtp"]
|
|
assert meta["notifications"]["endpoints"]["gotify"]
|
|
assert meta["notifications"]["matchers"]
|
|
assert meta["notifications"]["matcher_fields"]
|
|
assert meta["acme"]["accounts"]
|
|
assert meta["acme"]["accounts"]["letsencrypt"]["tos_url"]
|
|
assert meta["acme"]["plugins"]
|
|
assert meta["acme"]["directories"]
|
|
assert meta["acme"]["challenge_schema"]
|
|
assert meta["mapping"]["pci"]
|
|
assert meta["mapping"]["usb"]
|
|
assert meta["mapping"]["dir"]
|
|
# Single-node lab must not invent ghost peers.
|
|
assert meta["replication"] == []
|
|
assert "pve02" not in str(meta["sdn"]["controllers"])
|
|
assert meta["metrics"]["servers"]
|
|
assert meta["ha_groups"]
|
|
assert meta["ceph"]["pools"]
|
|
assert meta["ceph"]["flags"]
|
|
assert "noup" in meta["ceph"]["flags"]
|
|
assert meta["ceph"]["flags"]["nobackfill"] == 0
|
|
assert meta["ceph"]["health"]["status"] == "HEALTH_OK"
|
|
assert meta["ceph"]["version"]
|
|
assert len([r for r in lab_profile().resources if r.kind == "ceph-osd"]) == 3
|
|
assert any(r.kind == "storage" and r.external_id == "ceph" for r in lab_profile().resources)
|
|
local = next(r for r in lab_profile().resources if r.external_id == "local")
|
|
assert int(cast(int, local.state["total_bytes"])) > 0
|
|
assert lab_profile().tasks[0].payload["node"] == "pve01"
|
|
assert meta["qemu_cpu_flags"]
|
|
assert meta["metrics"]["export_data"]
|
|
assert meta["jobs"]["schedule_analyze_results"]
|
|
ops = cast(dict[str, Any], default_node_ops_for_seed("pve01"))
|
|
assert ops["vzdump"]["defaults"]["storage"] == "local"
|
|
assert ops["status"]["mem"] > 0
|
|
assert ops["status"]["maxmem"] > 0
|
|
assert ops["capabilities"]["cpu"]
|
|
assert ops["capabilities"]["machines"]
|
|
assert ops["hosts"]["data"]
|
|
assert ops["journal"]
|
|
assert ops["syslog"]
|
|
assert ops["netstat"]
|
|
assert ops["rrddata"]
|
|
assert ops["status"]["uptime"]
|
|
assert ops["ip"]
|
|
assert meta["cluster_config"]["totem"]
|
|
assert meta["cluster_config"]["config_digest"]
|
|
assert meta["cluster_config"]["corosync_conf"]
|
|
assert meta["cluster_config"]["corosync_authkey"]
|
|
first_node = next(iter(meta["cluster_config"]["added_nodes"].values()))
|
|
assert first_node["pve_fp"].count(":") == 31
|
|
assert first_node["pve_addr"]
|
|
assert meta["backup_jobs"]["backup-daily"]["storage"] == "local"
|
|
assert meta["quorate"] == 1
|
|
guest = cast(
|
|
dict[str, Any],
|
|
enrich_guest_state({"name": "demo", "status": "stopped"}, kind="qemu", vmid="100"),
|
|
)
|
|
assert guest["scsi0"].startswith("local-lvm:")
|
|
assert guest["ostype"] == "l26"
|
|
assert str(guest["net0"]).startswith("virtio=")
|
|
assert guest["agent"]["results"]["info"]
|
|
assert guest["rrddata"]
|
|
assert guest["migrate_preconditions"]
|
|
storage = cast(
|
|
dict[str, Any],
|
|
enrich_storage_state({"total_bytes": 1000, "used_bytes": 100}, storage_id="local"),
|
|
)
|
|
assert storage["rrddata"]
|
|
assert storage["file_restore"]
|
|
assert storage["import_metadata"]
|
|
|
|
small_meta = cast(dict[str, Any], cluster_domain_metadata(small_profile()))
|
|
assert len(small_meta["replication"]) == 2
|
|
assert small_meta["replication"][0]["guest"] == 101
|
|
assert len(small_meta["backup_jobs"]) == 2
|
|
assert len(small_meta["ha_groups"]) == 2
|
|
assert small_meta["sdn"]["zones"]["public"]["digest"]
|
|
assert small_meta["sdn"]["zones"]["public"]["state"] == "available"
|
|
assert len(cast(dict[str, Any], cluster_domain_metadata(large_profile()))["replication"]) == 8
|
|
assert len(cast(dict[str, Any], cluster_domain_metadata(large_profile()))["ha_groups"]) == 4
|
|
assert len(cast(dict[str, Any], cluster_domain_metadata(big_profile()))["replication"]) == 16
|
|
assert len(cast(dict[str, Any], cluster_domain_metadata(big_profile()))["ha_groups"]) == 6
|
|
assert len(cast(dict[str, Any], cluster_domain_metadata(big_profile()))["ceph"]["pools"]) == 3
|
|
|
|
ops = cast(dict[str, Any], default_node_ops_for_seed("pve1", node_index=1, node_count=3))
|
|
apt_repos = cast(dict[str, Any], cast(dict[str, Any], ops["apt"])["repositories"])
|
|
assert "files" in apt_repos and "standard-repos" in apt_repos
|
|
cert = cast(list[dict[str, Any]], cast(dict[str, Any], ops["certificates"])["info"])[0]
|
|
assert cert["pem"] and cert["san"] and cert["public-key-bits"] == 4096
|
|
|
|
|
|
def test_sized_profile_artifact_targets() -> None:
|
|
from app.simulation.seed import _CLUSTER_SIZE_SPECS
|
|
|
|
assert _CLUSTER_SIZE_SPECS["small"]["backups"] == 10
|
|
assert _CLUSTER_SIZE_SPECS["small"]["snapshots"] == 12
|
|
assert _CLUSTER_SIZE_SPECS["large"]["backups"] == 80
|
|
assert _CLUSTER_SIZE_SPECS["large"]["snapshots"] == 100
|
|
assert _CLUSTER_SIZE_SPECS["big"]["backups"] == 160
|
|
assert _CLUSTER_SIZE_SPECS["big"]["snapshots"] == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_simulation_state_wipes_api_created_identity() -> None:
|
|
executed: list[str] = []
|
|
|
|
class FakeConnection:
|
|
async def execute(self, sql: str, *args: object) -> str:
|
|
del args
|
|
executed.append(" ".join(sql.split()))
|
|
return "DELETE 0"
|
|
|
|
await clear_simulation_state(FakeConnection())
|
|
joined = "\n".join(executed)
|
|
for table in (
|
|
"resources",
|
|
"nodes",
|
|
"principals",
|
|
"identity_groups",
|
|
"roles",
|
|
"storage_contents",
|
|
"api_tokens",
|
|
):
|
|
assert f"DELETE FROM {table}" in joined # noqa: S608 - asserting SQL text
|
|
assert "DELETE FROM realms WHERE name NOT IN" in joined
|
|
assert any(sql.startswith("UPDATE clusters") for sql in executed)
|