60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
import importlib.util
|
|
import pathlib
|
|
|
|
_PATH = pathlib.Path(__file__).resolve().parent.parent / "scripts" / "capacity-scan.py"
|
|
_spec = importlib.util.spec_from_file_location("capacity_scan", _PATH)
|
|
cs = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(cs)
|
|
|
|
|
|
def test_parse_table_keys_on_header_and_ignores_extra_cols():
|
|
md = """
|
|
intro text
|
|
| node | cores | ram_gb | disk_gb | notes |
|
|
|------|-------|--------|---------|-------|
|
|
| pve0 | 20 | 64 | 4000 | nvme |
|
|
| pve1 | 20 | 64 | 4000 | nvme |
|
|
|
|
trailing text
|
|
"""
|
|
rows = cs.parse_table(md, ["node", "cores", "ram_gb", "disk_gb"])
|
|
assert rows == [
|
|
{"node": "pve0", "cores": "20", "ram_gb": "64", "disk_gb": "4000", "notes": "nvme"},
|
|
{"node": "pve1", "cores": "20", "ram_gb": "64", "disk_gb": "4000", "notes": "nvme"},
|
|
]
|
|
|
|
|
|
def test_parse_table_returns_empty_when_header_absent():
|
|
assert cs.parse_table("no tables here", ["node", "cores"]) == []
|
|
|
|
|
|
def test_compute_rollup_sums_allocations_and_flags_headroom():
|
|
node_rows = [{"node": "pve0", "cores": "20", "ram_gb": "64", "disk_gb": "4000"}]
|
|
workload_rows = [
|
|
{"workload": "dns1", "node": "pve0", "cores": "1", "ram_mb": "512", "disk_gb": "10"},
|
|
{"workload": "forgejo", "node": "pve0", "cores": "4", "ram_mb": "8192", "disk_gb": "100"},
|
|
]
|
|
nodes = cs.compute_rollup(node_rows, workload_rows)
|
|
pve0 = nodes["pve0"]
|
|
assert pve0["alloc_cores"] == 5
|
|
assert pve0["alloc_ram_gb"] == 8.5 # (512 + 8192) / 1024
|
|
assert pve0["alloc_disk_gb"] == 110.0
|
|
assert pve0["ram_headroom_pct"] == 87 # round(100 * (64 - 8.5) / 64)
|
|
assert pve0["oversubscribed"] is False
|
|
|
|
|
|
def test_compute_rollup_flags_oversubscription():
|
|
node_rows = [{"node": "tiny", "cores": "2", "ram_gb": "4", "disk_gb": "50"}]
|
|
workload_rows = [
|
|
{"workload": "hog", "node": "tiny", "cores": "4", "ram_mb": "1024", "disk_gb": "10"},
|
|
]
|
|
nodes = cs.compute_rollup(node_rows, workload_rows)
|
|
assert nodes["tiny"]["oversubscribed"] is True # 4 cores > 2
|
|
|
|
|
|
def test_compute_rollup_ignores_workloads_on_unknown_nodes():
|
|
nodes = cs.compute_rollup(
|
|
[{"node": "pve0", "cores": "20", "ram_gb": "64", "disk_gb": "4000"}],
|
|
[{"workload": "ghost", "node": "nope", "cores": "1", "ram_mb": "512", "disk_gb": "10"}],
|
|
)
|
|
assert nodes["pve0"]["alloc_cores"] == 0
|