"""Tests for --repair/--update CLI paths and Supabase component config flags.""" from __future__ import annotations import configparser import importlib.util import sys from argparse import Namespace from pathlib import Path from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # Tkinter / GUI stubs (must be installed before any knoe import) # --------------------------------------------------------------------------- for _mod in ( "tkinter", "tkinter.ttk", "tkinter.scrolledtext", "tkinter.messagebox", "tkinter.filedialog", "Foundation", "objc", ): if _mod not in sys.modules: sys.modules[_mod] = MagicMock() REPO_ROOT = Path(__file__).resolve().parents[1] # --------------------------------------------------------------------------- # Helpers to load render_supabase without executing its __main__ block # --------------------------------------------------------------------------- _RENDER_PATH = REPO_ROOT / "supabase" / "helm" / "render_supabase.py" def _load_render_module(): spec = importlib.util.spec_from_file_location("knoe_render_supabase_flags", _RENDER_PATH) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod _render_mod = _load_render_module() _build_overlay = _render_mod._build_overlay _read_cfg = _render_mod._read_cfg def _write_cfg(path: Path, content: str) -> None: path.write_text(content.strip() + "\n", encoding="utf-8") def _minimal_cfg(tmp_path: Path, extra: str = "") -> configparser.ConfigParser: cfg_path = tmp_path / "knoe.cfg" _write_cfg( cfg_path, f""" [Inputs] init_password.db_password = test-password init_password.db_host_port = 5432 {extra} [Global] NAMESPACE = test-ns STORAGE_BACKEND = local """, ) return _read_cfg(cfg_path) def _make_installer(tmp_path: Path): """Return a minimal KnoeConsoleInstaller with mocked controller.""" from knoe.core.actions import KnoeConsoleInstaller from knoe.core.controller import KnoeController cfg_path = tmp_path / "knoe.cfg" if not cfg_path.exists(): _write_cfg(cfg_path, "[Inputs]\n[Global]\n[Install]\nSTATUS = New\n") controller = MagicMock(spec=KnoeController) controller.project_root = tmp_path controller.state = MagicMock() installer = KnoeConsoleInstaller.__new__(KnoeConsoleInstaller) installer.controller = controller installer.cfg_path = cfg_path installer.inputs = {} installer.knoe_cfg_data = {"Inputs": {}, "Global": {}, "Install": {"STATUS": "New"}} installer._log_file = None installer.verbose = False installer.dependencies = [] return installer def _defaults(installer): """Call _default_inputs() with all internal deps stubbed.""" from knoe.core.actions import KnoeConsoleInstaller with ( patch.object(KnoeConsoleInstaller, "_load_inputs_from_cfg", return_value={}), patch.object(KnoeConsoleInstaller, "_initial_db_namespace", return_value="test-ns"), patch.object(KnoeConsoleInstaller, "_cnpg_cluster_name", return_value="knoe-db"), patch.object(KnoeConsoleInstaller, "_get_local_owner", return_value="root"), patch.object(KnoeConsoleInstaller, "_env_defaults", return_value={}), ): return installer._default_inputs() # =========================================================================== # 1. CLI argument parsing — --repair and --update must be recognised # =========================================================================== def test_main_parser_accepts_repair_flag(): """install.py --repair and --update must parse without error.""" import argparse parser = argparse.ArgumentParser() parser.add_argument("--repair", action="store_true") parser.add_argument("--update", action="store_true") parser.add_argument("--silent", action="store_true") parser.add_argument("--reset", action="store_true") parser.add_argument("--no-gui", action="store_true") parser.add_argument("--gui", action="store_true") parser.add_argument("-c", "--config", default=None) parser.add_argument("-v", "--verbose", action="store_true") parser.add_argument("-d", "--debug", action="store_true") parser.add_argument("-l", "--log", nargs="?", const=True, default=None) parser.add_argument("--prepare-k3s-pipeline", action="store_true") parser.add_argument("command", nargs="?", default=None) parser.add_argument("command_args", nargs="*") args = parser.parse_args(["--repair"]) assert args.repair is True assert args.update is False args2 = parser.parse_args(["--update"]) assert args2.update is True assert args2.repair is False def test_main_update_flag_dispatches_to_run_update(tmp_path, monkeypatch): """install.py --update must run console update flow and exit with its status.""" import knoe.ui.screens as screens_mod cfg_path = tmp_path / "knoe.cfg" _write_cfg(cfg_path, "[Inputs]\n[Global]\n") installer = MagicMock() installer.run_update.return_value = 0 installer.run_repair.return_value = 0 installer.run.return_value = 0 monkeypatch.setattr(sys, "argv", ["install.py", "--update", "-c", str(cfg_path)]) monkeypatch.setattr(screens_mod, "ensure_op_signed_in", lambda: None) monkeypatch.setattr(screens_mod, "ensure_knoey_vault", lambda: None) monkeypatch.setattr(screens_mod, "KnoeController", MagicMock(return_value=MagicMock())) monkeypatch.setattr(screens_mod, "KnoeConsoleInstaller", MagicMock(return_value=installer)) with pytest.raises(SystemExit) as ex: screens_mod.main() assert ex.value.code == 0 installer.run_update.assert_called_once() installer.run_repair.assert_not_called() installer.run.assert_not_called() def test_main_rejects_repair_and_update_together(tmp_path, monkeypatch): """CLI must reject using --repair and --update together.""" import knoe.ui.screens as screens_mod cfg_path = tmp_path / "knoe.cfg" _write_cfg(cfg_path, "[Inputs]\n[Global]\n") monkeypatch.setattr( sys, "argv", ["install.py", "--repair", "--update", "-c", str(cfg_path)], ) with pytest.raises(SystemExit): screens_mod.main() def test_main_rejects_delete_db_without_reset(tmp_path, monkeypatch): """CLI must require --reset when --delete-db is provided.""" import knoe.ui.screens as screens_mod cfg_path = tmp_path / "knoe.cfg" _write_cfg(cfg_path, "[Inputs]\n[Global]\n") monkeypatch.setattr( sys, "argv", ["install.py", "--silent", "--delete-db", "-c", str(cfg_path)], ) with pytest.raises(SystemExit): screens_mod.main() def test_main_delete_db_is_forwarded_to_console_installer(tmp_path, monkeypatch): """install.py --delete-db must be forwarded to KnoeConsoleInstaller in silent mode.""" import knoe.ui.screens as screens_mod cfg_path = tmp_path / "knoe.cfg" _write_cfg(cfg_path, "[Inputs]\n[Global]\n") installer = MagicMock() installer.run.return_value = 0 installer.run_update.return_value = 0 installer.run_repair.return_value = 0 installer_factory = MagicMock(return_value=installer) monkeypatch.setattr( sys, "argv", ["install.py", "--silent", "--reset", "--delete-db", "-c", str(cfg_path)], ) monkeypatch.setattr(screens_mod, "ensure_op_signed_in", lambda: None) monkeypatch.setattr(screens_mod, "ensure_knoey_vault", lambda: None) monkeypatch.setattr(screens_mod, "KnoeController", MagicMock(return_value=MagicMock())) monkeypatch.setattr(screens_mod, "KnoeConsoleInstaller", installer_factory) with pytest.raises(SystemExit) as ex: screens_mod.main() assert ex.value.code == 0 installer_factory.assert_called_once() _, kwargs = installer_factory.call_args assert kwargs.get("reset_cluster") is True assert kwargs.get("delete_db") is True installer.run.assert_called_once() def test_repair_flag_does_not_trigger_silent_install(tmp_path): """--repair must dispatch to run_repair(), NOT run() (full install).""" from knoe.core.actions import KnoeConsoleInstaller installer = _make_installer(tmp_path) run_called = [] run_repair_called = [] with ( patch.object(KnoeConsoleInstaller, "run", side_effect=lambda: run_called.append(1) or 0), patch.object( KnoeConsoleInstaller, "run_repair", side_effect=lambda: run_repair_called.append(1) or 0, ), ): installer.run_repair() assert run_repair_called, "run_repair() was not called" assert not run_called, "run() (full install) must NOT be called by --repair" def test_update_calls_repair_first(tmp_path): """run_update() must invoke run_repair() before applying config changes.""" from knoe.core.actions import KnoeConsoleInstaller installer = _make_installer(tmp_path) call_order: list[str] = [] with ( patch.object( KnoeConsoleInstaller, "run_repair", side_effect=lambda: call_order.append("repair") or 0, ), patch.object(KnoeConsoleInstaller, "_load_inputs_from_cfg", return_value={}), patch.object(KnoeConsoleInstaller, "_default_inputs", return_value={}), patch.object(KnoeConsoleInstaller, "_service_namespace", return_value="test-ns"), patch.object(KnoeConsoleInstaller, "_script_env_for_namespace", return_value={}), patch.object( KnoeConsoleInstaller, "_reconcile_supabase", side_effect=lambda env: call_order.append("reconcile"), ), patch.object( KnoeConsoleInstaller, "_step_db_build", side_effect=lambda: call_order.append("db_build"), ), patch.object( KnoeConsoleInstaller, "_step_cnpg_deploy", side_effect=lambda: call_order.append("cnpg_deploy"), ), patch.object(KnoeConsoleInstaller, "_write_cfg", return_value=None), patch.object(KnoeConsoleInstaller, "_close_log_file", return_value=None), ): installer.run_update() assert call_order and call_order[0] == "repair", ( "run_repair() must be first in run_update()" ) def test_update_forces_cnpg_rollout_only_for_deploy_step(tmp_path): """run_update() must force rollout for CNPG deploy, then restore configured value.""" from knoe.core.actions import KnoeConsoleInstaller installer = _make_installer(tmp_path) seen_force_rollout: list[bool] = [] with ( patch.object(KnoeConsoleInstaller, "run_repair", return_value=0), patch.object( KnoeConsoleInstaller, "_load_inputs_from_cfg", return_value={ "init_cnpg_deploy.run_deploy": "true", "init_cnpg_deploy.force_rollout": "false", "init_db_build.run_build": "false", "init_cluster.supabase_enabled": "false", }, ), patch.object(KnoeConsoleInstaller, "_default_inputs", return_value={}), patch.object(KnoeConsoleInstaller, "_service_namespace", return_value="test-ns"), patch.object(KnoeConsoleInstaller, "_script_env_for_namespace", return_value={}), patch.object(KnoeConsoleInstaller, "_write_cfg", return_value=None), patch.object(KnoeConsoleInstaller, "_close_log_file", return_value=None), patch.object( KnoeConsoleInstaller, "_step_cnpg_deploy", side_effect=lambda: seen_force_rollout.append( installer._get_input_bool("init_cnpg_deploy.force_rollout", False) ), ), ): installer.run_update() assert seen_force_rollout == [True] assert installer.inputs["init_cnpg_deploy.force_rollout"] == "false" # =========================================================================== # 2. Config defaults contain all new Supabase flags # =========================================================================== def test_default_inputs_contains_supabase_component_flags(tmp_path): """_default_inputs() must include all new supabase_* keys.""" installer = _make_installer(tmp_path) defaults = _defaults(installer) expected_keys = [ "init_cluster.supabase_enabled", "init_cluster.supabase_studio_enabled", "init_cluster.supabase_studio_url", "init_cluster.supabase_auth_enabled", "init_cluster.supabase_realtime_enabled", "init_cluster.supabase_meta_enabled", "init_cluster.supabase_analytics_enabled", ] for key in expected_keys: assert key in defaults, f"Missing default for {key!r}" assert defaults["init_cluster.supabase_studio_url"] == "db.0.knoe.dev" assert defaults["init_cluster.supabase_studio_enabled"] in ("false", "False", False) # =========================================================================== # 3. render_supabase.py — supabase_studio_url wires ingress host # =========================================================================== def test_render_supabase_studio_url_overrides_ingress_host(tmp_path, monkeypatch): """supabase_studio_url must become studio_ingress_host in overlay.""" for ev in ( "SUPABASE_HOSTNAME", "SUPABASE_HOST", "SUPABASE_STUDIO_URL", "SUPABASE_STUDIO_ENABLED", "KNOE_DB_NAMESPACE", "KNOE_DB_SERVICE", ): monkeypatch.delenv(ev, raising=False) cfg = _minimal_cfg( tmp_path, extra=( "init_cluster.supabase_studio_enabled = true\n" "init_cluster.supabase_studio_url = studio.knoe.org\n" ), ) args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) assert overlay["deployment"]["studio"]["enabled"] is True studio_hosts = overlay.get("studioIngress", {}).get("hosts", []) assert studio_hosts and studio_hosts[0]["host"] == "studio.knoe.org" def test_render_supabase_studio_multi_url_creates_multiple_ingress_hosts(tmp_path, monkeypatch): """Comma-separated supabase_studio_url must produce one ingress host entry per URL.""" for ev in ( "SUPABASE_HOSTNAME", "SUPABASE_HOST", "SUPABASE_STUDIO_URL", "SUPABASE_STUDIO_ENABLED", "KNOE_DB_NAMESPACE", "KNOE_DB_SERVICE", ): monkeypatch.delenv(ev, raising=False) cfg = _minimal_cfg( tmp_path, extra=( "init_cluster.supabase_studio_enabled = true\n" "init_cluster.supabase_studio_url = supabase.knoe.org,db.knoe.org\n" ), ) args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) assert overlay["deployment"]["studio"]["enabled"] is True studio_hosts = overlay.get("studioIngress", {}).get("hosts", []) host_names = [h["host"] for h in studio_hosts] assert "supabase.knoe.org" in host_names, f"Expected supabase.knoe.org in {host_names}" assert "db.knoe.org" in host_names, f"Expected db.knoe.org in {host_names}" assert len(host_names) == 2, f"Expected exactly 2 hosts, got {host_names}" # First entry is the primary (used for SUPABASE_PUBLIC_URL) assert host_names[0] == "supabase.knoe.org" def test_render_supabase_studio_disabled_sets_enabled_false(tmp_path, monkeypatch): """deployment.studio.enabled must be False when supabase_studio_enabled=false.""" for ev in ( "SUPABASE_HOSTNAME", "SUPABASE_HOST", "SUPABASE_STUDIO_URL", "SUPABASE_STUDIO_ENABLED", "KNOE_DB_NAMESPACE", "KNOE_DB_SERVICE", ): monkeypatch.delenv(ev, raising=False) cfg = _minimal_cfg(tmp_path, extra="init_cluster.supabase_studio_enabled = false\n") args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) assert overlay["deployment"]["studio"]["enabled"] is False def test_render_supabase_k8s_adds_gce_tls_annotations_for_api_and_studio(tmp_path, monkeypatch): """k8s/GKE overlay must attach explicit ManagedCertificate + FrontendConfig annotations.""" monkeypatch.setenv("KNOE_MODE", "k8s") monkeypatch.setenv("SUPABASE_API_HOSTNAME", "api.0.knoe.dev") monkeypatch.setenv("SUPABASE_STUDIO_HOSTNAME", "db.0.knoe.dev") cfg = _minimal_cfg(tmp_path) args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) api_annotations = overlay.get("ingress", {}).get("annotations", {}) studio_annotations = overlay.get("studioIngress", {}).get("annotations", {}) assert api_annotations.get("kubernetes.io/ingress.class") == "gce" assert studio_annotations.get("kubernetes.io/ingress.class") == "gce" assert api_annotations.get("networking.gke.io/managed-certificates") == "supabase-api-managed-cert" assert studio_annotations.get("networking.gke.io/managed-certificates") == "supabase-studio-managed-cert" assert api_annotations.get("networking.gke.io/v1beta1.FrontendConfig") == "supabase-api-frontend-config" assert studio_annotations.get("networking.gke.io/v1beta1.FrontendConfig") == "supabase-studio-frontend-config" assert "ingress.gcp.kubernetes.io/pre-shared-cert" not in api_annotations assert "ingress.gcp.kubernetes.io/pre-shared-cert" not in studio_annotations def test_render_supabase_render_writes_gke_public_ingress_tls_manifest(tmp_path, monkeypatch): """render() must emit ManagedCertificate/FrontendConfig manifests for Supabase public ingresses in k8s mode.""" import subprocess as sp monkeypatch.setenv("KNOE_MODE", "k8s") monkeypatch.setenv("SUPABASE_API_HOSTNAME", "api.0.knoe.dev") monkeypatch.setenv("SUPABASE_STUDIO_HOSTNAME", "db.0.knoe.dev") cfg_path = tmp_path / "knoe.cfg" _write_cfg( cfg_path, """ [Inputs] init_password.db_password = test-password init_password.db_host_port = 5432 [Global] NAMESPACE = supabase STORAGE_BACKEND = local DEPLOYMENT_MODE = k8s """, ) def _fake_run(cmd, capture_output=True, text=True): return sp.CompletedProcess(cmd, 0, stdout="apiVersion: v1\nkind: ConfigMap\nmetadata:\n name: supabase-placeholder\n", stderr="") monkeypatch.setattr(_render_mod.subprocess, "run", _fake_run) args = Namespace(config=str(cfg_path), output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) _render_mod.render(args) tls_manifest = (tmp_path / "k8s" / "public-ingress-tls.yaml") assert tls_manifest.exists(), "Expected public-ingress-tls.yaml to be generated in k8s mode" rendered = tls_manifest.read_text(encoding="utf-8") assert "kind: ManagedCertificate" in rendered assert "kind: FrontendConfig" in rendered assert "supabase-api-managed-cert" in rendered assert "supabase-studio-managed-cert" in rendered # =========================================================================== # 4. render_supabase.py — per-component flags respected # =========================================================================== def test_render_supabase_component_flags_disable_components(tmp_path, monkeypatch): """Disabled per-component flags must set deployment..enabled=False.""" for ev in ( "SUPABASE_HOSTNAME", "SUPABASE_HOST", "SUPABASE_STUDIO_URL", "SUPABASE_STUDIO_ENABLED", "SUPABASE_AUTH_ENABLED", "SUPABASE_REALTIME_ENABLED", "SUPABASE_META_ENABLED", "SUPABASE_ANALYTICS_ENABLED", "KNOE_DB_NAMESPACE", "KNOE_DB_SERVICE", ): monkeypatch.delenv(ev, raising=False) cfg = _minimal_cfg( tmp_path, extra=( "init_cluster.supabase_auth_enabled = false\n" "init_cluster.supabase_realtime_enabled = false\n" "init_cluster.supabase_meta_enabled = true\n" "init_cluster.supabase_analytics_enabled = false\n" ), ) args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) assert overlay["deployment"]["auth"]["enabled"] is False assert overlay["deployment"]["realtime"]["enabled"] is False assert overlay["deployment"]["meta"]["enabled"] is True assert overlay["deployment"]["analytics"]["enabled"] is False def test_render_supabase_component_flags_default_to_enabled(tmp_path, monkeypatch): """When no per-component flags are set, all components default to enabled.""" for ev in ( "SUPABASE_HOSTNAME", "SUPABASE_HOST", "SUPABASE_STUDIO_URL", "SUPABASE_STUDIO_ENABLED", "SUPABASE_AUTH_ENABLED", "SUPABASE_REALTIME_ENABLED", "SUPABASE_META_ENABLED", "SUPABASE_ANALYTICS_ENABLED", "KNOE_DB_NAMESPACE", "KNOE_DB_SERVICE", ): monkeypatch.delenv(ev, raising=False) cfg = _minimal_cfg(tmp_path) args = Namespace(output_dir=str(tmp_path / "gen"), manifests_dir=str(tmp_path / "k8s")) overlay, _ = _build_overlay(cfg, args) for component in ("auth", "realtime", "meta", "analytics"): assert overlay["deployment"][component]["enabled"] is True, ( f"Expected {component} enabled by default" ) # =========================================================================== # 5. Studio reconcile respects the enabled flag # =========================================================================== def test_reconcile_supabase_skips_studio_when_disabled(tmp_path): """_reconcile_supabase() must skip Studio when supabase_studio_enabled=false.""" from knoe.core.actions import KnoeConsoleInstaller installer = _make_installer(tmp_path) base = _defaults(installer) installer.inputs = { **base, "init_cluster.supabase_enabled": "true", "init_cluster.supabase_studio_enabled": "false", } studio_reconcile_called = [] with patch.object( KnoeConsoleInstaller, "_reconcile_supabase_studio", side_effect=lambda env: studio_reconcile_called.append(1), ): installer._reconcile_supabase({}) assert not studio_reconcile_called, ( "_reconcile_supabase_studio() must NOT be called when studio is disabled" ) def test_reconcile_supabase_calls_studio_when_enabled(tmp_path): """_reconcile_supabase() must invoke Studio reconcile when enabled.""" from knoe.core.actions import KnoeConsoleInstaller installer = _make_installer(tmp_path) base = _defaults(installer) installer.inputs = { **base, "init_cluster.supabase_enabled": "true", "init_cluster.supabase_studio_enabled": "true", } studio_reconcile_called = [] with patch.object( KnoeConsoleInstaller, "_reconcile_supabase_studio", side_effect=lambda env: studio_reconcile_called.append(1), ): installer._reconcile_supabase({}) assert studio_reconcile_called, ( "_reconcile_supabase_studio() must be called when studio is enabled" ) def test_supabase_deploy_includes_knoe_schema_and_lint_remediation_sql(): """supabase/deploy.sh must include recurring knoe-schema and lint remediation SQL.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert "CREATE ROLE knoe LOGIN PASSWORD" in script assert "CREATE SCHEMA IF NOT EXISTS knoe AUTHORIZATION knoe;" in script assert "migrations CASCADE;" in script assert "GRANT CREATE ON DATABASE postgres TO supabase_storage_admin;" in script assert "GRANT USAGE, CREATE ON SCHEMA public TO supabase_storage_admin;" in script assert "EXTENSION %I SET SCHEMA" in script assert "ALTER TABLE %I.spatial_ref_sys ENABLE ROW LEVEL SECURITY" in script assert "CREATE POLICY spatial_ref_sys_select_all ON %I.spatial_ref_sys" in script def test_supabase_deploy_db_frontdoor_readiness_and_storageclass_guards_present(): """supabase/deploy.sh must enforce DB-frontdoor PVC class and report concrete blockers.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert "SUPABASE_DB_FRONTDOOR_STORAGE_CLASS" in script assert "DB frontdoor PVC storageClass" in script assert 'wait_for_supabase_ready "$ns" "$timeout_s" "$db_ctx" "Supabase frontdoor (DB cluster)"' in script assert "Blocking PVCs:" in script assert "Blocking Pods:" in script def test_supabase_deploy_defaults_non_db_storage_class_to_csi_pd_standard_on_gke(): """GKE Supabase non-DB PVC defaults must use a CSI pd-standard/WFFC class, not legacy `standard`.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert 'local sc_name="${SUPABASE_GKE_STORAGE_CLASS:-supabase-gke-standard-rwo}"' in script assert "provisioner: pd.csi.storage.gke.io" in script assert "volumeBindingMode: WaitForFirstConsumer" in script assert 'local preferred="${SUPABASE_DB_FRONTDOOR_STORAGE_CLASS:-$default_storage_class}"' in script assert "Ignoring legacy 'standard' storage class for Supabase GKE non-DB PVCs" in script assert 'preferred="standard"' in script assert 'local sc_name="${SUPABASE_GKE_STORAGE_CLASS:-standard}"' not in script assert 'local preferred="${SUPABASE_DB_FRONTDOOR_STORAGE_CLASS:-standard}"' not in script def test_supabase_deploy_reconciles_gke_non_db_pvcs_and_forces_single_functions_replica(): """GKE split deploy must reconcile mismatched APP PVC classes and force single-replica functions semantics.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert "reconcile_supabase_app_pvcs()" in script assert "for pvc in supabase-deno supabase-functions supabase-imgproxy supabase-minio supabase-storage" in script assert "Recreating APP PVC" in script assert 'reconcile_supabase_app_pvcs "$ns" "$storage_class"' in script assert 'SUPABASE_GKE_FORCE_FUNCTIONS_SINGLE_REPLICA="$force_gke_functions_single_replica"' in script assert 'functions_cfg["replicaCount"] = 1' in script def test_supabase_retained_disk_cleanup_uses_cfg_project_and_reasoned_failures(): """Retained disk cleanup should resolve project/flags from cfg fallbacks and report explicit skip/failure reasons.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert "supabase_cfg_first_nonempty_value()" in script assert '"Inputs:init_cluster.project_id"' in script assert '"Inputs:init_cluster.SUPABASE_AUTO_CLEAN_RETAINED_GCE_DISKS"' in script assert "Supabase retained cleanup settings: pv_auto_clean=true, gce_disk_auto_clean=" in script assert "reason=missing_project" in script assert "Missing project resolution: no GCP project could be resolved from env/config/context." in script assert "Missing gcloud auth: no active authenticated gcloud account/session was found." in script def test_supabase_deploy_split_app_and_db_values_disable_general_node_role_enforcement(): """Split APP/DB Supabase values must disable legacy knoe node-role scheduling enforcement.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert 'app_values.setdefault("scheduling", {})["enforceGeneralNodeRole"] = False' in script assert 'db_values.setdefault("scheduling", {})["enforceGeneralNodeRole"] = False' in script def test_supabase_deploy_split_app_values_clear_stale_scheduling_constraints(): """APP split values must scrub stale node/affinity/topology placement constraints for GKE scheduling.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert "def clear_stale_app_scheduling(data):" in script assert 'cfg["nodeSelector"] = {}' in script assert 'cfg["affinity"] = {}' in script assert 'cfg["topologySpreadConstraints"] = []' in script assert "clear_stale_app_scheduling(app_values)" in script def test_supabase_deploy_runtime_constraint_cleanup_clears_topology_spread_constraints(): """Runtime stale-constraint cleanup must clear topology spread constraints in addition to node selectors/affinity.""" script = (REPO_ROOT / "supabase" / "deploy.sh").read_text(encoding="utf-8") assert '"topologySpreadConstraints":null' in script def test_deploy_entrypoint_supabase_storageclass_preflight_is_advisory_not_blocking(): """Top-level deploy preflight must not hard-exit before Supabase can reconcile/create its GKE StorageClass.""" script = (REPO_ROOT / "deploy.sh").read_text(encoding="utf-8") assert "storage-class preflight advisory" in script assert "Supabase deploy will create/reconcile it in this run" in script assert "Required StorageClass" not in script def test_init_gitlab_adds_noop_rerun_fast_path_guards(): """GitLab init must include fast-path guards to skip heavy rerun work when operator/CR are unchanged.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert 'GITLAB_OPERATOR_CHART_VERSION="${GITLAB_OPERATOR_CHART_VERSION:-}"' in script assert "resolve_helm_release_chart_version()" in script assert "resolve_operator_watch_namespace()" in script assert "kubectl_apply_reports_changed()" in script assert "operator_requires_upgrade=1" in script assert "Reusing installed GitLab chart version from existing GitLab CR" in script assert "gitlab_apply_output=" in script assert "gitlab_reconcile_required=0" in script assert "Fast-path: no GitLab reconcile triggers detected; skipping long wait." in script assert "Short health check OK: GitLab CR condition Available=True." in script assert "gitlab_generation_before=" in script assert "gitlab_generation_after=" in script def test_init_gitlab_enforces_authoritative_gitaly_storageclass_fields_in_rendered_cr(): """GitLab CR render path must explicitly set all authoritative Gitaly storageClass fields to prevent fallback drift.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "gitlab.gitaly.persistence.storageClass" in script assert "gitaly.persistence.storageClass" in script assert "storageClass: ${GITALY_STORAGE_CLASS}" in script def test_init_gitlab_uses_single_frontdoor_owner_model_for_k8s(): """k8s/GKE GitLab deploy must default to one front-door owner and clean stale fallback ingress.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert 'GITLAB_FRONTDOOR_OWNER="${GITLAB_FRONTDOOR_OWNER:-$_default_gitlab_frontdoor_owner}"' in script assert '_default_gitlab_frontdoor_owner="fallback"' in script assert "GITLAB_WEBSERVICE_INGRESS_ENABLED=false" in script assert "deleting stale operator ingress" in script def test_init_gitlab_renders_ingress_class_from_config_without_hardcoded_kong(): """GitLab ingress class rendering must follow configured ingress class consistently.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "class: ${GITLAB_INGRESS_CLASS}" in script assert "kubernetes.io/ingress.class: ${GITLAB_INGRESS_CLASS}" in script assert " ingress:\n class: kong" not in script assert "kubernetes.io/ingress.class: kong" not in script def test_init_gitlab_renders_non_empty_trusted_proxies_for_forwarded_headers(): """GitLab CR render path must set trusted_proxies so forwarded host/proto are accepted behind ingress/LB.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert 'GITLAB_TRUSTED_PROXIES_RAW="${GITLAB_TRUSTED_PROXIES:-${_default_gitlab_trusted_proxies}}"' in script assert "GitLab trusted_proxies configured:" in script assert "trusted_proxies:${GITLAB_TRUSTED_PROXIES_YAML}" in script def test_init_gitlab_split_cluster_ownership_is_explicit_and_db_app_workloads_blocked(): """Split-cluster GitLab ownership must be explicit: APP owns app workloads; DB app workloads are blocked.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "gitlab_split_cluster_ownership_diagnostics()" in script assert "GitLab split-cluster ownership policy: APP context" in script assert "DB context '${db_ctx}' must not host GitLab app workloads" in script assert "GitLab split-cluster ownership violation: DB cluster contains GitLab app workloads" in script assert "gitlab_split_cluster_ownership_diagnostics" in script def test_init_gitlab_convergence_no_longer_blocks_on_historical_restart_count_only(): """Convergence gate must not fail solely on historical restartCount for currently healthy webservice pods.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "GITLAB_WEBSERVICE_NOT_READY_BLOCK_SECONDS" in script assert "readyReplicas=" in script assert "availableReplicas=" in script assert "restartCount=" not in script assert "gitlab_webservice_blocked_reasons \"$dep_selector\" \"$restart_threshold\"" not in script def test_init_gitlab_old_replicaset_summary_helper_is_non_fatal_when_empty(): """ReplicaSet summary helper should not return non-zero on healthy empty summary under set -e.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "gitlab_old_replicaset_live_summary()" in script assert 'if [[ -n "$summary" ]]; then' in script assert "return 0" in script def test_init_gitlab_blocks_wrong_live_gitaly_claim_template_storageclass(): """Live Gitaly StatefulSet repo-data claim-template class mismatches must be hard-blocked unless destructive repair is enabled.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "GitLab Gitaly StatefulSet repo-data claim-template storageClass mismatch" in script assert "live.volumeClaimTemplate.repo-data.storageClassName=" in script assert "GITLAB_REPAIR_BLOCKED_AUTOCLEAN=1" in script assert "cleanup_gitlab_wrong_gitaly_template_storage" in script def test_init_gitlab_sanitizes_removed_chart9_top_level_replica_keys_before_apply(): """GitLab chart v9 path must sanitize only removed top-level registry min/max replica keys before CR apply.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "gitlab_chart9_find_deprecated_top_level_replica_keys" in script assert "gitlab_chart9_strip_deprecated_top_level_replica_keys" in script assert "sanitize_gitlab_cr_rendered_values_for_chart_version" in script assert "preflight_validate_gitlab_cr_rendered_values" in script assert 'if (line ~ /^ registry:[[:space:]]*$/) {' in script assert 'GITLAB_CR_RENDERED="$(sanitize_gitlab_cr_rendered_values_for_chart_version "$GITLAB_CR_RENDERED")"' in script assert 'preflight_validate_gitlab_cr_rendered_values "$GITLAB_CR_RENDERED"' in script def test_init_gitlab_rendered_registry_uses_hpa_replica_fields_without_legacy_top_level_keys(): """Rendered GitLab registry values must keep hpa.* replicas and avoid removed top-level min/max fields.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert " registry:" in script assert " hpa:" in script assert " minReplicas: 1" in script assert " maxReplicas: 1" in script assert " minReplicas: 1\n maxReplicas: 1\n hpa:" not in script def test_init_gitlab_rendered_shell_kas_sidekiq_webservice_use_chart9_authoritative_replica_fields(): """Chart 9.10.3 authoritative replica fields must be rendered per-component for operator-managed GitLab workloads.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert " webservice:" in script assert " replicaCount: 1" in script assert " minReplicas: 1" in script assert " maxReplicas: 1" in script assert " sidekiq:" in script assert " minReplicas: 1" in script assert " maxReplicas: 1" in script assert " sidekiq:\n replicaCount: 1\n hpa:" not in script assert " gitlab-shell:" in script assert " kas:" in script assert " gitlab-shell:\n replicaCount: 1\n hpa:" not in script assert " kas:\n replicaCount: 1\n hpa:" not in script def test_init_gitlab_replica_source_of_truth_failure_logs_rendered_cr_replica_fields(): """Replica source-of-truth mismatch failures should print rendered CR replica field diagnostics.""" script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") assert "gitlab_rendered_replica_source_fields_from_cr" in script assert "Rendered GitLab CR replica source fields:" in script def test_gitlab_and_gitea_init_storage_node_defaults_are_config_driven(): """GitLab/Gitea init scripts must not hardcode physical host defaults for storage node pinning.""" gitlab_script = (REPO_ROOT / "etc" / "init_gitlab.sh").read_text(encoding="utf-8") gitea_script = (REPO_ROOT / "etc" / "init_gitea.sh").read_text(encoding="utf-8") assert "GITLAB_STORAGE_NODE:-gandalf.knoe.org" not in gitlab_script assert "GITEA_NODE_SELECTOR:-gandalf.knoe.org" not in gitea_script assert "gandalf.knoe.org" not in gitlab_script assert "gandalf.knoe.org" not in gitea_script assert 'NODE_SELECTOR_KEY="${GITLAB_NODE_SELECTOR_KEY:-${NODE_SELECTOR_KEY:-kubernetes.io/hostname}}"' in gitlab_script assert 'NODE_SELECTOR_KEY="${GITEA_NODE_SELECTOR_KEY:-${NODE_SELECTOR_KEY:-kubernetes.io/hostname}}"' in gitea_script def test_ui_screens_do_not_hardcode_physical_hosts(): """UI screen init/base modules must not hardcode physical node hostnames.""" screens_init = (REPO_ROOT / "knoe" / "ui" / "screens" / "__init__.py").read_text( encoding="utf-8" ) screens_base = (REPO_ROOT / "knoe" / "ui" / "screens" / "base.py").read_text( encoding="utf-8" ) assert "gandalf.knoe.org" not in screens_init assert "gandalf.knoe.org" not in screens_base assert "myrddin.knoe.org" not in screens_base def test_init_kong_allows_disabling_svc_ingress_tls_for_k8s_path(): """etc/init_kong.sh must support non-TLS svc ingress rendering for k8s/GKE.""" script = (REPO_ROOT / "etc" / "init_kong.sh").read_text(encoding="utf-8") assert 'SERVICE_INGRESS_TLS_ENABLED="${SERVICE_INGRESS_TLS_ENABLED:-0}"' in script assert 'SERVICE_INGRESS_TLS_ENABLED="${SERVICE_INGRESS_TLS_ENABLED:-1}"' in script assert 'if is_truthy "${SERVICE_INGRESS_TLS_ENABLED:-0}"; then' in script assert "Rendering svc ingress without TLS" in script assert 'SERVICE_MANAGED_CERT_NAME:-svc-knoe-managed-cert' in script assert 'SERVICE_FRONTEND_CONFIG_NAME:-svc-knoe-frontend-config' in script assert 'networking.gke.io/managed-certificates: ${service_managed_cert_name}' in script assert 'networking.gke.io/v1beta1.FrontendConfig: ${service_frontend_config_name}' in script assert "SERVICE_PRE_SHARED_CERT is ignored" in script assert "ingress.gcp.kubernetes.io/pre-shared-cert-" in script def test_deploy_frontdoor_validation_tracks_tls_path_states_and_https_probe(): """deploy.sh diagnostics must distinguish missing TLS path from attached-but-not-serving and require HTTPS probing.""" script = (REPO_ROOT / "deploy.sh").read_text(encoding="utf-8") assert "MISSING_TLS_PATH" in script assert "NO_TLS_PATH_CONFIGURED" in script assert "TLS_PATH_ATTACHED_BUT_NOT_SERVING" in script assert "CERT_NOT_READY" in script assert "TLS_HANDSHAKE_BROKEN" in script assert "BACKEND_UNHEALTHY" in script assert "def _probe_https_endpoint" in script assert "def _classify_tls_failure" in script def test_reset_k3s_namespace_delete_db_deletes_pvcs_and_bound_pvs(tmp_path, monkeypatch): """DB delete reset must request PVC deletion and force cleanup of bound PVs.""" import subprocess as sp from knoe.core import actions as actions_mod script = tmp_path / "scripts" / "reset-ns.sh" script.parent.mkdir(parents=True, exist_ok=True) script.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8") calls: list[list[str]] = [] def _fake_run(cmd, **kwargs): calls.append(list(cmd)) if cmd[:6] == ["kubectl", "-n", "knoe-db", "get", "pvc", "-o"]: return sp.CompletedProcess(cmd, 0, stdout="pv-a\npv-b\n", stderr="") return sp.CompletedProcess(cmd, 0, stdout="", stderr="") monkeypatch.setattr(actions_mod.subprocess, "run", _fake_run) actions_mod._reset_k3s_namespace( tmp_path, ["knoe-db"], "", "", delete_pvcs_namespaces=["knoe-db"], force_delete_bound_pvs=True, ) assert any( cmd[:3] == ["bash", str(script), "-n"] and "--delete-pvcs" in cmd for cmd in calls ) assert ["kubectl", "patch", "pv", "pv-a", "--type=merge", "-p", '{"spec":{"persistentVolumeReclaimPolicy":"Delete"}}'] in calls assert ["kubectl", "patch", "pv", "pv-b", "--type=merge", "-p", '{"spec":{"persistentVolumeReclaimPolicy":"Delete"}}'] in calls assert ["kubectl", "delete", "pv", "pv-a", "--ignore-not-found", "--wait=false"] in calls assert ["kubectl", "delete", "pv", "pv-b", "--ignore-not-found", "--wait=false"] in calls