manual_slop/tests/conftest.py

import pytest
import asyncio
import subprocess
import time
import requests
import os
import signal
import sys
import datetime
import shutil
from pathlib import Path
from typing import Generator, Any
from unittest.mock import patch

# Ensure project root and src/ are in path for imports
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))

# Import the App class after patching if necessary, but here we just need the type hint
from gui_2 import App

class VerificationLogger:
 def __init__(self, test_name: str, script_name: str) -> None:
  self.test_name = test_name
  self.script_name = script_name
  self.entries = []
  self.start_time = time.time()
  # Route artifacts to tests/logs/
  self.logs_dir = Path(f"tests/logs/{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}")
  self.logs_dir.mkdir(parents=True, exist_ok=True)

 def log_state(self, field: str, before: Any, after: Any) -> None:
  delta = ""
  if isinstance(before, (int, float)) and isinstance(after, (int, float)):
   diff = after - before
   delta = f"{'+' if diff > 0 else ''}{diff}"
  self.entries.append({
   "Field": field,
   "Before": str(before),
   "After": str(after),
   "Delta": delta
  })

 def finalize(self, title: str, status: str, result_msg: str) -> None:
  round(time.time() - self.start_time, 2)
  log_file = self.logs_dir / f"{self.script_name}.txt"
  with open(log_file, "w", encoding="utf-8") as f:
   f.write(f"[ Test: {self.test_name} ]\n")
   f.write(f"({title})\n\n")
   f.write(f"{self.test_name}: before vs after\n")
   f.write(f"{'Field':<25} {'Before':<20} {'After':<20} {'Delta':<15}\n")
   f.write("-" * 80 + "\n")
   for e in self.entries:
    f.write(f"{e['Field']:<25} {e['Before']:<20} {e['After']:<20} {e['Delta']:<15}\n")
   f.write("-" * 80 + "\n")
   f.write(f"{status} {self.test_name} ({result_msg})\n\n")
  print(f"[FINAL] {self.test_name}: {status} - {result_msg}")

@pytest.fixture(autouse=True)
def reset_ai_client() -> Generator[None, None, None]:
 """
 Autouse fixture that resets the ai_client global state before each test.
 This is critical for preventing state pollution between tests.
 """
 import ai_client
 import mcp_client
 ai_client.reset_session()
 # Reset callbacks to None or default to ensure no carry-over
 ai_client.confirm_and_run_callback = None
 ai_client.comms_log_callback = None
 ai_client.tool_log_callback = None
 # Clear all event listeners
 ai_client.events.clear()
 # Reset provider/model to defaults
 ai_client.set_provider("gemini", "gemini-2.5-flash-lite")
 # Reset MCP client state
 mcp_client.configure([], [])
 yield
 ai_client.reset_session()

@pytest.fixture
def vlogger(request) -> VerificationLogger:
 """Fixture to provide a VerificationLogger instance to a test."""
 test_name = request.node.name
 script_name = Path(request.node.fspath).stem
 return VerificationLogger(test_name, script_name)

def kill_process_tree(pid: int | None) -> None:
 """Robustly kills a process and all its children."""
 if pid is None:
  return
 try:
  print(f"[Fixture] Attempting to kill process tree for PID {pid}...")
  if os.name == 'nt':
  # /F is force, /T is tree (includes children)
   subprocess.run(["taskkill", "/F", "/T", "/PID", str(pid)], 
    stdout=subprocess.DEVNULL, 
    stderr=subprocess.DEVNULL, 
    check=False)
  else:
  # On Unix, kill the process group
   os.killpg(os.getpgid(pid), signal.SIGKILL)
  print(f"[Fixture] Process tree {pid} killed.")
 except Exception as e:
  print(f"[Fixture] Error killing process tree {pid}: {e}")

@pytest.fixture
def mock_app() -> Generator[App, None, None]:
 """
 Mock version of the App for simple unit tests that don't need a loop.
 """
 with (
  patch('src.models.load_config', return_value={
   'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
   'projects': {'paths': [], 'active': ''},
   'gui': {'show_windows': {}}
  }),
  patch('gui_2.save_config'),
  patch('gui_2.project_manager'),
  patch('gui_2.session_logger'),
  patch('gui_2.immapp.run'),
  patch('src.app_controller.AppController._load_active_project'),
  patch('src.app_controller.AppController._fetch_models'),
  patch.object(App, '_load_fonts'),
  patch.object(App, '_post_init'),
  patch('src.app_controller.AppController._prune_old_logs'),
  patch('src.app_controller.AppController.start_services'),
  patch('src.app_controller.AppController._init_ai_and_hooks'),
  patch('gui_2.PerformanceMonitor')
 ):
  app = App()
  yield app
  if hasattr(app, 'controller'):
   app.controller.shutdown()
  elif hasattr(app, 'shutdown'):
   app.shutdown()

@pytest.fixture
def app_instance() -> Generator[App, None, None]:
 """
 Centralized App instance with all external side effects mocked.
 Matches the pattern used in test_token_viz.py and test_gui_phase4.py.
 """
 with (
  patch('src.models.load_config', return_value={
   'ai': {'provider': 'gemini', 'model': 'gemini-2.5-flash-lite'},
   'projects': {'paths': [], 'active': ''},
   'gui': {'show_windows': {}}
  }),
  patch('gui_2.save_config'),
  patch('gui_2.project_manager'),
  patch('gui_2.session_logger'),
  patch('gui_2.immapp.run'),
  patch('src.app_controller.AppController._load_active_project'),
  patch('src.app_controller.AppController._fetch_models'),
  patch.object(App, '_load_fonts'),
  patch.object(App, '_post_init'),
  patch('src.app_controller.AppController._prune_old_logs'),
  patch('src.app_controller.AppController.start_services'),
  patch('src.app_controller.AppController._init_ai_and_hooks'),
  patch('gui_2.PerformanceMonitor')
 ):
  app = App()
  yield app
  # Cleanup: Ensure background threads and asyncio loop are stopped
  if hasattr(app, 'controller'):
   app.controller.shutdown()

  if hasattr(app, 'shutdown'):
   app.shutdown()
  
  # Use controller._loop for cleanup
  loop = getattr(app.controller, '_loop', None) if hasattr(app, 'controller') else None
  if loop and not loop.is_closed():
   tasks = [t for t in asyncio.all_tasks(loop) if not t.done()]
   if tasks:
    # Cancel tasks so they can be gathered
    for task in tasks:
     task.cancel()
    # We can't really run the loop if it's already stopping or thread is dead,
    # but we try to be clean.
    try:
     if loop.is_running():
      loop.call_soon_threadsafe(loop.stop)
    except: pass
   
   # Finally close the loop if we can
   try:
    if not loop.is_running():
     loop.close()
   except: pass

@pytest.fixture(scope="session")
def live_gui() -> Generator[tuple[subprocess.Popen, str], None, None]:
 """
    Session-scoped fixture that starts sloppy.py with --enable-test-hooks.
    Includes high-signal environment telemetry and workspace isolation.
    """
 gui_script = os.path.abspath("sloppy.py")
 diag = VerificationLogger("live_gui_startup", "live_gui_diag")
 diag.log_state("GUI Script", "N/A", "gui_2.py")
 
 # 1. Create a isolated workspace for the live GUI
 temp_workspace = Path("tests/artifacts/live_gui_workspace")
 if temp_workspace.exists():
  for _ in range(5):
   try:
    shutil.rmtree(temp_workspace)
    break
   except PermissionError:
    time.sleep(0.5)
 # Create the workspace directory before writing files
 temp_workspace.mkdir(parents=True, exist_ok=True)
 
 # Create minimal project files to avoid cluttering root
 # NOTE: Do NOT create config.toml here - we use SLOP_CONFIG env var
 # to point to the actual project root config.toml
 (temp_workspace / "manual_slop.toml").write_text("[project]\nname = 'TestProject'\n", encoding="utf-8")
 (temp_workspace / "conductor" / "tracks").mkdir(parents=True, exist_ok=True)
 
 # Resolve absolute paths for shared resources
 project_root = Path(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 config_file = project_root / "config.toml"
 cred_file = project_root / "credentials.toml"
 mcp_file = project_root / "mcp_env.toml"
 
 # Preserve GUI layout for tests
 layout_file = Path("manualslop_layout.ini")
 if layout_file.exists():
  shutil.copy2(layout_file, temp_workspace / layout_file.name)
 
 # Check if already running (shouldn't be)
 try:
  resp = requests.get("http://127.0.0.1:8999/status", timeout=0.5)
  if resp.status_code == 200:
   print("[Fixture] WARNING: Hook Server already up on port 8999. Test state might be polluted.")
   # Optionally try to reset it
   try: requests.post("http://127.0.0.1:8999/api/gui", json={"action": "click", "item": "btn_reset"}, timeout=1)
   except: pass
 except: pass

 print(f"\n[Fixture] Starting {gui_script} --enable-test-hooks in {temp_workspace}...")
 os.makedirs("logs", exist_ok=True)
 log_file_name = Path(gui_script).name.replace('.', '_')
 log_file = open(f"logs/{log_file_name}_test.log", "w", encoding="utf-8")
 
 # Use environment variable to point to temp config if App supports it,
 # or just run from that CWD.
 env = os.environ.copy()
 env["PYTHONPATH"] = str(project_root.absolute())
 if config_file.exists():
  env["SLOP_CONFIG"] = str(config_file.absolute())
 if cred_file.exists():
  env["SLOP_CREDENTIALS"] = str(cred_file.absolute())
 if mcp_file.exists():
  env["SLOP_MCP_ENV"] = str(mcp_file.absolute())
 
 process = subprocess.Popen(
  ["uv", "run", "python", "-u", gui_script, "--enable-test-hooks"],
  stdout=log_file,
  stderr=log_file,
  text=True,
  cwd=str(temp_workspace.absolute()),
  env=env,
  creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == 'nt' else 0
 )
 
 diag.log_state("GUI Process PID", "N/A", process.pid)
 
 max_retries = 15
 ready = False
 print(f"[Fixture] Waiting up to {max_retries}s for Hook Server on port 8999...")
 start_time = time.time()
 while time.time() - start_time < max_retries:
  try:
   response = requests.get("http://127.0.0.1:8999/status", timeout=0.5)
   if response.status_code == 200:
    ready = True
    print(f"[Fixture] GUI Hook Server for {gui_script} is ready after {round(time.time() - start_time, 2)}s.")
    break
  except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
   if process.poll() is not None:
    print(f"[Fixture] {gui_script} process died unexpectedly during startup.")
    break
   time.sleep(0.5)
 
 diag.log_state("Startup Success", "N/A", str(ready))
 diag.log_state("Startup Time", "N/A", f"{round(time.time() - start_time, 2)}s")
 
 if not ready:
  diag.finalize("Live GUI Startup Telemetry", "FAIL", "Hook server failed to respond.")
  print(f"[Fixture] TIMEOUT/FAILURE: Hook server for {gui_script} failed to respond.")
  kill_process_tree(process.pid)
  pytest.fail(f"Failed to start {gui_script} with test hooks.")
 
 diag.finalize("Live GUI Startup Telemetry", "PASS", "Hook server successfully initialized.")
 
 try:
  yield process, gui_script
 finally:
  print(f"\n[Fixture] Finally block triggered: Shutting down {gui_script}...")
  # Reset the GUI state before shutting down
  try:
   from api_hook_client import ApiHookClient
   client = ApiHookClient()
   client.reset_session()
   time.sleep(0.5)
  except: pass
  kill_process_tree(process.pid)
  time.sleep(1.0)
  log_file.close()
  # Cleanup temp workspace with retry for Windows file locks
  for _ in range(5):
   try:
    shutil.rmtree(temp_workspace)
    break
   except PermissionError:
    time.sleep(0.5)
   except:
    break