test(clean-install): add opt-in clone-and-verify pytest test

tests/test_clean_install.py

@@ -0,0 +1,114 @@
+"""Opt-in clean install verification test.
+
+Clones the Manual Slop repo to tmp_path, runs `uv sync`, launches
+`sloppy.py --enable-test-hooks`, and verifies the Hook API responds.
+Catches "works on my machine" failures by exercising the full install-and-launch
+path in an isolated environment.
+
+Opt-in: set RUN_CLEAN_INSTALL_TEST=1 to enable. Otherwise skipped.
+Requires network access to REPO_URL.
+"""
+import os
+import socket
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+
+import pytest
+
+
+REPO_URL = "https://git.cozyair.dev/ed/manual_slop"
+STARTUP_TIMEOUT_SECONDS = 30
+READINESS_POLL_INTERVAL = 0.5
+HOOK_PORT = 8999
+
+
+def _http_get_json(url: str, timeout: float) -> dict:
+ with urllib.request.urlopen(url, timeout=timeout) as response:
+  raw = response.read().decode("utf-8")
+  return _safe_json_loads(raw)
+
+
+def _safe_json_loads(raw: str) -> dict:
+ import json
+ return json.loads(raw)
+
+
+@pytest.mark.clean_install
+def test_clean_install_runs_with_hooks(tmp_path: Path) -> None:
+ if os.environ.get("RUN_CLEAN_INSTALL_TEST") != "1":
+  pytest.skip("Set RUN_CLEAN_INSTALL_TEST=1 to enable")
+
+ clone_dir = tmp_path / "manual_slop"
+ project_root = Path(__file__).resolve().parent.parent
+
+ subprocess.run(
+  ["git", "clone", REPO_URL, str(clone_dir)],
+  capture_output=True, text=True, timeout=60,
+  check=True,
+ )
+
+ subprocess.run(
+  ["uv", "sync"],
+  cwd=str(clone_dir),
+  capture_output=True, text=True, timeout=180,
+  check=True,
+ )
+
+ creationflags = 0
+ if os.name == "nt":
+  creationflags = subprocess.CREATE_NEW_PROCESS_GROUP
+
+ process = subprocess.Popen(
+  ["uv", "run", "sloppy.py", "--enable-test-hooks"],
+  cwd=str(clone_dir),
+  stdout=subprocess.PIPE,
+  stderr=subprocess.PIPE,
+  creationflags=creationflags,
+ )
+
+ try:
+  start = time.time()
+  ready = False
+  while time.time() - start < STARTUP_TIMEOUT_SECONDS:
+   if process.poll() is not None:
+    stderr_out = process.stderr.read(2000) if process.stderr else b""
+    pytest.fail(
+     f"Process exited early. stderr: {stderr_out.decode('utf-8', errors='replace')}"
+    )
+   try:
+    data = _http_get_json(f"http://127.0.0.1:{HOOK_PORT}/status", timeout=1.0)
+    if data.get("status") in ("running", "ready", "ok"):
+     ready = True
+     break
+   except (urllib.error.URLError, socket.timeout, ConnectionError, TimeoutError, OSError):
+    pass
+   time.sleep(READINESS_POLL_INTERVAL)
+
+  assert ready, (
+   f"Hook server did not respond within {STARTUP_TIMEOUT_SECONDS}s. "
+   f"stderr: {process.stderr.read(2000).decode('utf-8', errors='replace') if process.stderr else 'N/A'}"
+  )
+
+  mma_data = _http_get_json(f"http://127.0.0.1:{HOOK_PORT}/api/mma_status", timeout=5.0)
+  assert isinstance(mma_data, dict), f"mma_status returned non-dict: {mma_data!r}"
+
+ finally:
+  _cleanup_process(process)
+
+
+def _cleanup_process(process: subprocess.Popen) -> None:
+ if os.name == "nt":
+  subprocess.run(
+   ["taskkill", "/F", "/T", "/PID", str(process.pid)],
+   capture_output=True,
+  )
+ else:
+  process.terminate()
+  try:
+   process.wait(timeout=5)
+  except subprocess.TimeoutExpired:
+   process.kill()