mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 10:13:58 +00:00
975 lines
38 KiB
Python
975 lines
38 KiB
Python
"""Sidebar navigation, page transitions, footer and app identity."""
|
|
|
|
import os
|
|
import platform
|
|
from pathlib import Path
|
|
import tkinter as tk
|
|
from tkinter import ttk, messagebox, filedialog
|
|
from knoe import config as inst_config
|
|
from knoe.core.topology import cleanup_topology_collectors
|
|
from knoe.core.env import get_resource_path, _deployment_mode_from_env
|
|
|
|
|
|
class NavigationMixin:
|
|
"""Sidebar navigation, page transitions, footer and app identity."""
|
|
|
|
def _set_app_identity(self):
|
|
"""Set the installer identity: process/menu name and Dock icon on macOS.
|
|
|
|
Notes:
|
|
- We set Tk's appname for consistency.
|
|
- On macOS, attempt to set the process/menu name to 'Knoe.DB Installer'
|
|
via the shared setup_macos_app_identity() helper.
|
|
- Prefer the Knoe Tools.app .icns from the built app; fallback to local PNG/GIF.
|
|
"""
|
|
# Set Tk application name
|
|
try:
|
|
self.root.tk.call("tk", "appname", "Knoe.DB Installer")
|
|
except Exception:
|
|
pass
|
|
|
|
# macOS: set process/menu name and Dock icon
|
|
if platform.system() == "Darwin":
|
|
try:
|
|
from knoe.ui.screens import setup_macos_app_identity
|
|
setup_macos_app_identity()
|
|
except Exception:
|
|
pass
|
|
|
|
icns_candidates = [
|
|
get_resource_path(
|
|
"knoe-app/dist/Knoe Tools.app/Contents/Resources/Knoe Tools.icns"
|
|
),
|
|
get_resource_path(
|
|
"knoe-app/dist/Knoe Tools.app/Contents/Resources/Knoe.icns"
|
|
),
|
|
]
|
|
icns_path = next((p for p in icns_candidates if p.exists()), None)
|
|
if icns_path is not None:
|
|
try:
|
|
# Use PyObjC if available
|
|
from AppKit import NSApplication, NSImage
|
|
|
|
img = NSImage.alloc().initWithContentsOfFile_(str(icns_path))
|
|
if img is not None:
|
|
NSApplication.sharedApplication().setApplicationIconImage_(img)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# If no .icns was found, try setting Dock icon from configured PNG
|
|
try:
|
|
import ctypes
|
|
import ctypes.util
|
|
objc = ctypes.cdll.LoadLibrary(ctypes.util.find_library("objc"))
|
|
objc.objc_getClass.restype = ctypes.c_void_p
|
|
objc.sel_registerName.restype = ctypes.c_void_p
|
|
objc.objc_msgSend.restype = ctypes.c_void_p
|
|
|
|
def msg(obj, sel_name, *args, restype=ctypes.c_void_p, argtypes=None):
|
|
if obj is None: return None
|
|
sel = objc.sel_registerName(sel_name.encode("ascii"))
|
|
objc.objc_msgSend.restype = restype
|
|
if argtypes is not None:
|
|
objc.objc_msgSend.argtypes = [ctypes.c_void_p, ctypes.c_void_p] + argtypes
|
|
else:
|
|
objc.objc_msgSend.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
|
|
return objc.objc_msgSend(obj, sel, *args)
|
|
|
|
app_cls = objc.objc_getClass(b"NSApplication")
|
|
if app_cls:
|
|
app = msg(app_cls, "sharedApplication")
|
|
if app:
|
|
cfg_icon = inst_config.get_ui_icon_image_path()
|
|
if cfg_icon.exists():
|
|
ns_image_cls = objc.objc_getClass(b"NSImage")
|
|
if ns_image_cls:
|
|
NSUTF8StringEncoding = 4
|
|
ns_str_cls = objc.objc_getClass(b"NSString")
|
|
icon_path_ns = msg(ns_str_cls, "stringWithCString:encoding:", str(cfg_icon).encode("utf-8"), NSUTF8StringEncoding, argtypes=[ctypes.c_char_p, ctypes.c_uint])
|
|
img = msg(ns_image_cls, "alloc")
|
|
img = msg(img, "initWithContentsOfFile:", icon_path_ns, argtypes=[ctypes.c_void_p])
|
|
if img:
|
|
msg(app, "setApplicationIconImage:", img, argtypes=[ctypes.c_void_p])
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
from AppKit import NSApplication, NSImage
|
|
|
|
cfg_icon = inst_config.get_ui_icon_image_path()
|
|
if cfg_icon.exists():
|
|
png_img = NSImage.alloc().initWithContentsOfFile_(str(cfg_icon))
|
|
if png_img is not None:
|
|
NSApplication.sharedApplication().setApplicationIconImage_(
|
|
png_img
|
|
)
|
|
# do not return; still set Tk icon below for consistency
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback: Tk icon from image assets (PNG/GIF)
|
|
# Prefer config-defined icon image
|
|
try:
|
|
cfg_icon = inst_config.get_ui_icon_image_path()
|
|
except Exception:
|
|
cfg_icon = get_resource_path("img/knoeIcon.png")
|
|
img_candidates = [
|
|
cfg_icon,
|
|
get_resource_path("img/knoe.png"),
|
|
get_resource_path("img/knoe-type.png"),
|
|
get_resource_path("img/knoe-type.gif"),
|
|
get_resource_path("img/Knoe.png"),
|
|
get_resource_path("img/knoeLogoSepia.png"),
|
|
]
|
|
for p in img_candidates:
|
|
try:
|
|
if p.exists():
|
|
self._app_iconphoto = tk.PhotoImage(file=str(p))
|
|
try:
|
|
self.root.iconphoto(True, self._app_iconphoto)
|
|
except Exception:
|
|
pass
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
def show_config_dialog(self):
|
|
"""Show configuration dialog for Docker image import directory."""
|
|
dialog = tk.Toplevel(self.root)
|
|
dialog.title("Configuration")
|
|
dialog.geometry("500x200")
|
|
dialog.resizable(False, False)
|
|
dialog.transient(self.root)
|
|
dialog.grab_set()
|
|
|
|
# Center on parent
|
|
x = self.root.winfo_x() + (self.root.winfo_width() // 2) - 250
|
|
y = self.root.winfo_y() + (self.root.winfo_height() // 2) - 100
|
|
dialog.geometry(f"+{x}+{y}")
|
|
|
|
container = tk.Frame(dialog, padx=20, pady=20)
|
|
container.pack(fill="both", expand=True)
|
|
|
|
tk.Label(
|
|
container,
|
|
text="Docker Image Import Directory:",
|
|
font=("SF Pro Text", 12, "bold"),
|
|
).pack(anchor="w")
|
|
tk.Label(
|
|
container,
|
|
text="Images found here will be loaded instead of pulled from Docker Hub.",
|
|
font=("SF Pro Text", 10),
|
|
fg="#666666",
|
|
).pack(anchor="w", pady=(0, 10))
|
|
|
|
row = tk.Frame(container)
|
|
row.pack(fill="x")
|
|
|
|
entry = tk.Entry(
|
|
row,
|
|
textvariable=self.docker_import_dir,
|
|
font=("SF Pro Text", 11),
|
|
bg="white",
|
|
fg="black",
|
|
insertbackground="black",
|
|
)
|
|
entry.pack(side="left", fill="x", expand=True, padx=(0, 5))
|
|
|
|
def browse():
|
|
path = filedialog.askdirectory(
|
|
initialdir=self.docker_import_dir.get() or Path.home()
|
|
)
|
|
if path:
|
|
self.docker_import_dir.set(path)
|
|
|
|
tk.Button(row, text="Browse...", command=browse).pack(side="right")
|
|
|
|
def save():
|
|
path = self.docker_import_dir.get().strip()
|
|
self.knoe_cfg_data["Global"]["DOCKER_IMPORT_DIR"] = path
|
|
self._save_knoe_cfg()
|
|
dialog.destroy()
|
|
|
|
btns = tk.Frame(container)
|
|
btns.pack(fill="x", pady=(20, 0))
|
|
|
|
tk.Button(btns, text="Save", command=save, width=10).pack(side="right")
|
|
tk.Button(btns, text="Cancel", command=dialog.destroy, width=10).pack(
|
|
side="right", padx=10
|
|
)
|
|
|
|
def configure_styles(self):
|
|
"""Configure ttk styles"""
|
|
base_bg = "white"
|
|
self.style.configure("TFrame", background="white")
|
|
self.style.configure(
|
|
"TLabel", background="white", foreground="black", font=("SF Pro Text", 11)
|
|
)
|
|
self.style.configure("Header.TLabel", font=("SF Pro Text", 16, "bold"))
|
|
self.style.configure("Small.TLabel", font=("SF Pro Text", 10))
|
|
|
|
# Consistent button styling
|
|
self.style.configure(
|
|
"Nav.TButton",
|
|
font=("SF Pro Text", 11),
|
|
padding=(12, 8),
|
|
background="#F5F5DC",
|
|
bordercolor="#F5F5DC",
|
|
lightcolor="#F5F5DC",
|
|
darkcolor="#F5F5DC",
|
|
relief="flat",
|
|
)
|
|
|
|
# Map state colors to avoid black boxes on hover/active
|
|
self.style.map(
|
|
"Nav.TButton",
|
|
background=[("active", "#E5E5D5"), ("pressed", "#D5D5C5")],
|
|
bordercolor=[("active", "#E5E5D5"), ("pressed", "#D5D5C5")],
|
|
lightcolor=[("active", "#E5E5D5"), ("pressed", "#D5D5C5")],
|
|
darkcolor=[("active", "#E5E5D5"), ("pressed", "#D5D5C5")],
|
|
)
|
|
|
|
# Notebook styling to match light theme and avoid dark mode shifts on macOS
|
|
self.style.theme_use("default")
|
|
self.style.configure(
|
|
"TNotebook", background="white", borderwidth=0, highlightthickness=0
|
|
)
|
|
self.style.configure(
|
|
"TNotebook.Tab",
|
|
background="#F5F5DC",
|
|
foreground="black",
|
|
lightcolor="#F5F5DC",
|
|
bordercolor="#CCCCCC",
|
|
darkcolor="#F5F5DC",
|
|
borderwidth=1,
|
|
padding=[10, 5],
|
|
)
|
|
self.style.map(
|
|
"TNotebook.Tab",
|
|
background=[("selected", "white")],
|
|
bordercolor=[("selected", "#CCCCCC")],
|
|
lightcolor=[("selected", "white")],
|
|
focuscolor=[("selected", "white")],
|
|
)
|
|
|
|
# Ensure common controls inherit white background
|
|
try:
|
|
self.style.configure("TCheckbutton", background=base_bg, foreground="black")
|
|
self.style.configure(
|
|
"TCombobox", fieldbackground="white", background=base_bg
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
_MODE_COLORS = {"min": "#8E44AD", "k3d": "#4A90D9", "k3s": "#27AE60", "gke": "#E67E22"}
|
|
|
|
def _set_deployment_mode(self, mode: str) -> None:
|
|
"""Switch the active deployment mode and persist it to the environment."""
|
|
self.deployment_mode.set(mode)
|
|
os.environ["KNOE_MODE"] = mode
|
|
|
|
def _refresh_mode_tabs(self) -> None:
|
|
"""No-op: mode tabs replaced by welcome screen card selector."""
|
|
pass
|
|
|
|
def _is_min_mode(self) -> bool:
|
|
"""Return True when the user has selected minimal (containerd-only) mode."""
|
|
return self.deployment_mode.get() == "min"
|
|
|
|
def _create_sidebar_nav(self):
|
|
"""Create the left-hand navigation menu."""
|
|
# ── INSTALLER label ──────────────────────────────────────────────
|
|
tk.Label(
|
|
self.sidebar,
|
|
text="INSTALLER",
|
|
bg="#F5F5DC",
|
|
fg="#8B8B7A",
|
|
font=("SF Pro Text", 10, "bold"),
|
|
anchor="w",
|
|
).pack(fill="x", padx=20, pady=(10, 10))
|
|
|
|
for text, page_id in self.nav_items:
|
|
lbl = tk.Label(
|
|
self.sidebar,
|
|
text=text,
|
|
bg="#F5F5DC",
|
|
fg="black",
|
|
font=("SF Pro Text", 11),
|
|
anchor="w",
|
|
padx=20,
|
|
pady=5,
|
|
cursor="hand2",
|
|
)
|
|
lbl.pack(fill="x")
|
|
lbl.bind("<Button-1>", lambda e, p=page_id: self.show_page(p))
|
|
self.nav_widgets[page_id] = lbl
|
|
|
|
def _update_nav_highlight(self, active_id):
|
|
"""Highlight the current active page in the sidebar."""
|
|
# Handle dep_id mapping to 'Dependencies'
|
|
if active_id.startswith("dep_"):
|
|
active_id = "deps_summary"
|
|
|
|
for page_id, widget in self.nav_widgets.items():
|
|
if page_id == active_id:
|
|
widget.configure(bg="#E5E5D5", font=("SF Pro Text", 11, "bold"))
|
|
else:
|
|
widget.configure(bg="#F5F5DC", font=("SF Pro Text", 11))
|
|
|
|
def create_navigation(self):
|
|
pass
|
|
|
|
def on_prev(self):
|
|
# Custom prev navigation for dependency pages when filtering
|
|
current_id = self.pages[self.page_index][0]
|
|
# Close Terminal if leaving build summary via Prev
|
|
if current_id == "build_summary":
|
|
try:
|
|
self.close_build_terminal()
|
|
except Exception:
|
|
pass
|
|
if current_id.startswith("dep_"):
|
|
seq = self._dep_navigation_sequence()
|
|
try:
|
|
i = seq.index(current_id)
|
|
except ValueError:
|
|
i = -1
|
|
if i > 0:
|
|
self.show_page(seq[i - 1])
|
|
return
|
|
else:
|
|
# Go back to summary if there is no previous in sequence
|
|
self.show_page("deps_summary")
|
|
return
|
|
if current_id == "deps_summary":
|
|
self.show_page("welcome")
|
|
return
|
|
if current_id == "network_scan":
|
|
# Go back to last relevant dep or deps_summary
|
|
if self.all_dependencies_installed():
|
|
self.show_page("deps_summary")
|
|
else:
|
|
seq = self._dep_navigation_sequence()
|
|
self.show_page(seq[-1] if seq else "deps_summary")
|
|
return
|
|
if current_id == "env_setup":
|
|
if self._is_min_mode():
|
|
self.show_page("deps_summary")
|
|
else:
|
|
self.show_page("network_scan")
|
|
return
|
|
if current_id == "init_cluster":
|
|
if getattr(self, "_showing_service_overlay", False):
|
|
self._showing_service_overlay = False
|
|
self.show_page("init_cluster")
|
|
return
|
|
self.show_page("env_setup")
|
|
return
|
|
if current_id == "cluster_nodes":
|
|
self.show_page("init_cluster")
|
|
return
|
|
if current_id == "common_services":
|
|
if self._should_show_cluster_nodes_screen():
|
|
self.show_page("cluster_nodes")
|
|
else:
|
|
self.show_page("init_cluster")
|
|
return
|
|
if current_id == "database_options":
|
|
self.show_page("common_services")
|
|
return
|
|
if current_id == "init_password":
|
|
if self._is_min_mode():
|
|
self.show_page("env_setup")
|
|
else:
|
|
self.show_page("init_db_build")
|
|
return
|
|
if current_id == "init_db_build":
|
|
self.show_page("database_options")
|
|
return
|
|
if current_id == "init_scripts":
|
|
self.show_page("init_password")
|
|
return
|
|
if current_id == "kerberos_config":
|
|
self.show_page("init_scripts")
|
|
return
|
|
if current_id == "knoe_users":
|
|
self.show_page("kerberos_config")
|
|
return
|
|
if current_id == "argocd_config":
|
|
if self.kerberos_enabled.get():
|
|
self.show_page("knoe_users")
|
|
else:
|
|
self.show_page("kerberos_config")
|
|
return
|
|
if current_id == "gitops_config":
|
|
self.show_page("argocd_config")
|
|
return
|
|
if current_id == "supabase_config":
|
|
self.show_page("gitops_config")
|
|
return
|
|
if current_id == "init_cnpg_deploy":
|
|
self.show_page("supabase_config")
|
|
return
|
|
if current_id == "security":
|
|
if self._is_min_mode():
|
|
self.show_page("init_scripts")
|
|
return
|
|
if current_id == "create_installer":
|
|
self.show_page("init_cnpg_deploy")
|
|
return
|
|
# Default prev
|
|
if self.page_index > 0:
|
|
self.show_page(self.page_index - 1)
|
|
return
|
|
|
|
def on_next(self):
|
|
# Special handling for dynamic labels
|
|
current_id = self.pages[self.page_index][0]
|
|
print(
|
|
f"[DEBUG] on_next: current_id='{current_id}', page_index={self.page_index}"
|
|
)
|
|
|
|
if current_id == "welcome":
|
|
self.show_page("deps_summary")
|
|
return
|
|
|
|
if current_id == "deps_summary":
|
|
# Determine where to go from summary
|
|
# Always go to the next dependency or Network Scan
|
|
if self.all_dependencies_installed():
|
|
print("[DEBUG] on_next: all deps installed, going to network_scan")
|
|
self.show_page("network_scan")
|
|
return
|
|
|
|
# Go to first missing dependency page
|
|
seq = self._dep_navigation_sequence()
|
|
print(f"[DEBUG] on_next: missing deps sequence: {seq}")
|
|
if seq:
|
|
self.show_page(seq[0])
|
|
else:
|
|
print(
|
|
"[DEBUG] on_next: all deps seem OK in sequence, going to network_scan"
|
|
)
|
|
self.show_page("network_scan")
|
|
return
|
|
|
|
if current_id.startswith("dep_"):
|
|
# Navigate within dependency sequence
|
|
seq = self._dep_navigation_sequence()
|
|
print(f"[DEBUG] on_next: dep sequence: {seq}")
|
|
try:
|
|
i = seq.index(current_id)
|
|
except ValueError:
|
|
i = -1
|
|
|
|
if i >= 0 and i < len(seq) - 1:
|
|
self.show_page(seq[i + 1])
|
|
return
|
|
else:
|
|
# After last relevant dep page, go back to summary if anything is still missing
|
|
if not self.all_dependencies_installed():
|
|
print(
|
|
"[DEBUG] on_next: some deps still missing, returning to deps_summary"
|
|
)
|
|
self.show_page("deps_summary")
|
|
else:
|
|
print(
|
|
"[DEBUG] on_next: all deps now installed, going to network_scan"
|
|
)
|
|
self.show_page("network_scan")
|
|
return
|
|
|
|
if current_id == "network_scan":
|
|
if self._is_min_mode():
|
|
self.show_page("env_setup")
|
|
return
|
|
# Capture network scan info
|
|
try:
|
|
self.knoe_cfg_data["Network"][
|
|
"KDC_AUTO_DETECTED"
|
|
] = self.kerberos_kdc.get()
|
|
self.knoe_cfg_data["Network"]["KERBEROS_AUTO_ENABLED"] = str(
|
|
self.kerberos_enabled.get()
|
|
)
|
|
self._save_knoe_cfg()
|
|
except Exception:
|
|
pass
|
|
self.show_page("env_setup")
|
|
return
|
|
|
|
if current_id == "env_setup":
|
|
# Collect values, validate, write env.sh, then go to Database Creation
|
|
vals: dict[str, str] = {}
|
|
raw_vals: dict[str, str] = {}
|
|
try:
|
|
for k in (
|
|
"KNOE_HOME",
|
|
"KNOE_CONF",
|
|
"PROLE_DATA",
|
|
"PROLE_LOGS",
|
|
"KNOE_SERVICE",
|
|
):
|
|
raw_vals[k] = self._env_entries[k].get().strip()
|
|
|
|
# Persist to knoe.cfg as *literal* overrides only.
|
|
# New policy: do not store variable expressions for filesystem paths.
|
|
defaults_raw = {}
|
|
try:
|
|
defaults_raw = self._env_setup_defaults_raw() or {}
|
|
except Exception:
|
|
defaults_raw = {}
|
|
defaults_resolved = {}
|
|
resolved = {}
|
|
try:
|
|
defaults_resolved = self._resolve_env_paths_for_fs(defaults_raw) or {}
|
|
except Exception:
|
|
defaults_resolved = {}
|
|
try:
|
|
resolved = self._resolve_env_paths_for_fs({**defaults_raw, **raw_vals}) or {}
|
|
except Exception:
|
|
resolved = {}
|
|
|
|
sys_env = self.knoe_cfg_data.get("System Environment")
|
|
if not isinstance(sys_env, dict):
|
|
sys_env = {}
|
|
for k in (
|
|
"KNOE_HOME",
|
|
"KNOE_CONF",
|
|
"PROLE_DATA",
|
|
"PROLE_LOGS",
|
|
"KNOE_SERVICE",
|
|
):
|
|
rv = (resolved.get(k) or "").strip()
|
|
dv = (defaults_resolved.get(k) or "").strip()
|
|
if not rv:
|
|
sys_env.pop(k, None)
|
|
continue
|
|
if "$" in rv:
|
|
# Should not happen (resolved values should be literal), but
|
|
# never persist variable tokens for filesystem paths.
|
|
sys_env.pop(k, None)
|
|
continue
|
|
if dv and rv == dv:
|
|
# Default value: omit from config.
|
|
sys_env.pop(k, None)
|
|
continue
|
|
sys_env[k] = rv
|
|
if sys_env:
|
|
self.knoe_cfg_data["System Environment"] = sys_env
|
|
else:
|
|
self.knoe_cfg_data.pop("System Environment", None)
|
|
|
|
vals.update(raw_vals)
|
|
vals["DATABASE_NAMESPACE"] = (self.db_namespace.get() or "").strip()
|
|
vals["CLUSTER_NAME"] = (
|
|
self.cnpg_cluster_name.get() if hasattr(self, "cnpg_cluster_name") else ""
|
|
).strip()
|
|
except Exception:
|
|
# Best-effort fallback: use internal defaults for env.sh.
|
|
vals = self._env_defaults()
|
|
# Basic validation: require non-empty KNOE_HOME
|
|
if not vals.get("KNOE_HOME"):
|
|
try:
|
|
messagebox.showerror("Environment", "Please specify KNOE_HOME")
|
|
except Exception:
|
|
pass
|
|
return
|
|
try:
|
|
self._save_env_to_file(vals)
|
|
# Reload our environment so subsequent steps (Build) inherit it
|
|
self._after_env_saved()
|
|
except Exception as e:
|
|
try:
|
|
messagebox.showwarning("Environment", f"Could not save env.sh: {e}")
|
|
except Exception:
|
|
pass
|
|
return
|
|
if self._is_min_mode():
|
|
self.show_page("init_password")
|
|
else:
|
|
self.show_page("init_cluster")
|
|
return
|
|
|
|
if current_id == "init_password":
|
|
p1 = self.db_password.get()
|
|
p2 = self.db_password_confirm.get()
|
|
if not p1:
|
|
try:
|
|
messagebox.showerror("Password", "Password cannot be empty.")
|
|
except Exception:
|
|
pass
|
|
return
|
|
if p1 != p2:
|
|
try:
|
|
messagebox.showerror("Password", "Passwords do not match.")
|
|
except Exception:
|
|
pass
|
|
return
|
|
if not self._validate_port_forward_overlaps():
|
|
return
|
|
# Preparing actions are executed as the first part of Initialization Scripts.
|
|
self.show_page("init_scripts")
|
|
return
|
|
|
|
if current_id == "init_db_build":
|
|
self.show_page("init_password")
|
|
return
|
|
|
|
if current_id == "build":
|
|
if getattr(self, "_built_success", False):
|
|
self.show_page("create_installer")
|
|
else:
|
|
self.perform_build()
|
|
return
|
|
|
|
if current_id == "init_cluster":
|
|
if not self._validate_and_save_cluster_config():
|
|
return
|
|
if not self._cluster_ready_for_navigation():
|
|
return
|
|
# Cluster Nodes placement is only relevant for multi-node, non-k3d clusters.
|
|
if self._should_show_cluster_nodes_screen():
|
|
self.show_page("cluster_nodes")
|
|
else:
|
|
self.show_page("common_services")
|
|
return
|
|
|
|
if current_id == "cluster_nodes":
|
|
policy = getattr(self, "_cluster_nodes_current_policy", None) or {}
|
|
if hasattr(self, "_validate_and_save_cluster_nodes_policy"):
|
|
if not self._validate_and_save_cluster_nodes_policy(policy):
|
|
return
|
|
try:
|
|
env_key = self._cluster_env_key()
|
|
cleanup_topology_collectors(
|
|
kubectl_base_cmd=self._kubectl_base_cmd(env_key),
|
|
)
|
|
except Exception:
|
|
pass
|
|
self.show_page("common_services")
|
|
return
|
|
|
|
if current_id == "common_services":
|
|
if not getattr(self, "_common_services_success", False):
|
|
try:
|
|
messagebox.showerror(
|
|
"Common Services",
|
|
"All common services must be green before continuing.",
|
|
)
|
|
except Exception:
|
|
pass
|
|
return
|
|
self.show_page("database_options")
|
|
return
|
|
|
|
if current_id == "database_options":
|
|
if self._generate_knoe_db_dockerfile():
|
|
self.show_page("init_db_build")
|
|
return
|
|
|
|
if current_id == "init_scripts":
|
|
self.knoe_cfg_data["Initialization Scripts"]["STATUS"] = (
|
|
"Completed" if getattr(self, "_scripts_success", False) else "Attempted"
|
|
)
|
|
self._save_knoe_cfg()
|
|
if self._is_min_mode():
|
|
self.show_page("security")
|
|
elif self.kerberos_enabled.get():
|
|
self.show_page("kerberos_config")
|
|
else:
|
|
self.show_page("argocd_config")
|
|
return
|
|
|
|
if current_id == "knoe_users":
|
|
self.knoe_cfg_data.setdefault("Knoe User Accounts", {})["STATUS"] = (
|
|
"Provisioned" if getattr(self, "_knoe_users_success", False) else "Attempted"
|
|
)
|
|
self._save_knoe_cfg()
|
|
self.show_page("argocd_config")
|
|
return
|
|
|
|
if current_id == "kerberos_config":
|
|
# Capture kerberos config
|
|
self.knoe_cfg_data["Kerberos Authentication"]["ENABLED"] = str(
|
|
self.kerberos_enabled.get()
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"][
|
|
"REALM"
|
|
] = self.kerberos_realm.get()
|
|
self.knoe_cfg_data["Kerberos Authentication"][
|
|
"KDC"
|
|
] = self.kerberos_kdc.get()
|
|
self.knoe_cfg_data["Kerberos Authentication"][
|
|
"SERVER"
|
|
] = self.kerberos_kdc.get()
|
|
self.knoe_cfg_data["Kerberos Authentication"][
|
|
"USER"
|
|
] = self.kerberos_user.get()
|
|
self.knoe_cfg_data["Kerberos Authentication"][
|
|
"PASSWORD"
|
|
] = self.kerberos_password.get()
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_PORT_FORWARD"] = (
|
|
os.environ.get("KRB5_AD_PORT_FORWARD", "1")
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_TCP_PORTS"] = (
|
|
os.environ.get("KRB5_AD_TCP_PORTS", "88 389 445 464 636")
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_UDP_PORTS"] = (
|
|
os.environ.get("KRB5_AD_UDP_PORTS", "88 464")
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_HOST_NETWORK"] = (
|
|
os.environ.get("KRB5_AD_PROXY_HOST_NETWORK", "1")
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_IMAGE"] = (
|
|
os.environ.get("KRB5_AD_PROXY_IMAGE", "alpine/socat")
|
|
)
|
|
self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_SERVICE"] = (
|
|
os.environ.get("KRB5_AD_SERVICE_NAME", "knoe-kerberos-ad-dc")
|
|
)
|
|
self._save_knoe_cfg()
|
|
self.show_page("knoe_users")
|
|
return
|
|
if current_id == "argocd_config":
|
|
self.knoe_cfg_data["Optional Features"]["ARGOCD_ENABLED"] = str(
|
|
self.argocd_enabled.get()
|
|
)
|
|
argocd_status = (
|
|
"Skipped"
|
|
if not self.argocd_enabled.get()
|
|
else (
|
|
"Deployed"
|
|
if getattr(self, "_argocd_success", False)
|
|
else "Attempted"
|
|
)
|
|
)
|
|
argocd_ns = (self.argocd_namespace.get() or "argocd").strip() or "argocd"
|
|
self.knoe_cfg_data.setdefault("Global", {})["ARGOCD_NAMESPACE"] = argocd_ns
|
|
self.knoe_cfg_data["ArgoCD"] = {
|
|
"STATUS": argocd_status,
|
|
"ARGOCD_NAMESPACE": argocd_ns,
|
|
}
|
|
self._save_knoe_cfg()
|
|
self.show_page("gitops_config")
|
|
return
|
|
|
|
if current_id == "gitops_config":
|
|
self.knoe_cfg_data["Optional Features"]["GITOPS_ENABLED"] = str(
|
|
self.gitops_enabled.get()
|
|
)
|
|
gitops_status = (
|
|
"Skipped"
|
|
if not self.gitops_enabled.get()
|
|
else (
|
|
"Deployed"
|
|
if getattr(self, "_gitops_success", False)
|
|
else "Attempted"
|
|
)
|
|
)
|
|
# Derive the correct namespace from the chosen provider.
|
|
# gitops_git_provider (combobox: Gitea/GitLab/Forgejo) is the active selector.
|
|
git_provider = (
|
|
getattr(self, "gitops_git_provider", None) and
|
|
self.gitops_git_provider.get()
|
|
) or ""
|
|
if not git_provider:
|
|
git_provider = (
|
|
getattr(self, "gitops_provider", None) and
|
|
self.gitops_provider.get()
|
|
) or "Gitea"
|
|
is_gitlab = git_provider.lower() == "gitlab"
|
|
if is_gitlab:
|
|
gitops_ns = (self.gitops_namespace.get() or "gitlab").strip() or "gitlab"
|
|
else:
|
|
gitops_ns = (self.gitops_namespace.get() or "gitea").strip() or "gitea"
|
|
self.knoe_cfg_data["GitOps"] = {
|
|
"STATUS": gitops_status,
|
|
"GITOPS_PROVIDER": git_provider.lower(),
|
|
"GITOPS_NAMESPACE": gitops_ns,
|
|
}
|
|
if is_gitlab:
|
|
self.knoe_cfg_data["GitOps"]["GITLAB_NAMESPACE"] = gitops_ns
|
|
self._save_knoe_cfg()
|
|
self.show_page("supabase_config")
|
|
return
|
|
|
|
if current_id == "supabase_config":
|
|
self.knoe_cfg_data["Optional Features"]["SUPABASE_ENABLED"] = str(
|
|
self.supabase_enabled.get()
|
|
)
|
|
supabase_status = (
|
|
"Skipped"
|
|
if not self.supabase_enabled.get()
|
|
else (
|
|
"Deployed"
|
|
if getattr(self, "_supabase_success", False)
|
|
else "Attempted"
|
|
)
|
|
)
|
|
self.knoe_cfg_data["Supabase"] = {
|
|
"STATUS": supabase_status,
|
|
}
|
|
self._save_knoe_cfg()
|
|
self.show_page("init_cnpg_deploy")
|
|
return
|
|
|
|
if current_id == "init_cnpg_deploy":
|
|
self.knoe_cfg_data["Deployment"]["STATUS"] = (
|
|
"Deployed" if getattr(self, "_cnpg_success", False) else "Attempted"
|
|
)
|
|
self._save_knoe_cfg()
|
|
self.show_page("security")
|
|
return
|
|
|
|
if current_id == "security":
|
|
if "Install" in self.knoe_cfg_data:
|
|
self.knoe_cfg_data["Install"]["STATUS"] = "Finished"
|
|
self._save_knoe_cfg()
|
|
self.root.destroy()
|
|
return
|
|
|
|
if current_id == "create_installer":
|
|
print("[DEBUG] on_next: at create_installer, Finish clicked. Closing.")
|
|
if "Install" in self.knoe_cfg_data:
|
|
self.knoe_cfg_data["Install"]["STATUS"] = "Finished"
|
|
self._save_knoe_cfg()
|
|
self.root.destroy()
|
|
return
|
|
|
|
# Default next
|
|
if self.page_index < len(self.pages) - 1:
|
|
print(f"[DEBUG] on_next: default next to index {self.page_index + 1}")
|
|
self.show_page(self.page_index + 1)
|
|
else:
|
|
print("[DEBUG] on_next: already at last page")
|
|
return
|
|
|
|
def _should_show_cluster_nodes_screen(self) -> bool:
|
|
"""Show Cluster Nodes only for multi-node, non-k3d clusters."""
|
|
try:
|
|
env_key = (self.cluster_env.get() or "").strip()
|
|
except Exception:
|
|
env_key = ""
|
|
if _deployment_mode_from_env(env_key) == "k3d":
|
|
return False
|
|
|
|
topo = getattr(self, "ansible_topology", None) or {}
|
|
groups = topo.get("groups") or {}
|
|
hosts = topo.get("hosts") or {}
|
|
|
|
k3s_hosts = groups.get("k3s_hosts") or []
|
|
if isinstance(k3s_hosts, list) and len(k3s_hosts) > 1:
|
|
return True
|
|
|
|
try:
|
|
return len(hosts) > 1
|
|
except Exception:
|
|
return False
|
|
|
|
def update_footer(self):
|
|
# Default states for tk.Button
|
|
self.prev_button.configure(state="normal")
|
|
self.next_button.configure(state="normal")
|
|
first = self.page_index == 0
|
|
# Base label
|
|
self.next_button.configure(text="Next")
|
|
|
|
if getattr(self, "_showing_service_overlay", False):
|
|
if not getattr(self, "_common_services_success", False):
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
|
|
# Page-specific adjustments
|
|
pid = self.pages[self.page_index][0]
|
|
if pid == "welcome":
|
|
if not self.deployment_mode.get():
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
elif pid == "build":
|
|
# Build page: show Build or Next depending on state
|
|
if getattr(self, "_built_success", False):
|
|
self.next_button.configure(text="Next")
|
|
else:
|
|
self.next_button.configure(text="Build")
|
|
elif pid == "supabase_config":
|
|
# Supabase screen: allow Next when disabled or after success
|
|
if self.supabase_enabled.get():
|
|
if getattr(self, "_supabase_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
elif pid == "gitops_config":
|
|
if self.gitops_enabled.get():
|
|
if getattr(self, "_gitops_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
elif pid == "common_services":
|
|
if getattr(self, "_common_services_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
elif pid == "argocd_config":
|
|
if self.argocd_enabled.get():
|
|
if getattr(self, "_argocd_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
elif pid == "init_scripts":
|
|
# Initialization Scripts: Next is disabled until success
|
|
if getattr(self, "_scripts_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
elif pid == "kerberos_config":
|
|
if self.kerberos_enabled.get():
|
|
if getattr(self, "_kerberos_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
else:
|
|
self.next_button.configure(state="normal")
|
|
elif pid == "knoe_users":
|
|
# Next is enabled once provisioning succeeds; also allow skipping
|
|
# by clicking Next a second time (idempotent re-runs are safe).
|
|
if getattr(self, "_knoe_users_success", False):
|
|
self.next_button.configure(state="normal")
|
|
else:
|
|
self.next_button.configure(state="disabled")
|
|
# Visibility rules
|
|
self.prev_button.pack_forget()
|
|
self.next_button.pack_forget()
|
|
if self.deploy_button:
|
|
self.deploy_button.pack_forget()
|
|
if self.launch_button:
|
|
self.launch_button.pack_forget()
|
|
if self.gear_button:
|
|
self.gear_button.pack_forget()
|
|
|
|
if pid == "init_password" and self.gear_button:
|
|
self.gear_button.pack(side="left", padx=(20, 0), pady=12)
|
|
|
|
if pid == "create_installer":
|
|
self.next_button.configure(text="Finish")
|
|
self.next_button.pack(side="right", padx=(0, 20), pady=12)
|
|
return
|
|
|
|
if first:
|
|
self.next_button.pack(side="right", padx=(0, 20), pady=12)
|
|
else:
|
|
# [Prev] [Next] clustered right
|
|
self.next_button.pack(side="right", padx=(0, 20), pady=12)
|
|
self.prev_button.pack(side="right", padx=(0, 8), pady=12)
|
|
|
|
def _splash_should_hide_nav(self) -> bool:
|
|
"""Deprecated: Welcome page no longer performs dependency checks or gates navigation."""
|
|
return False
|