"""Optional ArgoCD deployment screen. This screen intentionally mirrors the layout and behavior of the GitOps (Gitea) optional feature screen. """ from __future__ import annotations import os import subprocess import threading import tkinter as tk from knoe import screen as ui from knoe.core.env import PROJECT_ROOT, _normalize_cluster_env class ArgoCDScreenMixin: def _render_argocd_config_page(self): """Optional feature screen: ArgoCD.""" content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 ui.canvas_text( self, right_margin, 40, "Prole", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "Infrastructure Automated.", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) self._render_title("ArgoCD", y=150) description = ( "Deploys ArgoCD via etc/init_argocd.sh.\n" "Defaults to the 'argocd' namespace; update below to override." ) self._render_paragraph(description, y=210) # Enable toggle enabled_check = tk.Checkbutton( self.bg_canvas, text="Enable ArgoCD", variable=self.argocd_enabled, bg="white", fg="black", activebackground="white", activeforeground="black", selectcolor="white", command=self._on_argocd_toggle, font=("SF Pro Text", 11), ) enabled_window = self.bg_canvas.create_window( 48, 250, window=enabled_check, anchor="nw" ) self._canvas_items.append(enabled_window) self._overlay_widgets.append(enabled_check) # Namespace entry ui.canvas_text( self, 48, 290, "Kubernetes Namespace", fill="black", font=("SF Pro Text", 12, "bold"), ) ns_entry = tk.Entry( self.bg_canvas, textvariable=self.argocd_namespace, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) ns_window = self.bg_canvas.create_window( 48, 315, window=ns_entry, anchor="nw", width=220, height=30 ) self._canvas_items.append(ns_window) self._overlay_widgets.append(ns_entry) y = 360 initial_status = "Ready" if self.argocd_enabled.get() else "Disabled" self._argocd_status_var = tk.StringVar(value=initial_status) status_item = ui.canvas_text( self, 48, y, f"Status: {initial_status}", fill="black", font=("SF Pro Text", 12), ) def update_status_text(*_): try: self.bg_canvas.itemconfig( status_item, text=f"Status: {self._argocd_status_var.get()}" ) except Exception: pass self._argocd_status_var.trace_add("write", update_status_text) # Console output self._argocd_console = self._create_console_output( y=400, title="Deployment Output", width=880, height=330 ) # Deploy button self._argocd_deploy_button = tk.Button( self.bg_canvas, text="Deploy ArgoCD", command=self.run_argocd_deploy, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", activeforeground="black", highlightbackground="#F5F5DC", highlightcolor="#F5F5DC", highlightthickness=0, relief="flat", bd=0, cursor="hand2", disabledforeground="#8B8B7A", font=("SF Pro Text", 11), padx=20, pady=10, ) btn_window = self.bg_canvas.create_window( 48, 760, window=self._argocd_deploy_button, anchor="nw" ) self._overlay_widgets.append(self._argocd_deploy_button) self._canvas_items.append(btn_window) def run_argocd_deploy(self): if getattr(self, "_argocd_deploying", False): return if not self.argocd_enabled.get(): self._argocd_status_var.set("Disabled") self.update_footer() return self._argocd_deploying = True self._argocd_deploy_button.configure(state="disabled") self._argocd_status_var.set("Deploying...") self._argocd_success = False self._argocd_console.clear() self.update_footer() def worker(): namespace = ( (self.argocd_namespace.get() or "").strip() or os.environ.get("ARGOCD_NAMESPACE") or "argocd" ) env = os.environ.copy() env["PROLE_HOME"] = str(PROJECT_ROOT) env["PROLE_SERVICE"] = str(PROJECT_ROOT) env["NAMESPACE"] = namespace env["ARGOCD_NAMESPACE"] = namespace self._argocd_console.write("Starting ArgoCD deployment...\n") script_path = PROJECT_ROOT / "etc" / "init_argocd.sh" if not script_path.exists(): self._argocd_console.write(f"Error: {script_path} not found.\n") self.safe_after( lambda: self._argocd_status_var.set("Failed (Script not found)") ) self.safe_after( lambda: self._argocd_deploy_button.configure(state="normal") ) self._argocd_deploying = False return mode = (os.environ.get("PROLE_MODE") or "").strip() if not mode: cluster_env = _normalize_cluster_env(self.cluster_env.get()) if cluster_env == "dev": mode = "k3d" elif cluster_env in ("service", "prod"): mode = "k3s" if cluster_env == "service" else "k8s" else: mode = "k3d" args = ["--mode", mode, "--namespace", namespace, "update"] cfg_path = None try: if self._cfg_path_override and self._cfg_path_override.exists(): cfg_path = self._cfg_path_override else: conf_dir = self._resolve_prole_conf_dir() candidate = conf_dir / "prole.cfg" if candidate.exists(): cfg_path = candidate if not cfg_path: from knoe import prole_conf candidate = prole_conf.entrypoint_path( prole_conf.resolve_prole_conf_dir(PROJECT_ROOT) ) if candidate.exists(): cfg_path = candidate except Exception: cfg_path = None if cfg_path: args.extend(["-c", str(cfg_path)]) self._argocd_console.write(f"Mode: {mode}\n") cmd = ["bash", str(script_path)] + args try: proc = subprocess.Popen( cmd, cwd=str(PROJECT_ROOT), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) except Exception as e: self._argocd_console.write(f"Failed to start deploy: {e}\n") self.safe_after( lambda: self._argocd_status_var.set("Failed (Launch error)") ) self.safe_after( lambda: self._argocd_deploy_button.configure(state="normal") ) self._argocd_deploying = False return if proc.stdout: for line in iter(proc.stdout.readline, ""): if line: self._argocd_console.write(line) proc.stdout.close() rc = proc.wait() if rc == 0: self._argocd_success = True self.safe_after( lambda: self._argocd_status_var.set("Deployed Successfully") ) self._argocd_console.write( "\nArgoCD deployment completed successfully.\n" ) else: self._argocd_success = False self.safe_after( lambda: self._argocd_status_var.set(f"Failed (Code {rc})") ) self._argocd_console.write( f"\nArgoCD deployment failed with exit code {rc}.\n" ) self._argocd_deploying = False self.safe_after( lambda: self._argocd_deploy_button.configure(state="normal") ) self.safe_after(self.update_footer) threading.Thread(target=worker, daemon=True).start() def _on_argocd_toggle(self): enabled = self.argocd_enabled.get() try: self.prole_cfg_data["Optional Features"]["ARGOCD_ENABLED"] = str(enabled) # Keep compatibility with other parts of the installer that expect Global.ARGOCD_NAMESPACE. self.prole_cfg_data.setdefault("Global", {})["ARGOCD_NAMESPACE"] = ( (self.argocd_namespace.get() or "").strip() or "argocd" ) self._save_prole_cfg() except Exception: pass if not enabled: self._argocd_success = False if hasattr(self, "_argocd_status_var"): self._argocd_status_var.set("Disabled") try: self._argocd_console.write("ArgoCD disabled. Deployment skipped.\n") except Exception: pass self.update_footer()