92686159d1
Prepare dependency generations before selecting them. Keep shipped tool bytes separate from writable additions, and store facts beside their entries. Validate proposed plugin sets before config publication. Restore the previous config if the facts write fails. Consolidate duplicate updater, backup, setup, and voice helpers. Repair launcher selection, dependency consumers, download ownership, update feeds, and native Windows process and file handling. Verification: 206 changed/prior-failing Python files reported 4630 passed, one failed, and 330 skipped. Fix the remaining Hindsight fixture boundary. The final targeted rerun reported 234 passed and two skipped. The store review regression batch reported 83 passed and one skipped. Desktop TypeScript checks, 56 selected Electron tests, 24 release tests, and the removed-import/compatibility guards passed. This is an integration checkpoint, not full audit acceptance. The complete Python suite has not run on this fixed tree. Crash-atomic plugin publication, generation cleanup, receipt correlation, and packaged lifecycle acceptance remain open in docs/pm-audit-status.md.
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
import json
|
|
import pytest
|
|
import os
|
|
import queue
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
|
|
def _stdout_queue(proc: subprocess.Popen) -> queue.Queue[dict]:
|
|
out: queue.Queue[dict] = queue.Queue()
|
|
assert proc.stdout is not None
|
|
|
|
def drain() -> None:
|
|
for line in proc.stdout or []:
|
|
out.put(json.loads(line))
|
|
|
|
threading.Thread(target=drain, daemon=True).start()
|
|
return out
|
|
|
|
|
|
def _read_json_line(out: queue.Queue[dict], timeout: float = 2.0) -> dict:
|
|
try:
|
|
return out.get(timeout=timeout)
|
|
except queue.Empty as exc:
|
|
raise AssertionError("timed out waiting for compute host JSON") from exc
|
|
|
|
|
|
@pytest.mark.platforms("linux")
|
|
def test_compute_host_line_json_hello_and_shutdown():
|
|
repo = Path(__file__).resolve().parents[2]
|
|
env = dict(os.environ)
|
|
env["PYTHONPATH"] = str(repo) + os.pathsep + env.get("PYTHONPATH", "")
|
|
proc = subprocess.Popen(
|
|
[sys.executable, "-m", "tui_gateway.compute_host"],
|
|
cwd=str(repo),
|
|
env=env,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
bufsize=1,
|
|
)
|
|
assert proc.stdin is not None
|
|
out = _stdout_queue(proc)
|
|
try:
|
|
hello = _read_json_line(out)
|
|
assert hello["type"] == "hello"
|
|
assert hello["host_pid"] == proc.pid
|
|
|
|
proc.stdin.write(json.dumps({"type": "bogus", "request_id": "b"}) + "\n")
|
|
proc.stdin.flush()
|
|
error = _read_json_line(out)
|
|
assert error["type"] == "error"
|
|
assert error["message"] == "unknown frame type: bogus"
|
|
|
|
proc.stdin.write(json.dumps({"type": "shutdown", "request_id": "stop"}) + "\n")
|
|
proc.stdin.flush()
|
|
assert _read_json_line(out)["type"] == "shutdown.ack"
|
|
proc.wait(timeout=2)
|
|
finally:
|
|
if proc.poll() is None:
|
|
proc.kill()
|