prole/knoe/ui/screens/validate.py

"""Validation screen with auto-refresh status monitoring."""

import subprocess
import threading
import time
import tkinter as tk
import webbrowser
from tkinter import ttk, messagebox, filedialog

from knoe import screen as ui


class ValidateScreenMixin:
    """Validation screen with auto-refresh status monitoring."""

    def create_validate_screen(self):
        """Create the Validate screen"""
        frame = self._page_container()
        self.screens["validate"] = frame

        # Title
        title = ttk.Label(frame, text="Validate Deployment", style="Title.TLabel")
        title.pack(pady=(0, 20))

        # Prometheus link
        prometheus_frame = tk.Frame(frame, bg="white")
        prometheus_frame.pack(pady=(0, 20))

        prometheus_label = tk.Label(
            prometheus_frame,
            text="Prometheus: ",
            bg="white",
            fg="#6e6e73",
            font=("Helvetica", 11),
        )
        prometheus_label.pack(side="left")

        prometheus_link = tk.Label(
            prometheus_frame,
            text="http://localhost:9090",
            bg="white",
            fg="#4a9eff",
            font=("Helvetica", 11, "underline"),
            cursor="hand2",
        )
        prometheus_link.pack(side="left")
        prometheus_link.bind(
            "<Button-1>", lambda e: webbrowser.open("http://localhost:9090")
        )

        # Status display
        status_frame = tk.Frame(frame, bg="white")
        status_frame.pack(fill="both", expand=True, pady=10)

        status_label = ttk.Label(
            status_frame, text="Cluster Status", style="Heading.TLabel"
        )
        status_label.pack(anchor="w", pady=(0, 10))

        # Status text area - using TerminalConsole for consistency
        # Use a background frame to ensure NO borders are visible around the console
        console_bg = tk.Frame(status_frame, bg="white", highlightthickness=0, bd=0)
        console_bg.pack(fill="both", expand=True, padx=1, pady=1)
        self._validation_console = ui.TerminalConsole(
            console_bg, highlightthickness=0, bd=0
        )
        self._validation_console.pack(fill="both", expand=True)
        self.status_text = self._validation_console.text

        # Auto-refresh checkbox
        refresh_frame = tk.Frame(frame, bg="white")
        refresh_frame.pack(pady=10)

        self.auto_refresh_var = tk.BooleanVar(value=True)
        refresh_check = tk.Checkbutton(
            refresh_frame,
            text="Auto-refresh every 10 seconds",
            variable=self.auto_refresh_var,
            bg="white",
            fg="#1d1d1f",
            selectcolor="#f0f0f0",
            activebackground="white",
            activeforeground="#1d1d1f",
            font=("Helvetica", 10),
            command=self.toggle_auto_refresh,
        )
        refresh_check.pack(side="left", padx=10)

        # Manual refresh button
        refresh_btn = tk.Button(
            refresh_frame,
            text="Refresh Now",
            command=self.refresh_status,
            bg="#4a9eff",
            fg="white",
            activebackground="#3a8eef",
            highlightbackground="white",
            font=("Helvetica", 10),
            padx=15,
            pady=5,
            cursor="hand2",
            relief="flat",
        )
        refresh_btn.pack(side="left", padx=10)

        # Start auto-refresh
        self.refresh_status()
        self.toggle_auto_refresh()

    def toggle_auto_refresh(self):
        """Toggle auto-refresh"""
        if self.auto_refresh_var.get():
            if not self.validation_running:
                self.validation_running = True
                self.validation_thread = threading.Thread(
                    target=self.auto_refresh_loop, daemon=True
                )
                self.validation_thread.start()
        else:
            self.validation_running = False

    def auto_refresh_loop(self):
        """Auto-refresh loop"""
        while self.validation_running:
            time.sleep(10)
            if self.validation_running:
                self.root.after(0, self.refresh_status)

    def refresh_status(self):
        """Refresh the cluster status"""

        def update():
            try:
                full_status = ""

                # Get k3d cluster list
                try:
                    cluster_result = subprocess.run(
                        ["k3d", "cluster", "list"],
                        capture_output=True,
                        text=True,
                        timeout=5,
                    )
                    full_status += (
                        f"=== k3d Cluster Status ===\n{cluster_result.stdout}\n\n"
                    )
                except Exception as e:
                    full_status += f"=== k3d Cluster Status ===\nError: {str(e)}\n\n"

                # Get kubectl cnpg status
                try:
                    namespace = (self.db_namespace.get() or "default").strip()
                    cmd = self._kubectl_base_cmd() + [
                        "cnpg",
                        "status",
                        "knoe-db",
                        "-n",
                        namespace,
                    ]
                    result = subprocess.run(
                        cmd, capture_output=True, text=True, timeout=10
                    )
                    if result.returncode == 0:
                        status_output = result.stdout
                    else:
                        # Try default namespace if prole fails, or just show error
                        status_output = f"Error: {result.stderr}\n\nNote: Make sure kubectl cnpg plugin is installed and cluster exists in '{namespace}' namespace."

                    full_status += f"=== CloudNativePG Status (namespace: {namespace}) ===\n{status_output}\n"
                except FileNotFoundError:
                    full_status += "=== CloudNativePG Status ===\nError: kubectl not found. Please install dependencies first.\n"
                except subprocess.TimeoutExpired:
                    full_status += (
                        "=== CloudNativePG Status ===\nError: Command timed out\n"
                    )
                except Exception as e:
                    full_status += f"=== CloudNativePG Status ===\nError: {str(e)}\n"

                full_status += f"\nLast updated: {time.strftime('%Y-%m-%d %H:%M:%S')}"

                self.status_text.delete("1.0", tk.END)
                self.status_text.insert("1.0", full_status)

            except Exception as e:
                self.status_text.delete("1.0", tk.END)
                self.status_text.insert("1.0", f"Error: {str(e)}")

        threading.Thread(target=update, daemon=True).start()