railiance-cluster/tests/test_cluster_resources.py
codex ea0452b9a9
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Publish cluster resource evidence and gate ThreePhoenix
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02669-87ee-7a31-b111-edc95a16e0fa
2026-08-22 12:34:25 +02:00

38 lines
2.3 KiB
Python

from __future__ import annotations
import importlib.util
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("cluster_resources", ROOT / "tools" / "observe_cluster_resources.py")
assert SPEC and SPEC.loader
module = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(module)
class ClusterResourceTests(unittest.TestCase):
def test_quantity_conversion(self) -> None:
self.assertEqual(250, module.cpu_millicores("250m"))
self.assertEqual(2000, module.cpu_millicores("2"))
self.assertEqual(1024**3, module.bytes_value("1Gi"))
def test_aggregation_and_missing_metrics(self) -> None:
snapshots = {
"nodes": {"items": [{"metadata": {"name": "node-1", "labels": {"kubernetes.io/hostname": "node-1"}}, "spec": {}, "status": {"allocatable": {"cpu": "2", "memory": "4Gi", "ephemeral-storage": "20Gi", "pods": "110"}, "conditions": [{"type": "Ready", "status": "True"}]}}]},
"pods": {"items": [{"metadata": {"name": "api-abc123", "namespace": "apps", "labels": {"app.kubernetes.io/name": "api"}}, "spec": {"containers": [{"resources": {"requests": {"cpu": "250m", "memory": "128Mi"}, "limits": {"cpu": "1", "memory": "512Mi"}}}]}}]},
"pvcs": {"items": [{"metadata": {"name": "data", "namespace": "apps"}, "spec": {"storageClassName": "local-path", "resources": {"requests": {"storage": "1Gi"}}}, "status": {"phase": "Bound"}}]},
"pvs": {"items": [{"spec": {"capacity": {"storage": "2Gi"}}}]},
"storageclasses": {"items": [{"metadata": {"name": "local-path", "annotations": {"storageclass.kubernetes.io/is-default-class": "true"}}, "provisioner": "local-path", "reclaimPolicy": "Delete", "volumeBindingMode": "WaitForFirstConsumer"}]},
"node_top": None,
"pod_top": None,
}
result = module.build_observation(snapshots, "2026-08-22T00:00:00Z")
self.assertEqual(250, result["allocation"]["denominators"]["cpu_requested_millicores"])
self.assertEqual(1750, result["allocation"]["residual_capacity"]["cpu_millicores"])
self.assertIn("pod metrics unavailable", result["measurement_gaps"])
self.assertFalse(result["failure_domain"]["threephoenix_target_is_live"])
if __name__ == "__main__":
unittest.main()