mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 12:03:59 +00:00
c94c62e1bb
- persist and load DB master password via Ansible Vault bootstrap flow - enforce knoe-system service namespace and explicit app/db kubecontext targeting - improve OpenBao/CNPG deploy reliability and logging; add retries/readiness diagnostics - tighten reset/delete cluster behavior and expand installer/deploy pipeline test coverage Co-authored-by: Junie <junie@jetbrains.com>
99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from types import SimpleNamespace
|
|
|
|
from cryptography import x509
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
from knoe.core.ops import cloudnative_pg
|
|
|
|
|
|
def _decode_tls_cert_pem(manifest: str) -> bytes:
|
|
for line in manifest.splitlines():
|
|
key, _, value = line.partition(":")
|
|
if key.strip() == "tls.crt":
|
|
return base64.b64decode(value.strip())
|
|
raise AssertionError("tls.crt not found in manifest")
|
|
|
|
|
|
def test_bootstrap_cnpg_tls_secrets_falls_back_to_short_cn_for_long_namespace(monkeypatch):
|
|
namespace = "knoe-db-17-7-043-18-140-17-7-043-18-140-17-7-043-18-140"
|
|
cluster_name = "knoe-db"
|
|
applied_manifests: list[str] = []
|
|
|
|
monkeypatch.setattr(cloudnative_pg, "_ensure_namespace", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(cloudnative_pg, "_kubectl_ok", lambda *_args, **_kwargs: False)
|
|
|
|
def _fake_run(args, input=None, **_kwargs):
|
|
assert args == ["kubectl", "apply", "-n", namespace, "-f", "-"]
|
|
assert input is not None
|
|
applied_manifests.append(input)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(cloudnative_pg.subprocess, "run", _fake_run)
|
|
|
|
cloudnative_pg.bootstrap_cnpg_tls_secrets(namespace=namespace, cluster_name=cluster_name)
|
|
|
|
tls_manifest = next(m for m in applied_manifests if "kubernetes.io/tls" in m)
|
|
cert = x509.load_pem_x509_certificate(_decode_tls_cert_pem(tls_manifest))
|
|
|
|
cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
|
|
assert cn == "knoe-db-rw"
|
|
assert len(cn) <= 64
|
|
|
|
san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName).value
|
|
san_dns = san.get_values_for_type(x509.DNSName)
|
|
assert f"{cluster_name}-rw.{namespace}.svc" in san_dns
|
|
|
|
|
|
def test_bootstrap_cnpg_tls_secrets_keeps_fqdn_cn_when_within_limit(monkeypatch):
|
|
namespace = "knoe-db"
|
|
cluster_name = "knoe-db"
|
|
applied_manifests: list[str] = []
|
|
|
|
monkeypatch.setattr(cloudnative_pg, "_ensure_namespace", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(cloudnative_pg, "_kubectl_ok", lambda *_args, **_kwargs: False)
|
|
|
|
def _fake_run(args, input=None, **_kwargs):
|
|
assert args == ["kubectl", "apply", "-n", namespace, "-f", "-"]
|
|
assert input is not None
|
|
applied_manifests.append(input)
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(cloudnative_pg.subprocess, "run", _fake_run)
|
|
|
|
cloudnative_pg.bootstrap_cnpg_tls_secrets(namespace=namespace, cluster_name=cluster_name)
|
|
|
|
tls_manifest = next(m for m in applied_manifests if "kubernetes.io/tls" in m)
|
|
cert = x509.load_pem_x509_certificate(_decode_tls_cert_pem(tls_manifest))
|
|
|
|
cn = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
|
|
assert cn == f"{cluster_name}.{namespace}.svc"
|
|
assert len(cn) <= 64
|
|
|
|
|
|
def test_bootstrap_cnpg_tls_secrets_honors_kubecontext(monkeypatch):
|
|
namespace = "knoe-db-0"
|
|
cluster_name = "knoe-db"
|
|
commands: list[list[str]] = []
|
|
|
|
monkeypatch.setattr(cloudnative_pg, "_ensure_namespace", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(cloudnative_pg, "_kubectl_ok", lambda *_args, **_kwargs: False)
|
|
|
|
def _fake_run(args, input=None, **_kwargs):
|
|
commands.append(list(args))
|
|
assert input is not None
|
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
|
|
|
monkeypatch.setattr(cloudnative_pg.subprocess, "run", _fake_run)
|
|
|
|
cloudnative_pg.bootstrap_cnpg_tls_secrets(
|
|
namespace=namespace,
|
|
cluster_name=cluster_name,
|
|
env={"KUBECONTEXT": "gke-db-context"},
|
|
)
|
|
|
|
assert commands
|
|
assert all(cmd[:3] == ["kubectl", "--context", "gke-db-context"] for cmd in commands)
|