84 lines
4.3 KiB
Python
84 lines
4.3 KiB
Python
"""Start real HTTP services with fake GPIO, then a real headless Godot cabinet."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from http.server import HTTPServer
|
|
from test_room import ROOT, server
|
|
sys.path.insert(0, str(ROOT / "tests/gpio"))
|
|
from test_bridge import bridge, FakePin
|
|
|
|
OP = "test-operator-token-1234"
|
|
CAB = "test-cabinet-token-5678"
|
|
|
|
def post(path, data=None, token=OP):
|
|
request = urllib.request.Request("http://127.0.0.1:8766" + path, data=json.dumps(data or {}).encode(),
|
|
headers={"Authorization": "Bearer " + token, "Content-Type": "application/json"})
|
|
with urllib.request.urlopen(request, timeout=3) as response: return json.load(response)
|
|
|
|
def wait_for(predicate):
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
state = post("/operator/state")
|
|
if predicate(state): return state
|
|
time.sleep(.1)
|
|
raise AssertionError("Timed out waiting for room state: " + str(state))
|
|
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
room = server.Fleet(Path(temp) / "room.sqlite3", json.loads((ROOT / "tools/room_controller/catalog.json").read_text()))
|
|
gpio = bridge.Outputs(FakePin, "cabinet_01_gpio", [17], ledger=Path(temp) / "gpio.sqlite3")
|
|
services = [HTTPServer(("127.0.0.1", 8766), server.handler_for(room, OP, {"cabinet_01": CAB, "cabinet_02": CAB + "-two"})),
|
|
HTTPServer(("127.0.0.1", 8765), bridge.handler_for(gpio))]
|
|
threads = [threading.Thread(target=s.serve_forever, daemon=True) for s in services]
|
|
for thread in threads: thread.start()
|
|
process = None
|
|
second = None
|
|
try:
|
|
try:
|
|
post("/operator/state", token=CAB)
|
|
raise AssertionError("Cabinet token accessed operator API")
|
|
except urllib.error.HTTPError as exc:
|
|
assert exc.code == 401
|
|
env = dict(os.environ, ARCADE_ROOM_URL="http://127.0.0.1:8766", ARCADE_CABINET_TOKEN=CAB)
|
|
process = subprocess.Popen([sys.argv[1], "--headless", "--path", str(ROOT),
|
|
"--log-file", str(ROOT / ".godot/managed_integration.log"),
|
|
"--script", "tests/room_network/managed_agent.gd", "--", "--gpio=real"], env=env)
|
|
second = subprocess.Popen([sys.argv[1], "--headless", "--path", str(ROOT),
|
|
"--log-file", str(ROOT / ".godot/managed_second.log"),
|
|
"--script", "tests/room_network/managed_agent.gd", "--", "--cabinet=cabinet_02", "--test-no-reward"],
|
|
env=dict(env, ARCADE_CABINET_TOKEN=CAB + "-two"))
|
|
wait_for(lambda s: s["online"] and s["cabinets"]["cabinet_02"]["online"])
|
|
state = post("/operator/command", {"action": "prepare_scenario", "scenario": "proof_of_concept"})
|
|
post("/operator/command", {"action": "prepare", "cabinet": "cabinet_02", "game": "galaga"})
|
|
assert state["phase"] == "preparing"
|
|
wait_for(lambda s: s["phase"] == "ready" and s["cabinets"]["cabinet_02"]["phase"] == "ready")
|
|
post("/operator/command", {"action": "start"})
|
|
post("/operator/command", {"action": "start", "cabinet": "cabinet_02"})
|
|
final = wait_for(lambda s: s["reward"].startswith("GPIO accepted"))
|
|
wait_for(lambda s: s["cabinets"]["cabinet_02"]["phase"] == "won")
|
|
assert gpio.pins[17]["device"].activations == 1
|
|
assert sum(c["type"] == "reward" for c in final["commands"]) == 1
|
|
assert process.wait(timeout=8) == 0
|
|
assert second.wait(timeout=8) == 0
|
|
assert not any(c["type"] == "reward" for c in post("/operator/state")["cabinets"]["cabinet_02"]["commands"])
|
|
print("PASS: two real Godot cabinets -> Pac-Man scenario + Galaga manual -> ready/start/win -> exactly one mapped GPIO acceptance")
|
|
finally:
|
|
if process and process.poll() is None:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
if second and second.poll() is None:
|
|
second.terminate()
|
|
second.wait(timeout=5)
|
|
for service in services:
|
|
service.shutdown()
|
|
service.server_close()
|
|
for thread in threads: thread.join()
|
|
gpio.close()
|
|
for cabinet in room.rooms.values(): cabinet.db.close()
|