116 lines
3.1 KiB
Python
116 lines
3.1 KiB
Python
"""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/gui/mma_status", timeout=5.0)
|
|
assert isinstance(mma_data, dict), f"mma_status returned non-dict: {mma_data!r}"
|
|
assert "mma_status" in mma_data, f"mma_status missing 'mma_status' key: {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()
|