import tkinter as tk from tkinter import ttk from pathlib import Path import sys import pytest # Ensure `installer` is importable when tests are invoked directly via `pytest` # (the project also provides `tests/run_tests.sh` which sets `PYTHONPATH=.`). PROJECT_ROOT = Path(__file__).resolve().parents[2] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from knoe.ui.screens.base import ScreenBaseMixin from knoe.ui.screens.cluster import ClusterScreenMixin class _DummyClusterApp(ScreenBaseMixin, ClusterScreenMixin): """Minimal Tk app surface needed to render the Cluster Environment screen.""" def __init__(self, root: tk.Tk, canvas: tk.Canvas, kubectx_values: list[str]): self.root = root self.bg_canvas = canvas self._canvas_items: list[int] = [] self._overlay_widgets: list[tk.Widget] = [] # Tk variables used by the screen self.cluster_env = tk.StringVar(master=root, value="prod") self.selected_k3d_cluster = tk.StringVar(master=root, value="") self.selected_kubectx = tk.StringVar(master=root, value="") self.service_namespace = tk.StringVar(master=root, value="knoe-system") self.prod_artifacts_path = tk.StringVar(master=root, value="/tmp") # GitOps / GWorkspace / GCloud variables added to prod screen self.gitops_provider = tk.StringVar(master=root, value="gitea") self.gworkspace_domain = tk.StringVar(master=root, value="") self.gworkspace_customer_id = tk.StringVar(master=root, value="") self.gworkspace_admin_email = tk.StringVar(master=root, value="") self.gcloud_project_id = tk.StringVar(master=root, value="") self.gcloud_region = tk.StringVar(master=root, value="us-central1") self.gcloud_sa_key_path = tk.StringVar(master=root, value="") # Dual-cluster GKE kubecontext selectors self.app_cluster_kubecontext = tk.StringVar(master=root, value="") self.db_cluster_kubecontext = tk.StringVar(master=root, value="") # Legacy variables still referenced by non-UI logic elsewhere self.k3s_server_url = tk.StringVar(master=root, value="") self.k3s_token = tk.StringVar(master=root, value="") self._kubectx_values = kubectx_values # Avoid invoking real `kubectl` during layout tests. self._kubectx_applied = kubectx_values[0] if kubectx_values else "" # ---- Stubs for callbacks / helpers invoked by rendering ---- def _on_cluster_env_change(self, *_args): return None def _get_k3d_cluster_list(self) -> list[str]: return ["knoe-dev"] def _on_create_k3d_cluster(self): return None def _on_delete_k3d_cluster(self): return None def _on_k3d_cluster_select(self, *_args): return None def _on_kubectx_select(self, *_args): return None def _switch_kubectx(self, _context_name: str): return True def _current_kubectl_context(self) -> str: return self._kubectx_applied def _get_kubectx_list(self) -> list[str]: return list(self._kubectx_values) def _validate_and_save_cluster_config(self): return None def _deploy_k3s_services(self): return None def _verify_k3s_services(self): return None def update_footer(self): return None def _canvas_texts(cnv: tk.Canvas) -> list[str]: texts: list[str] = [] for item in cnv.find_all(): if cnv.type(item) == "text": texts.append(cnv.itemcget(item, "text")) return texts @pytest.mark.parametrize("env", ["service"]) def test_cluster_environment_layout_large_mock_data(env: str): root = tk.Tk() root.withdraw() try: # Keep the canvas height realistic: the installer window is fixed-size and # there is a navigation footer consuming vertical space. canvas_height = 780 canvas = tk.Canvas(root, width=975, height=canvas_height, bg="white") canvas.pack(fill="both", expand=False) # In some headless/withdrawn runs, `winfo_height()` may remain `1` unless we # force a full update cycle. root.update() kubectx_values = [f"ctx-{i:03d}" for i in range(120)] app = _DummyClusterApp(root, canvas, kubectx_values=kubectx_values) app.cluster_env.set(env) app._render_init_cluster_page() root.update() # K3S connection details should not be rendered anymore (kubectx flow). assert "K3s Connection:" not in _canvas_texts(canvas) # Kubernetes context selector + description should be a single row. label_item = None for item in canvas.find_all(): if canvas.type(item) == "text" and canvas.itemcget(item, "text") == "Kubernetes Context:": label_item = item break assert label_item is not None label_y = canvas.coords(label_item)[1] combo_y = canvas.coords(app._kubectx_combo_canvas_window)[1] assert abs(combo_y - label_y) <= 10 # Common Core Services Namespace input should be a single row. # This control moved off the cluster screen. assert "Common Core Services Namespace:" not in _canvas_texts(canvas) app._render_common_services_page() root.update() ns_label_item = None for item in canvas.find_all(): if ( canvas.type(item) == "text" and canvas.itemcget(item, "text") == "Common Core Services Namespace:" ): ns_label_item = item break assert ns_label_item is not None ns_label_y = canvas.coords(ns_label_item)[1] ns_entry_y = canvas.coords(app._service_namespace_entry_canvas_window)[1] assert abs(ns_entry_y - ns_label_y) <= 10 finally: root.destroy() def test_cluster_environment_layout_prod_dual_context(): """Prod env renders App Cluster + DB Cluster dropdowns, not a single kubectx combo.""" root = tk.Tk() root.withdraw() try: canvas = tk.Canvas(root, width=975, height=780, bg="white") canvas.pack(fill="both", expand=False) root.update() kubectx_values = [f"ctx-{i:03d}" for i in range(5)] app = _DummyClusterApp(root, canvas, kubectx_values=kubectx_values) app.cluster_env.set("prod") app._render_init_cluster_page() root.update() texts = _canvas_texts(canvas) assert "K3s Connection:" not in texts assert "App Cluster:" in texts assert "DB Cluster:" in texts assert "Kubernetes Context:" not in texts assert hasattr(app, "_app_kubectx_combo") assert hasattr(app, "_db_kubectx_combo") assert hasattr(app, "_kubectx_apply_btn") assert app._kubectx_apply_btn.cget("state") == "disabled" finally: root.destroy() @pytest.mark.parametrize("env", ["dev", "service"]) def test_service_namespace_default_is_knoe_system_for_dev_and_service(env: str): root = tk.Tk() root.withdraw() try: canvas = tk.Canvas(root, width=975, height=780, bg="white") canvas.pack(fill="both", expand=False) root.update() app = _DummyClusterApp(root, canvas, kubectx_values=["ctx-001"]) app.cluster_env.set(env) app.service_namespace.set("") app.knoe_cfg_data = {} assert app._get_service_namespace() == "knoe-system" finally: root.destroy() def test_kubectx_selection_requires_apply(): root = tk.Tk() root.withdraw() try: canvas = tk.Canvas(root, width=975, height=780, bg="white") canvas.pack(fill="both", expand=False) root.update() class _KubectxApplyApp(ScreenBaseMixin, ClusterScreenMixin): def __init__(self, root: tk.Tk, canvas: tk.Canvas): self.root = root self.bg_canvas = canvas self._canvas_items: list[int] = [] self._overlay_widgets: list[tk.Widget] = [] self.cluster_env = tk.StringVar(master=root, value="service") self.selected_k3d_cluster = tk.StringVar(master=root, value="") self.selected_kubectx = tk.StringVar(master=root, value="ctx-000") self.service_namespace = tk.StringVar(master=root, value="knoe-system") self.prod_artifacts_path = tk.StringVar(master=root, value="/tmp") self.gitops_provider = tk.StringVar(master=root, value="gitea") self.gworkspace_domain = tk.StringVar(master=root, value="") self.gworkspace_customer_id = tk.StringVar(master=root, value="") self.gworkspace_admin_email = tk.StringVar(master=root, value="") self.gcloud_project_id = tk.StringVar(master=root, value="") self.gcloud_region = tk.StringVar(master=root, value="us-central1") self.gcloud_sa_key_path = tk.StringVar(master=root, value="") self.app_cluster_kubecontext = tk.StringVar(master=root, value="") self.db_cluster_kubecontext = tk.StringVar(master=root, value="") self.k3s_server_url = tk.StringVar(master=root, value="") self.k3s_token = tk.StringVar(master=root, value="") self._kubectx_applied = "ctx-000" self._switch_calls: list[str] = [] def _on_cluster_env_change(self, *_args): return None def _get_k3d_cluster_list(self) -> list[str]: return [] def _on_create_k3d_cluster(self): return None def _on_delete_k3d_cluster(self): return None def _on_k3d_cluster_select(self, *_args): return None def _get_kubectx_list(self) -> list[str]: return ["ctx-000", "ctx-001"] def _switch_kubectx(self, context_name: str) -> bool: self._switch_calls.append(context_name) return True def _verify_k3s_services(self): return None def show_page(self, _page: str): return None def _validate_and_save_cluster_config(self): return True app = _KubectxApplyApp(root, canvas) app._render_init_cluster_page() root.update() # No pending change initially. assert app._kubectx_apply_btn.cget("state") == "disabled" # Selecting a different context should NOT switch immediately. app.selected_kubectx.set("ctx-001") app._on_kubectx_select() assert app._switch_calls == [] assert app._kubectx_apply_btn.cget("state") == "normal" # Apply triggers the switch. app._on_kubectx_apply() assert app._switch_calls == ["ctx-001"] finally: root.destroy() def test_prod_cluster_screen_renders_production_tabs_without_gitops_selector(): """Prod env must render production config tabs and hide inline GitOps selector.""" root = tk.Tk() root.withdraw() try: canvas = tk.Canvas(root, width=975, height=900, bg="white") canvas.pack(fill="both", expand=False) root.update() app = _DummyClusterApp(root, canvas, kubectx_values=["ctx-prod"]) app.cluster_env.set("prod") app._render_init_cluster_page() root.update() texts = _canvas_texts(canvas) assert "GitOps Provider:" not in texts assert "Production Configuration" in texts notebooks = [w for w in app._overlay_widgets if isinstance(w, ttk.Notebook)] assert notebooks, "Expected production notebook to be rendered" prod_notebook = notebooks[0] tab_texts = [prod_notebook.tab(tab_id, "text") for tab_id in prod_notebook.tabs()] assert tab_texts == [ "Cloud", "Database", "Backup + Storage", "GCP Storage", "Auth + Routing", "Migration", "Plan / Apply", ] finally: root.destroy()