mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 11:03:59 +00:00
311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""
|
|
Unit tests for installer/core/controller.py
|
|
|
|
Covers KnoeController: check_docker_running, get_knoe_db_version, run_script.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from knoe.core.controller import KnoeController
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_controller(tmp_path: Path, **kwargs) -> KnoeController:
|
|
return KnoeController(project_root=tmp_path, **kwargs)
|
|
|
|
|
|
def _scaffold_project(tmp_path: Path) -> None:
|
|
"""Create minimum project structure expected by run_script."""
|
|
(tmp_path / "etc").mkdir(parents=True, exist_ok=True)
|
|
(tmp_path / "conf" / "postgresql").mkdir(parents=True, exist_ok=True)
|
|
(tmp_path / "knoe-db").mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# check_docker_running
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_check_docker_running_success(tmp_path):
|
|
ctrl = _make_controller(tmp_path)
|
|
with patch("knoe.core.controller.subprocess.run") as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
assert ctrl.check_docker_running() is True
|
|
|
|
|
|
def test_check_docker_running_called_process_error(tmp_path):
|
|
ctrl = _make_controller(tmp_path)
|
|
with patch("knoe.core.controller.subprocess.run",
|
|
side_effect=subprocess.CalledProcessError(1, "docker")):
|
|
assert ctrl.check_docker_running() is False
|
|
|
|
|
|
def test_check_docker_running_file_not_found(tmp_path):
|
|
ctrl = _make_controller(tmp_path)
|
|
with patch("knoe.core.controller.subprocess.run",
|
|
side_effect=FileNotFoundError("docker not found")):
|
|
assert ctrl.check_docker_running() is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_knoe_db_version
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_get_knoe_db_version_no_files(tmp_path):
|
|
ctrl = _make_controller(tmp_path)
|
|
version = ctrl.get_knoe_db_version()
|
|
# Falls back to defaults: 17.7 and 43 (zero-padded to 3 digits → 043)
|
|
assert version == "17.7-043"
|
|
|
|
|
|
def test_get_knoe_db_version_with_files(tmp_path):
|
|
(tmp_path / "conf" / "postgresql").mkdir(parents=True)
|
|
(tmp_path / "conf" / "postgresql" / ".version").write_text("18\n")
|
|
(tmp_path / "knoe-db").mkdir(parents=True)
|
|
(tmp_path / "knoe-db" / ".version").write_text("88\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
version = ctrl.get_knoe_db_version()
|
|
assert version == "18-088"
|
|
|
|
|
|
def test_get_knoe_db_version_non_numeric_release(tmp_path):
|
|
(tmp_path / "conf" / "postgresql").mkdir(parents=True)
|
|
(tmp_path / "conf" / "postgresql" / ".version").write_text("18\n")
|
|
(tmp_path / "knoe-db").mkdir(parents=True)
|
|
(tmp_path / "knoe-db" / ".version").write_text("latest\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
version = ctrl.get_knoe_db_version()
|
|
# Non-numeric release is not zero-padded
|
|
assert version == "18-latest"
|
|
|
|
|
|
def test_get_knoe_db_version_only_pg_version(tmp_path):
|
|
(tmp_path / "conf" / "postgresql").mkdir(parents=True)
|
|
(tmp_path / "conf" / "postgresql" / ".version").write_text("17\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
version = ctrl.get_knoe_db_version()
|
|
assert version.startswith("17-")
|
|
|
|
|
|
def test_get_knoe_db_version_pads_single_digit_release(tmp_path):
|
|
(tmp_path / "knoe-db").mkdir(parents=True)
|
|
(tmp_path / "knoe-db" / ".version").write_text("5\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
version = ctrl.get_knoe_db_version()
|
|
assert version.endswith("-005")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_script
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_run_script_basic(tmp_path):
|
|
"""run_script executes a bash script and captures stdout via on_line."""
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "hello.sh"
|
|
script.write_text("#!/bin/bash\necho 'hello from script'\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
lines: list[str] = []
|
|
rc = ctrl.run_script(
|
|
"hello.sh",
|
|
env={"KNOE_HOME": str(tmp_path)},
|
|
on_line=lambda l: lines.append(l),
|
|
)
|
|
assert rc == 0
|
|
assert any("hello from script" in l for l in lines)
|
|
|
|
|
|
def test_run_script_nonzero_exit(tmp_path):
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "fail.sh"
|
|
script.write_text("#!/bin/bash\nexit 42\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
rc = ctrl.run_script("fail.sh", env={"KNOE_HOME": str(tmp_path)})
|
|
assert rc == 42
|
|
|
|
|
|
def test_run_script_with_args(tmp_path):
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "args.sh"
|
|
script.write_text("#!/bin/bash\necho \"ARG:$1\"\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
lines: list[str] = []
|
|
rc = ctrl.run_script(
|
|
"args.sh",
|
|
args=["myvalue"],
|
|
env={"KNOE_HOME": str(tmp_path)},
|
|
on_line=lambda l: lines.append(l),
|
|
)
|
|
assert rc == 0
|
|
assert any("ARG:myvalue" in l for l in lines)
|
|
|
|
|
|
def test_run_script_copies_k8s_and_conf(tmp_path):
|
|
"""Covers the k8s sync path and the minimal conf allowlist sync."""
|
|
_scaffold_project(tmp_path)
|
|
# Create source k8s and conf dirs
|
|
(tmp_path / "k8s").mkdir()
|
|
(tmp_path / "k8s" / "test.yaml").write_text("kind: Pod\n")
|
|
(tmp_path / "conf" / "postgresql" / ".version").write_text("18\n")
|
|
(tmp_path / "conf" / "database_versions.json").write_text("{}\n")
|
|
(tmp_path / "conf" / "port-mapping.cfg").write_text("; test\n")
|
|
script = tmp_path / "etc" / "synced.sh"
|
|
script.write_text("#!/bin/bash\necho ok\n")
|
|
target = tmp_path / "knoe_home_test"
|
|
target.mkdir()
|
|
ctrl = _make_controller(tmp_path)
|
|
rc = ctrl.run_script("synced.sh", env={"KNOE_HOME": str(target)})
|
|
assert rc == 0
|
|
# k8s should have been synced
|
|
assert (target / "k8s").exists()
|
|
# conf allowlist should have been synced (but NOT knoe.cfg)
|
|
assert (target / "conf" / "postgresql" / ".version").read_text() == "18\n"
|
|
assert (target / "conf" / "database_versions.json").exists()
|
|
assert (target / "conf" / "port-mapping.cfg").exists()
|
|
assert not (target / "conf" / "knoe.cfg").exists()
|
|
|
|
|
|
def test_run_script_preserves_existing_knoe_cfg_symlink_entrypoint(tmp_path: Path):
|
|
"""Regression: run_script must never delete/overwrite curated $KNOE_HOME/conf/knoe.cfg layouts."""
|
|
|
|
_scaffold_project(tmp_path)
|
|
|
|
# Create a simple script to run.
|
|
script = tmp_path / "etc" / "ok.sh"
|
|
script.write_text("#!/bin/bash\necho ok\n")
|
|
|
|
# Ensure the source project has a knoe.cfg file; destructive sync currently overwrites
|
|
# the curated symlink entrypoint.
|
|
(tmp_path / "conf" / "knoe.cfg").write_text("SOURCE_ENTRY\n")
|
|
|
|
# Create a curated KNOE_HOME/conf layout with a symlink entrypoint.
|
|
knoe_home = tmp_path / "knoe_home_curated"
|
|
(knoe_home / "conf" / "service").mkdir(parents=True, exist_ok=True)
|
|
service_base = knoe_home / "conf" / "service" / "knoe.cfg"
|
|
service_base.write_text("SERVICE_BASE\n")
|
|
entry = knoe_home / "conf" / "knoe.cfg"
|
|
entry.symlink_to(Path("service") / "knoe.cfg")
|
|
|
|
# Ensure the destructive sync path triggers (source conf newer than target conf).
|
|
(tmp_path / "conf" / "__touch__.txt").write_text("x\n")
|
|
|
|
ctrl = _make_controller(tmp_path)
|
|
rc = ctrl.run_script("ok.sh", env={"KNOE_HOME": str(knoe_home)})
|
|
assert rc == 0
|
|
|
|
assert entry.exists()
|
|
assert entry.is_symlink()
|
|
assert service_base.exists()
|
|
assert service_base.read_text() == "SERVICE_BASE\n"
|
|
|
|
|
|
def test_run_script_copies_etc_lib_shell_for_etc_scripts(tmp_path):
|
|
"""Regression: etc/ scripts may source etc/lib/shell/common_core_lib.sh."""
|
|
_scaffold_project(tmp_path)
|
|
|
|
(tmp_path / "etc" / "lib" / "shell").mkdir(parents=True, exist_ok=True)
|
|
(tmp_path / "etc" / "lib" / "shell" / "common_core_lib.sh").write_text(
|
|
"#!/usr/bin/env bash\ncommon_core_preparse_config() { :; }\n"
|
|
)
|
|
|
|
script = tmp_path / "etc" / "needs_lib.sh"
|
|
script.write_text(
|
|
"#!/bin/bash\n"
|
|
"SCRIPT_DIR=$(cd \"$(dirname \"${BASH_SOURCE[0]}\")\" && pwd)\n"
|
|
"# shellcheck disable=SC1091\n"
|
|
"source \"$SCRIPT_DIR/lib/shell/common_core_lib.sh\"\n"
|
|
"common_core_preparse_config\n"
|
|
"echo ok\n"
|
|
)
|
|
|
|
target = tmp_path / "knoe_home_lib"
|
|
target.mkdir()
|
|
|
|
ctrl = _make_controller(tmp_path)
|
|
lines: list[str] = []
|
|
rc = ctrl.run_script(
|
|
"needs_lib.sh",
|
|
env={"KNOE_HOME": str(target)},
|
|
on_line=lambda l: lines.append(l),
|
|
)
|
|
assert rc == 0
|
|
assert any("ok" in l for l in lines)
|
|
assert (target / "etc" / "lib" / "shell" / "common_core_lib.sh").exists()
|
|
|
|
|
|
def test_run_script_copies_db_version(tmp_path):
|
|
"""Covers the knoe-db/.version copy path."""
|
|
_scaffold_project(tmp_path)
|
|
(tmp_path / "knoe-db" / ".version").write_text("88\n")
|
|
script = tmp_path / "etc" / "dbver.sh"
|
|
script.write_text("#!/bin/bash\necho ok\n")
|
|
target = tmp_path / "knoe_home_db"
|
|
target.mkdir()
|
|
ctrl = _make_controller(tmp_path)
|
|
rc = ctrl.run_script("dbver.sh", env={"KNOE_HOME": str(target)})
|
|
assert rc == 0
|
|
assert (target / "knoe-db" / ".version").read_text().strip() == "88"
|
|
|
|
|
|
def test_run_script_default_knoe_home(tmp_path):
|
|
"""Covers the KNOE_HOME fallback to ~/.knoe when not in env."""
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "home.sh"
|
|
script.write_text("#!/bin/bash\necho hometest\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
# Provide env without KNOE_HOME so it falls back to Path.home() / ".knoe"
|
|
fake_home = tmp_path / "fakehome"
|
|
fake_home.mkdir()
|
|
with patch("knoe.core.controller.Path.home", return_value=fake_home):
|
|
rc = ctrl.run_script("home.sh", env={})
|
|
assert rc == 0
|
|
|
|
|
|
def test_run_script_stderr_to_stdout_false(tmp_path):
|
|
"""Covers the stderr_to_stdout=False thread path."""
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "stderr.sh"
|
|
script.write_text("#!/bin/bash\necho 'out'\n>&2 echo 'err'\n")
|
|
ctrl = _make_controller(tmp_path)
|
|
stderr_lines: list[str] = []
|
|
rc = ctrl.run_script(
|
|
"stderr.sh",
|
|
env={"KNOE_HOME": str(tmp_path)},
|
|
stderr_to_stdout=False,
|
|
on_stderr_line=lambda l: stderr_lines.append(l),
|
|
)
|
|
assert rc == 0
|
|
assert any("err" in l for l in stderr_lines)
|
|
|
|
|
|
def test_run_script_verbose_sets_env(tmp_path):
|
|
"""Covers the verbose env var injection path."""
|
|
_scaffold_project(tmp_path)
|
|
script = tmp_path / "etc" / "verbose.sh"
|
|
script.write_text("#!/bin/bash\necho \"VERBOSE:${PROLE_VERBOSE:-unset}\"\n")
|
|
ctrl = _make_controller(tmp_path, verbose=True)
|
|
lines: list[str] = []
|
|
rc = ctrl.run_script(
|
|
"verbose.sh",
|
|
env={"KNOE_HOME": str(tmp_path)},
|
|
on_line=lambda l: lines.append(l),
|
|
)
|
|
assert rc == 0
|
|
assert any("VERBOSE:1" in l for l in lines)
|
|
|
|
|
|
def test_run_script_missing_source_script(tmp_path):
|
|
"""If the source script doesn't exist, target script also won't exist — graceful."""
|
|
_scaffold_project(tmp_path)
|
|
# Don't create the script — run_script should still try to run it
|
|
ctrl = _make_controller(tmp_path)
|
|
# Should return non-zero (bash can't find the file) but not raise
|
|
rc = ctrl.run_script("nonexistent.sh", env={"KNOE_HOME": str(tmp_path)})
|
|
assert rc != 0
|