prole/tests/test_navigation.py
chrisfu a069989315 Rename prole-db to knoe-db, add knoe-auth as cluster-internal KDC
Itemized changes:

1. knoe-auth: New cluster-internal KDC and SSO gateway service
   - Created etc/init_knoe_auth.sh based on init_kdc.sh with knoe-auth naming
   - Namespace defaults to SERVICE_NAMESPACE (knoe-system)
   - ConfigMap: knoe-auth-kdc-config, Secret: knoe-auth-secrets
   - Legacy cleanup removes old auth/dog/authority deployments

2. Orchestration: knoe-auth initializes before CloudNativePG
   - Updated prole.sh to insert init_knoe_auth.sh as step 2 (before CNPG)
   - Renumbered all subsequent initialization steps

3. Kong routing: Updated init_kong.sh to route to knoe-auth in SERVICE_NAMESPACE

4. Comment/reference updates for knoe-auth
   - Updated init_common_services.sh, init_service_layer.sh, init_kerberos.sh

5. prole-db renamed to knoe-db across the entire codebase
   - Renamed prole-db/ directory to knoe-db/
   - Renamed all prole-db Kubernetes manifests (deploy/opentofu, k8s/)
   - Renamed scripts: docker-root-knoe-db.sh, docker-run-knoe-db.sh, test-cnpg-knoe-db.sh
   - Renamed etc/init_prole-db-reset.sh to etc/init_knoe-db-reset.sh
   - Renamed etc/prole-db-passwwd.sh to etc/knoe-db-passwwd.sh
   - Renamed mock_val counterparts accordingly
   - Renamed tests/etc/test_init_prole-db-reset.sh to test_init_knoe-db-reset.sh
   - Renamed docs/prole-db-documentation-mcp-architecture.md to knoe-db variant
   - Renamed modes/k3d/prole-db/ to modes/k3d/knoe-db/
   - Renamed prole-db.iml to knoe-db.iml

6. Configuration updates
   - Updated conf/dev, conf/prod, conf/test, conf/service prole.cfg files
   - Updated conf/port-mapping.cfg
   - Updated etc/prole_cfg.sh and mock_val/prole_cfg.sh
   - Updated service/prole.cfg

7. Kubernetes manifests and deploy configuration
   - Updated deploy/opentofu/k3s ArgoCD application YAMLs
   - Updated kong-configmap.yaml and kustomization.yaml
   - Updated k3s/kong-config.yml and prole-resources.yaml
   - Updated prole-mssql-db deployment YAMLs
   - Updated supabase helm render and deploy scripts

8. Infrastructure and GCP Terraform
   - Updated deploy/gcp/terraform: folders, groups, IAM, service-projects

9. Python/installer code updates
   - Updated knoe/core: actions, build_context, controller, env, milestones
   - Updated knoe/milestone.py
   - Updated knoe/ui/screens: cfg, database, database_options, deploy, docker,
     navigation, security, services, validate
   - Updated knoe.spec, status.py

10. Shell script updates
    - Updated etc/: build_db, init_cloudnative_pg, init_cnpg_backup,
      init_db_manager, init_forgejo, init_gitlab, init_monitoring, init_openbao,
      init_port_forwards, init_postgrest, init_supabase_ports, status
    - Updated mock_val/ counterparts for all above scripts
    - Updated prole-net/init-prole-dns.sh
    - Updated bin/prole-kpf.sh, gitea/deploy.sh, supabase/deploy.sh

11. Test updates
    - Updated tests/etc/: test_init_cloudnative_pg*, test_init_cnpg_backup*,
      test_init_kdc*, test_init_kerberos*, test_init_kong*, test_prole_cfg*
    - Updated tests/installer/: test_actions_helpers, test_cfg_save_kubecontext,
      test_controller, test_core_classes, test_milestones, test_milestones_extended,
      test_namespace_propagation
    - Updated tests/: test_database_options, test_navigation,
      test_render_supabase_hostname, test_docker_build_fix,
      test_all_prole_home_fixes, silent_install_test, final_test

12. Documentation updates
    - Updated docs/: DOCKER-BUILD-FIX, PROLE-CFG-SECRETS, PROLE-HOME-DIRECTORY,
      build-system, patent
    - Updated scan/network_description.txt
    - Updated pom.xml

13. Miscellaneous script updates
    - Updated root-level: _adopt_replica_pvcs, _fix_replica_merlin, _import_pi,
      _patch_cluster, _prebind_pvcs, _rebind_d002, _rebind_d002b, test_resolve
    - Updated scripts/generate_spec.py

Co-authored-by: Junie <junie@jetbrains.com>
2026-03-22 22:16:21 -07:00

251 lines
9.5 KiB
Python

import sys
from unittest.mock import patch, MagicMock
import pytest
from pathlib import Path
# Mock tkinter and other GUI/macOS specific imports
sys.modules["tkinter"] = MagicMock()
sys.modules["tkinter.ttk"] = MagicMock()
sys.modules["tkinter.scrolledtext"] = MagicMock()
sys.modules["tkinter.messagebox"] = MagicMock()
sys.modules["tkinter.filedialog"] = MagicMock()
sys.modules["Foundation"] = MagicMock()
sys.modules["AppKit"] = MagicMock()
sys.modules["PIL"] = MagicMock()
sys.modules["PIL.Image"] = MagicMock()
sys.modules["PIL.ImageTk"] = MagicMock()
import install
from install import ProleInstaller
@pytest.fixture
def mock_installer(tmp_path):
test_cfg_src = Path(__file__).parent / "fixtures" / "test-prole.cfg"
cfg_path = tmp_path / "prole.cfg"
cfg_path.write_text(test_cfg_src.read_text())
with patch("install.tk.Tk"), patch("install.tk.Frame"), patch(
"install.tk.Canvas"
), patch("install.tk.Label"), patch("install.tk.Button"), patch(
"install.tk.Entry"
) as mock_entry, patch(
"install.ttk.Style"
), patch(
"install.Path.exists", return_value=True
), patch(
"install.Path.mkdir"
), patch.object(
ProleInstaller, "_save_env_to_file"
), patch.object(
ProleInstaller, "_save_prole_cfg"
), patch.object(
ProleInstaller, "_save_ansible_prole_vault"
), patch.object(
ProleInstaller, "_after_env_saved"
), patch(
"threading.Thread"
):
mock_entry.return_value.get.return_value.strip.return_value = "mock_val"
root = MagicMock()
# Mock winfo methods needed for centering
root.winfo_screenwidth.return_value = 1920
root.winfo_screenheight.return_value = 1080
app = ProleInstaller(root, config_path=str(cfg_path))
return app
def test_initial_page(mock_installer):
# Should start at index 0 (welcome)
assert mock_installer.page_index == 0
assert mock_installer.pages[mock_installer.page_index][0] == "welcome"
def test_show_page_by_id(mock_installer):
mock_installer.show_page("network_scan")
assert mock_installer.pages[mock_installer.page_index][0] == "network_scan"
def test_show_page_invalid_id(mock_installer):
# Current page is welcome (0)
mock_installer.page_index = 0
with patch("install.messagebox.showerror") as mock_error:
mock_installer.show_page("non_existent_page")
# Should stay at current index and NOT fallback to 0 if it was elsewhere,
# or just stay at 0 if it was at 0.
# The key fix was NOT defaulting to 0 when index_or_id is a string and not found.
assert mock_installer.page_index == 0
mock_error.assert_called_once()
def test_navigation_flow_standard(mock_installer):
# Welcome -> Dependencies
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "deps_summary"
# Mock all dependencies installed to skip dep pages
with patch.object(mock_installer, "all_dependencies_installed", return_value=True):
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "network_scan"
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "env_setup"
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "init_cluster"
with patch.object(
mock_installer, "_cluster_ready_for_navigation", return_value=True
):
mock_installer._common_services_success = True
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "database_options"
with patch.object(
mock_installer, "_generate_knoe_db_dockerfile", return_value=True
):
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "init_db_build"
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "init_password"
def test_ollama_screen_removed_and_supabase_flows_to_deploy(mock_installer):
# Ensure the standalone Ollama screen is no longer part of the GUI page set/nav.
page_ids = [pid for pid, _ in mock_installer.pages]
assert "ollama_config" not in page_ids
assert all(pid != "ollama_config" for _, pid in mock_installer.nav_items)
# Ensure Next from Supabase goes directly to Deployment.
mock_installer.prole_cfg_data.setdefault("Optional Features", {})
supabase_idx = next(i for i, (pid, _) in enumerate(mock_installer.pages) if pid == "supabase_config")
mock_installer.page_index = supabase_idx
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "init_cnpg_deploy"
def test_navigation_flow_missing_deps(mock_installer):
# Welcome -> Dependencies
mock_installer.show_page("welcome")
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "deps_summary"
# Mock one dependency missing
mock_dep_id = mock_installer.dependencies[0]["id"]
with patch.object(
mock_installer, "all_dependencies_installed", return_value=False
), patch.object(
mock_installer, "_dep_navigation_sequence", return_value=[f"dep_{mock_dep_id}"]
):
mock_installer.on_next()
assert (
mock_installer.pages[mock_installer.page_index][0] == f"dep_{mock_dep_id}"
)
# Next from dep page with all installed now
with patch.object(
mock_installer, "all_dependencies_installed", return_value=True
):
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "network_scan"
def test_on_prev(mock_installer):
# init_password -> init_db_build -> database_options
mock_installer.show_page("init_password")
mock_installer.on_prev()
assert mock_installer.pages[mock_installer.page_index][0] == "init_db_build"
mock_installer.on_prev()
assert mock_installer.pages[mock_installer.page_index][0] == "database_options"
# database_options -> init_cluster
mock_installer.on_prev()
assert mock_installer.pages[mock_installer.page_index][0] == "init_cluster"
# network_scan -> deps_summary
mock_installer.show_page("network_scan")
with patch.object(mock_installer, "all_dependencies_installed", return_value=True):
mock_installer.on_prev()
assert mock_installer.pages[mock_installer.page_index][0] == "deps_summary"
# deps_summary -> welcome
mock_installer.on_prev()
assert mock_installer.pages[mock_installer.page_index][0] == "welcome"
def test_infinite_loop_prevention(mock_installer):
# This specifically tests the scenario that caused the bug:
# trying to navigate to a missing ID should NOT reset to welcome (0)
mock_installer.show_page("network_scan")
current_idx = mock_installer.page_index
mock_installer.show_page("invalid_id")
assert mock_installer.page_index == current_idx
assert mock_installer.pages[mock_installer.page_index][0] == "network_scan"
def test_on_next_finish_loop(mock_installer):
# From last page, on_next should not crash or loop
last_idx = len(mock_installer.pages) - 1
mock_installer.show_page(last_idx)
mock_installer.on_next()
assert mock_installer.page_index == last_idx
def test_on_prev_first_page(mock_installer):
# From first page, on_prev should not crash or loop
mock_installer.show_page(0)
mock_installer.on_prev()
assert mock_installer.page_index == 0
def test_navigation_to_create_installer(mock_installer):
# From build page, if successful, next should go to create_installer
mock_installer.show_page("build")
mock_installer._built_success = True
mock_installer.on_next()
assert mock_installer.pages[mock_installer.page_index][0] == "create_installer"
# Last page Finish label
mock_installer.next_button = MagicMock()
mock_installer.prev_button = MagicMock()
mock_installer.update_footer()
mock_installer.next_button.configure.assert_any_call(text="Finish")
def test_all_pages_render(mock_installer):
with patch.object(
mock_installer, "_verify_k3s_services", return_value=None
), patch.object(
mock_installer, "check_cluster_status_async", return_value=None
), patch.object(
mock_installer, "_ensure_db_build_registry_async", return_value=None
), patch.object(
mock_installer, "_get_kubectx_list", return_value=["default"]
), patch.object(
mock_installer, "_apply_k3s_defaults", return_value=None
):
for page_id, _ in mock_installer.pages:
mock_installer.show_page(page_id)
# Exercise init_cluster conditional branches for service/prod modes
if hasattr(mock_installer.cluster_env, "get") and hasattr(
mock_installer.cluster_env.get, "return_value"
):
mock_installer.cluster_env.get.return_value = "prole-service-cluster"
else:
mock_installer.cluster_env.set("prole-service-cluster")
with patch.object(mock_installer, "_cluster_env_key", return_value="service"):
mock_installer.show_page("init_cluster")
if hasattr(mock_installer.cluster_env, "get") and hasattr(
mock_installer.cluster_env.get, "return_value"
):
mock_installer.cluster_env.get.return_value = "prole-prod-cluster"
else:
mock_installer.cluster_env.set("prole-prod-cluster")
with patch.object(mock_installer, "_cluster_env_key", return_value="prod"):
mock_installer.show_page("init_cluster")