diff --git a/tests/test_clean_install.py b/tests/test_clean_install.py new file mode 100644 index 00000000..9bfd7ddd --- /dev/null +++ b/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()