manual_slop/tests/test_default_layout_install.py

from __future__ import annotations

import os
import shutil
import signal
import subprocess
import time
from pathlib import Path
from typing import TextIO

import pytest


_PROJECT_ROOT: Path = Path(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
_SCRIPTS_DIR: Path = _PROJECT_ROOT / "tests" / "artifacts" / "tier2_state" / "default_layout_install_20260629"
_GUI_SCRIPT: str = str(_PROJECT_ROOT / "sloppy.py")
_INI_FILENAME: str = "manualslop_layout.ini"


def _spawn_sloppy_for(workspace: Path, log_suffix: str) -> subprocess.Popen:
 _SCRIPTS_DIR.mkdir(parents=True, exist_ok=True)
 log_path: Path = _SCRIPTS_DIR / f"test_{log_suffix}.log"
 log_file: TextIO = open(log_path, "w", encoding="utf-8", errors="replace")
 env: dict[str, str] = os.environ.copy()
 env["PYTHONPATH"] = str(_PROJECT_ROOT.absolute())
 args: list[str] = ["uv", "run", "python", "-u", _GUI_SCRIPT]
 creation: int = subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0
 proc: subprocess.Popen = subprocess.Popen(
  args,
  stdout=log_file,
  stderr=log_file,
  text=True,
  cwd=str(workspace.absolute()),
  env=env,
  creationflags=creation,
 )
 log_file.close()
 return proc


def _read_launch_log(log_suffix: str) -> str:
 log_path: Path = _SCRIPTS_DIR / f"test_{log_suffix}.log"
 if not log_path.exists():
  return ""
 try:
  return log_path.read_text(encoding="utf-8", errors="replace")
 except OSError:
  return ""


def _assert_install_applied(log_suffix: str) -> None:
 text: str = _read_launch_log(log_suffix)
 install_msgs: list[str] = [l for l in text.splitlines() if "pre-run installed" in l or "applied to live session" in l]
 assert install_msgs, (
  f"install was not invoked for this run; expected one of "
  f"'[GUI] pre-run installed default layout: ...' or "
  f"'[GUI] installed default layout: ... (and applied to live session)' "
  f"in stderr, got install-related lines: "
  f"{[l for l in text.splitlines() if 'default layout' in l]!r}"
 )


def _terminate(process: subprocess.Popen) -> None:
 if process.poll() is not None:
  return
 pid: int | None = process.pid
 if pid is None:
  return
 try:
  if os.name == "nt":
   subprocess.run(
    ["taskkill", "/F", "/T", "/PID", str(pid)],
    stdout=subprocess.DEVNULL,
    stderr=subprocess.DEVNULL,
    check=False,
   )
  else:
   os.killpg(os.getpgid(pid), signal.SIGKILL)
 except Exception:
  pass
 try:
  process.wait(timeout=2)
 except Exception:
  pass


def _read_ini(workspace: Path) -> str:
 ini_path: Path = workspace / _INI_FILENAME
 if not ini_path.exists():
  return ""
 try:
  return ini_path.read_text(encoding="utf-8", errors="replace")
 except OSError:
  return ""


def _has_window_with_collapsed_zero(text: str) -> bool:
 for line in text.splitlines():
  if line.startswith("[Window][") and line.rstrip().endswith("]"):
   return True
 return False


def _has_docking_block_with_docknodes(text: str) -> bool:
 if "[Docking][Data]" not in text:
  return False
 if "DockSpace" not in text:
  return False
 docknode_count: int = 0
 for line in text.splitlines():
  if line.strip().startswith("DockNode") and "ID=" in line:
   docknode_count += 1
 return docknode_count >= 1


def _every_window_has_dockid(text: str) -> bool:
 lines: list[str] = text.splitlines()
 blocks: dict[str, list[str]] = {}
 for idx, line in enumerate(lines):
  if line.startswith("[Window][") and line.rstrip().endswith("]"):
   block: list[str] = []
   for next_line in lines[idx + 1:]:
    if next_line.startswith("[") and "][" in next_line:
     break
    block.append(next_line)
   blocks[line] = block
 if not blocks:
  return False
 has_dockid: bool = True
 for header, block in blocks.items():
  if not any("DockId=" in bl for bl in block):
   has_dockid = False
   break
 return has_dockid


def _has_no_stale_window_names(text: str) -> bool:
 stale: set[str] = {
  "Projects", "Files", "Screenshots", "Discussion History",
  "Provider", "Message", "Response", "Tool Calls",
  "Comms History", "System Prompts",
 }
 for line in text.splitlines():
  if line.startswith("[Window][") and line.rstrip().endswith("]"):
   name: str = line[len("[Window]["):-1]
   if name in stale:
    return False
 return True


def _workspace_for(tmp_path: Path, test_name: str) -> Path:
 return tmp_path / f"_default_layout_install_{os.getpid()}_{test_name}"


def _start_subprocess_in(workspace: Path, log_suffix: str) -> subprocess.Popen:
 proc: subprocess.Popen = _spawn_sloppy_for(workspace, log_suffix)
 time.sleep(5.0)
 return proc


def _assert_installed_default(workspace: Path) -> None:
 ini_path: Path = workspace / _INI_FILENAME
 assert ini_path.exists(), f"expected {ini_path} to exist after launch"
 text: str = _read_ini(workspace)
 assert "[Window][Project Settings]" in text, (
  f"installed INI missing [Window][Project Settings]; got first 400 chars: {text[:400]!r}"
 )
 assert _has_window_with_collapsed_zero(text), (
  f"installed INI has no [Window][...] entry; got first 400 chars: {text[:400]!r}"
 )
 assert _has_docking_block_with_docknodes(text), (
  f"installed INI missing [Docking][Data] block with DockSpace + >=1 DockNode children; got first 400 chars: {text[:400]!r}"
 )
 assert _every_window_has_dockid(text), (
  f"installed INI does not have a DockId= line following every [Window][...] header; got first 400 chars: {text[:400]!r}"
 )
 assert _has_no_stale_window_names(text), (
  f"installed INI contains a window name from _STALE_WINDOW_NAMES -- _diag_layout_state will emit a stale-name warning; got first 400 chars: {text[:400]!r}"
 )


def test_default_layout_installed_when_ini_missing(tmp_path: Path) -> None:
 workspace: Path = _workspace_for(tmp_path, "ini_missing")
 workspace.mkdir(parents=True, exist_ok=True)
 ini_path: Path = workspace / _INI_FILENAME
 if ini_path.exists():
  ini_path.unlink()
 proc: subprocess.Popen = _start_subprocess_in(workspace, "ini_missing")
 try:
  _assert_installed_default(workspace)
  _assert_install_applied("ini_missing")
 finally:
  _terminate(proc)
  shutil.rmtree(workspace, ignore_errors=True)


def test_default_layout_installed_when_ini_empty(tmp_path: Path) -> None:
 workspace: Path = _workspace_for(tmp_path, "ini_empty")
 workspace.mkdir(parents=True, exist_ok=True)
 ini_path: Path = workspace / _INI_FILENAME
 ini_path.write_bytes(b"\n\n\n\n\n")
 proc: subprocess.Popen = _start_subprocess_in(workspace, "ini_empty")
 try:
  _assert_installed_default(workspace)
  _assert_install_applied("ini_empty")
 finally:
  _terminate(proc)
  shutil.rmtree(workspace, ignore_errors=True)


def test_default_layout_NOT_installed_when_layout_present(tmp_path: Path) -> None:
 workspace: Path = _workspace_for(tmp_path, "ini_custom")
 workspace.mkdir(parents=True, exist_ok=True)
 ini_path: Path = workspace / _INI_FILENAME
 custom_lines: list[str] = ["[Window][CustomPanelPersistent]", "Pos=10,10", "DockId=0xDEAD", ""]
 for i in range(32):
  custom_lines.append(f"; padding line {i}")
  custom_lines.append("[Window][Filler]")
  custom_lines.append(f"Pos={i},{i}")
  custom_lines.append("Size=100,100")
 content: str = "\n".join(custom_lines) + "\n"
 ini_path.write_text(content, encoding="utf-8")
 assert ini_path.stat().st_size >= 1000, "custom INI must be at least 1000 bytes for this test"
 proc: subprocess.Popen = _start_subprocess_in(workspace, "ini_custom")
 try:
  text: str = _read_ini(workspace)
  assert "[Window][CustomPanelPersistent]" in text, (
   f"custom INI should be preserved (no overwrite), but content changed: {text[:400]!r}"
  )
  assert "DockId=0xDEAD" in text, "custom DockId entry should survive install"
 finally:
  _terminate(proc)
  shutil.rmtree(workspace, ignore_errors=True)