mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 17:44:33 +00:00
c0a7d0c5dc
- improve installer/action/controller flow and shell-variable expansion handling across screens\n- adjust Supabase Helm rendering and storage deployment templates\n- align monitoring, cloudnative-pg and repair pipeline behavior with updated config paths\n- refresh and expand installer/core regression tests around milestones, navigation and repair logic Co-authored-by: Junie <junie@jetbrains.com>
592 lines
22 KiB
Python
592 lines
22 KiB
Python
"""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 prole 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("prole_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 / "prole.cfg"
|
|
_write_cfg(
|
|
cfg_path,
|
|
f"""
|
|
[Inputs]
|
|
init_password.db_password = test-password
|
|
init_password.db_host_port = 5432
|
|
{extra}
|
|
[Global]
|
|
NAMESPACE = test-ns
|
|
""",
|
|
)
|
|
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 / "prole.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.prole_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 / "prole.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_ansible_vault_credentials", 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 / "prole.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 / "prole.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 / "prole.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_ansible_vault_credentials", 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.prole.org"
|
|
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.prole.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.prole.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.prole.org,db.prole.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.prole.org" in host_names, f"Expected supabase.prole.org in {host_names}"
|
|
assert "db.prole.org" in host_names, f"Expected db.prole.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.prole.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
|
|
|
|
|
|
# ===========================================================================
|
|
# 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.<component>.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 "DROP TABLE IF EXISTS public.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 "ALTER EXTENSION %I SET SCHEMA knoe" in script
|
|
assert "ALTER TABLE public.spatial_ref_sys ENABLE ROW LEVEL SECURITY;" in script
|
|
assert "CREATE POLICY spatial_ref_sys_select_all ON public.spatial_ref_sys" 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
|