prole/tests/installer/test_database_context_selection.py
chrisfu 5cd43c873c feat(gke): finalize dual-cluster app/db topology
- separate default app and db cluster contexts (knoe-dev-0 and knoe-cnpg-0) across actions, ops flows, and installer config\n- update CNPG placement and GKE manifests to keep PostgreSQL on the db cluster while app services stay on the app cluster\n- add GCS-backed Barman ObjectStore manifest and cross-cluster Garage patch workflow\n- refresh UI cluster/database/service screens and monitoring wiring for dual-cluster operation\n- add reset/patch scripts, db context selection test coverage, and architecture/network documentation updates

Co-authored-by: Junie <junie@jetbrains.com>
2026-04-08 23:12:51 -07:00

93 lines
2.6 KiB
Python

from __future__ import annotations
from knoe.ui.screens.database import DatabaseScreenMixin
class _Var:
def __init__(self, value: str = ""):
self._value = value
def get(self):
return self._value
def set(self, value):
self._value = value
class _DummyDbScreen(DatabaseScreenMixin):
def __init__(self, *, inputs: dict | None = None, prole_cfg_data: dict | None = None):
self.inputs = inputs or {}
self.prole_cfg_data = prole_cfg_data or {}
self.selected_kubectx = _Var(
(self.inputs.get("init_cluster.selected_kubectx") or "")
)
def _get_input(self, key: str, default=""):
return self.inputs.get(key, default)
def _deployment_mode(self):
return "k8s"
def _kubectl_base_cmd(self, mode: str | None = None):
_ = mode
return ["kubectl", "--context", "gke_proj_us-west3_knoe-dev-0"]
def test_db_kubectl_base_cmd_prefers_db_cluster_context_over_dev_context():
screen = _DummyDbScreen(
prole_cfg_data={
"Global": {
"KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
"DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
}
}
)
cmd = screen._db_kubectl_base_cmd()
assert cmd == ["kubectl", "--context", "gke_proj_us-west3_knoe-cnpg-0"]
def test_resolve_db_kube_context_derives_cnpg_context_when_only_app_context_exists():
screen = _DummyDbScreen(
prole_cfg_data={
"Global": {
"KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
"DB_CLUSTER_NAME": "knoe-cnpg-0",
}
}
)
assert screen._resolve_db_kube_context() == "gke_proj_us-west3_knoe-cnpg-0"
def test_ensure_db_screen_context_keeps_explicit_user_override():
screen = _DummyDbScreen(
inputs={"init_cluster.selected_kubectx": "gke_proj_us-west3_custom-ops"},
prole_cfg_data={
"Global": {
"KUBECONTEXT": "gke_proj_us-west3_knoe-dev-0",
"DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
}
},
)
screen._ensure_db_screen_context()
assert screen.selected_kubectx.get() == "gke_proj_us-west3_custom-ops"
def test_db_target_note_includes_context_and_namespace_confirmation():
screen = _DummyDbScreen(
prole_cfg_data={
"Global": {
"DB_CLUSTER_KUBECONTEXT": "gke_proj_us-west3_knoe-cnpg-0",
}
}
)
note = screen._db_target_note("knoe-db")
assert "Target context: gke_proj_us-west3_knoe-cnpg-0" in note
assert "Target namespace: knoe-db" in note