railiance-platform/tests/test_apps_pg_capacity.py

from __future__ import annotations

from copy import deepcopy
import importlib.util
from pathlib import Path
import sys

import pytest
import yaml

ROOT = Path(__file__).parents[1]
SPEC = importlib.util.spec_from_file_location(
    "verify_apps_pg_capacity", ROOT / "tools/verify_apps_pg_capacity.py"
)
capacity = importlib.util.module_from_spec(SPEC)
assert SPEC.loader is not None
sys.modules[SPEC.name] = capacity
SPEC.loader.exec_module(capacity)
MAX_CONSUMERS = capacity.MAX_CONSUMERS
verify = capacity.verify
MANIFESTS = [ROOT / "helm/apps-pg-cluster.yaml", ROOT / "helm/apps-pg-2-cluster.yaml"]


def test_reviewed_apps_pg_cells_fit_capacity_and_use_distinct_backups() -> None:
    assert verify(MANIFESTS) == {"apps-pg": 3, "apps-pg-2": 0}


def test_fourth_role_is_rejected(tmp_path: Path) -> None:
    primary = yaml.safe_load(MANIFESTS[0].read_text(encoding="utf-8"))
    template = deepcopy(primary["spec"]["managed"]["roles"][0])
    while len(primary["spec"]["managed"]["roles"]) <= MAX_CONSUMERS:
        role = deepcopy(template)
        role["name"] = f"consumer_{len(primary['spec']['managed']['roles'])}"
        primary["spec"]["managed"]["roles"].append(role)
    path = tmp_path / "apps-pg.yaml"
    path.write_text(yaml.safe_dump(primary), encoding="utf-8")
    with pytest.raises(ValueError, match="3-consumer ceiling"):
        verify([path, MANIFESTS[1]])


def test_unbounded_role_is_rejected(tmp_path: Path) -> None:
    primary = yaml.safe_load(MANIFESTS[0].read_text(encoding="utf-8"))
    del primary["spec"]["managed"]["roles"][0]["connectionLimit"]
    path = tmp_path / "apps-pg.yaml"
    path.write_text(yaml.safe_dump(primary), encoding="utf-8")
    with pytest.raises(ValueError, match="connectionLimit"):
        verify([path, MANIFESTS[1]])