mirror of
https://github.com/mrvasil/ctf-ad-mirror-tool-rust.git
synced 2026-10-23 14:31:02 +00:00
109 lines
4.6 KiB
Python
109 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Optional interoperability check against ForcAD's real Socket.IO libraries.
|
|
|
|
Install python-socketio==5.10.0 python-engineio==4.8.0 aiohttp in a disposable
|
|
venv, then run this script with the built mirror-tool binary as its argument.
|
|
No external hosts, legacy config or real credentials are used.
|
|
"""
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
|
|
from aiohttp import web, ClientSession
|
|
import socketio
|
|
|
|
|
|
async def main():
|
|
connections = 0
|
|
active = set()
|
|
sio = socketio.AsyncServer(async_mode="aiohttp", ping_interval=0.2, ping_timeout=1)
|
|
app = web.Application()
|
|
sio.attach(app)
|
|
|
|
@sio.on("connect", namespace="/game_events")
|
|
async def connected(sid, environ):
|
|
nonlocal connections
|
|
connections += 1
|
|
active.add(sid)
|
|
await sio.emit("init_scoreboard", {"data": {
|
|
"teams": [{"id": 1, "name": "Fixture A"}],
|
|
"tasks": [{"id": 2, "name": "Notes"}],
|
|
"state": {"round": 1, "team_tasks": [{"team_id": 1, "task_id": 2, "status": 101}]},
|
|
}}, to=sid, namespace="/game_events")
|
|
|
|
@sio.on("disconnect", namespace="/game_events")
|
|
async def disconnected(sid):
|
|
active.discard(sid)
|
|
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
await web.TCPSite(runner, "127.0.0.1", 18333).start()
|
|
with tempfile.TemporaryDirectory(prefix="mirror-forcad-interop-") as temp:
|
|
config = Path(temp) / "mirror.toml"
|
|
config.write_text('''[web]
|
|
listen="127.0.0.1:16701"
|
|
state_file="state.json"
|
|
[scoreboard]
|
|
kind="forcad"
|
|
url="http://127.0.0.1:18333"
|
|
poll_interval_ms=100
|
|
stale_after_ms=1000
|
|
''')
|
|
env = os.environ.copy()
|
|
env["MIRROR_PASSWORD"] = "local-interop-fixture-only"
|
|
process = await asyncio.create_subprocess_exec(str(Path(sys.argv[1]).resolve()), "--config", str(config), env=env, stdout=asyncio.subprocess.DEVNULL)
|
|
try:
|
|
async with ClientSession() as client:
|
|
for _ in range(100):
|
|
try:
|
|
response = await client.get("http://127.0.0.1:16701/healthz")
|
|
await response.read()
|
|
break
|
|
except OSError:
|
|
await asyncio.sleep(0.03)
|
|
response = await client.post("http://127.0.0.1:16701/api/login", json={"password": env["MIRROR_PASSWORD"]})
|
|
assert response.status == 200
|
|
cookie = response.headers["Set-Cookie"].split(";", 1)[0]
|
|
await response.read()
|
|
|
|
async def snapshot():
|
|
async with client.get("http://127.0.0.1:16701/api/snapshot", headers={"Cookie": cookie}) as response:
|
|
assert response.status == 200
|
|
return (await response.json())["scoreboard"]
|
|
|
|
async def wait_for(predicate, timeout=4):
|
|
async with asyncio.timeout(timeout):
|
|
while True:
|
|
state = await snapshot()
|
|
if predicate(state):
|
|
return state
|
|
await asyncio.sleep(0.03)
|
|
|
|
await wait_for(lambda s: s["round"] == 1)
|
|
await sio.emit("update_scoreboard", {"data": {"round": 2, "team_tasks": [{"team_id": 1, "task_id": 2, "status": 104, "stolen": 17}]}}, namespace="/game_events")
|
|
delta = await wait_for(lambda s: s["round"] == 2)
|
|
assert delta["entries"]["2"]["1"]["stolen"] == 17
|
|
assert delta["teams"][0]["name"] == "Fixture A"
|
|
# Several real Engine.IO heartbeat exchanges must preserve one session.
|
|
await asyncio.sleep(0.7)
|
|
assert (await snapshot())["connected"]
|
|
assert connections == 1
|
|
await wait_for(lambda s: s["stale"])
|
|
for sid in list(active):
|
|
await sio.disconnect(sid, namespace="/game_events")
|
|
await wait_for(lambda s: s["round"] == 1 and s["connected"] and not s["stale"])
|
|
assert connections == 2
|
|
print(json.dumps({"library": "python-socketio 5.10.0 / python-engineio 4.8.0", "initial_and_delta": "passed", "metadata_retention": "passed", "heartbeat_session_reuse": "passed", "stale_data": "passed", "namespace_disconnect_reconnect": "passed", "connections": connections}, indent=2))
|
|
finally:
|
|
process.terminate()
|
|
await asyncio.wait_for(process.wait(), timeout=10)
|
|
await sio.shutdown()
|
|
await runner.cleanup()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|