48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
|
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": 2, "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]])
|