eda153c7a1
build_report calls the claim lookup. A workload with no claim, and a label with no claim, receive neither a unit id nor a declaration owner.
91 lines
3.7 KiB
Python
91 lines
3.7 KiB
Python
"""Release-unit names on the weekly allocation recommendation."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("weekly", ROOT / "scripts/weekly_allocation_report.py")
|
|
weekly = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(weekly)
|
|
|
|
END = "2026-10-19T06:00:00+00:00"
|
|
RESOURCE = "resource:railiance:reef-railiance:k3s"
|
|
INDEX = {
|
|
"schema": "release-registry.index.v1",
|
|
"units": [
|
|
{
|
|
"unit_id": "flex-auth-secrets-engine",
|
|
"declaration": {
|
|
"unit_id": "flex-auth-secrets-engine",
|
|
"owner": "access-engine",
|
|
"workloads": [
|
|
{"kind": "Deployment", "namespace": "flex-auth", "name": "flex-auth-secrets-engine"}
|
|
],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _workload(namespace, name, labels=None):
|
|
item = {
|
|
"namespace": namespace, "workload": name, "pods": 1,
|
|
"requests": {"cpu_millicores": 60, "memory_bytes": 256},
|
|
"limits": {"cpu_millicores": 1000, "memory_bytes": 1024},
|
|
}
|
|
if labels:
|
|
item["labels"] = labels
|
|
return item
|
|
|
|
|
|
class WeeklyReleaseOwnerTests(unittest.TestCase):
|
|
def report(self):
|
|
cluster = {
|
|
"schema_version": "railiance.cluster-resource-observation.v1",
|
|
"resource_id": RESOURCE, "captured_at": END,
|
|
"utilization": {"workloads": [
|
|
_workload("flex-auth", "flex-auth-secrets-engine"),
|
|
_workload("flex-auth", "flex-auth-tenant-engine"),
|
|
_workload("forgejo", "forge", {"railiance.io/release-unit": "flex-auth-secrets-engine"}),
|
|
]},
|
|
}
|
|
allocation = {
|
|
"schema": "custodian.allocation-reconcile.v1", "captured_at": END,
|
|
"host": {"cluster_resource_id": RESOURCE}, "count_host_and_cluster_cpus_once": True,
|
|
"workloads": [
|
|
{"namespace": "flex-auth", "workload": "flex-auth-secrets-engine", "owner": "namespace-map"},
|
|
{"namespace": "flex-auth", "workload": "flex-auth-tenant-engine", "owner": "namespace-map"},
|
|
{"namespace": "forgejo", "workload": "forge", "owner": "namespace-map"},
|
|
],
|
|
}
|
|
return weekly.build_report(
|
|
cluster, allocation,
|
|
{"captured_at": END, "window": "7d", "step": "5m", "queries": {}},
|
|
{"id": RESOURCE},
|
|
{"source_observation": END, "revisions": {"collector": "a" * 40}},
|
|
as_of=END, release_index=INDEX,
|
|
)
|
|
|
|
def by_name(self, report):
|
|
return {item["workload"]: item for item in report["recommendations"]}
|
|
|
|
def test_claim_label_and_gap_name_unit_and_owner(self):
|
|
first = self.by_name(self.report())
|
|
second = self.by_name(self.report())
|
|
claimed = first["flex-auth-secrets-engine"]
|
|
self.assertEqual(claimed["release_unit"], "flex-auth-secrets-engine")
|
|
self.assertEqual(claimed["owner"], "access-engine")
|
|
self.assertEqual(claimed["release_owner"], "access-engine")
|
|
self.assertEqual(second["flex-auth-secrets-engine"]["release_unit"], claimed["release_unit"])
|
|
self.assertEqual(second["flex-auth-secrets-engine"]["owner"], claimed["owner"])
|
|
unclaimed = first["flex-auth-tenant-engine"]
|
|
self.assertIsNone(unclaimed["release_unit"])
|
|
self.assertIsNone(unclaimed["release_owner"])
|
|
labeled = first["forge"]
|
|
self.assertIsNone(labeled["release_unit"])
|
|
self.assertIsNone(labeled["release_owner"])
|
|
self.assertNotEqual(labeled["owner"], "flex-auth-secrets-engine")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|