import tkinter as tk 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 installer.ui.screens.base import ScreenBaseMixin from installer.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") # 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 ["prole-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 _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", "prod"]) 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. 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 # The status notebook and Repair button must remain within the visible plane. visible_h = max(canvas.winfo_height(), int(canvas.cget("height"))) for item_id in ( app._k3s_service_notebook_canvas_window, app._k3s_deploy_btn_canvas_window, ): bbox = canvas.bbox(item_id) assert bbox is not None assert bbox[3] <= visible_h # Long output should be scrollable (TerminalConsole provides a vertical scrollbar). assert getattr(app._k3s_service_status_console, "scroll", None) is not None assert app._k3s_service_status_console.scroll.winfo_exists() # Ensure large content doesn't crash the UI and the console remains read-only. for i in range(300): app._k3s_service_status_console.write(f"line {i}\n") root.update() assert app._k3s_service_status_console.text.cget("state") == "disabled" 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="prod") 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.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()