338 lines
19 KiB
Python
338 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""BUILD192 opt-in integration: real import lock held while interactive matches work.
|
|
|
|
Reuses the BUILD191 multi-match fixture with BUILD192 batch ACK support. The only source replacement is
|
|
an in-memory countries response; the only delay is inserted inside a real import
|
|
transaction, after service._import_lock has been acquired and before its first SQL.
|
|
No production module, shared limiter or admission gate is patched.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
|
|
COUNTRIES_XML = b'<Countries><Country id="191" iso2="RU" iso3="RUS" name="Test country" nameen="Test country" /></Countries>'
|
|
|
|
|
|
def load_base(project):
|
|
source = project / "tests/multi_match_harness.py"
|
|
spec = importlib.util.spec_from_file_location("hockey_admission_base", source)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def build_scenario(base):
|
|
class AdmissionSocket(base.HarnessSocket):
|
|
async def send_json(self, message):
|
|
command = message.get("command") or {}
|
|
if message.get("type") != "vmix.command" or command.get("Function") != "SuspendCountdown":
|
|
return await super().send_json(message)
|
|
scenario = self.scenario
|
|
person = scenario.people[self.name]
|
|
valid = (self.name.startswith("op") and message["device_id"] == person["device"]
|
|
and message["match_id"] == person["game"] and message["assignment_id"] == person["assignment"]
|
|
and command.get("SelectedName") == "Clock.Text"
|
|
and command.get("Input") in {"Shared Score", "shared-score-guid"})
|
|
if not valid:
|
|
scenario.transport_contract_errors.append({"recipient": self.name, "timer_frame": message})
|
|
raise AssertionError((self.name, message))
|
|
assert not scenario.heavy_release.is_set(), "Interactive timer delivery happened after import release"
|
|
packet = json.loads(json.dumps(message))
|
|
self.sent.append(packet)
|
|
scenario.timer_command_frames.append(packet)
|
|
scenario.request_started[packet["request_id"]] = time.perf_counter()
|
|
# Native timer commands have no Value; keep them outside Payload state.
|
|
await asyncio.sleep(scenario.args.ack_delay_ms / 1000)
|
|
await scenario.hub.receive_command_ack(person["device"], {"type": "command.ack",
|
|
"request_id": packet["request_id"], "assignment_id": packet["assignment_id"],
|
|
"match_id": packet["match_id"], "ok": True, "reason": ""})
|
|
scenario.request_started.pop(packet["request_id"], None)
|
|
|
|
# Only the separately loaded fixture module is adapted; the project file stays unchanged.
|
|
base.HarnessSocket = AdmissionSocket
|
|
|
|
class AdmissionScenario(base.Scenario):
|
|
def __init__(self, project, directory, args):
|
|
super().__init__(project, directory, args)
|
|
from hockey_data.agent_bridge import _stable_vmix_inventory_fingerprint
|
|
self.inventory["inputs"][0]["fields"].append({"name": "Clock.Text", "type": "text", "index": "999"})
|
|
self.fingerprint = _stable_vmix_inventory_fingerprint(self.inventory["inputs"])
|
|
self.import_entered = asyncio.Event()
|
|
self.heavy_release = threading.Event()
|
|
self.waiters_started = asyncio.Event()
|
|
self.import_calls = self.import_results = 0
|
|
self.store_calls = 0
|
|
self.store_lock = threading.Lock()
|
|
self.timer_command_frames, self.interactive_results = [], []
|
|
self.admission_observations = {}
|
|
self.import_errors = []
|
|
self.before_release_verified = False
|
|
self.holder_wait_ms = None
|
|
self.holder_timed_out = False
|
|
self.timer_values = {f"M{i}": 900000 - 1000 * i for i in range(1, 4)}
|
|
self.original_store = self.service._store_tournament_directory_profiles
|
|
self.original_fetch = self.service.client.fetch_xml
|
|
def store_inside_actual_import(session, **kwargs):
|
|
# The original public sync_country_directory's persist_countries
|
|
# already owns this genuine RLock. Do not replace the lock itself.
|
|
if kwargs.get("tournament_external_id") == "" and kwargs.get("countries") is not None:
|
|
with self.store_lock:
|
|
self.store_calls += 1
|
|
number = self.store_calls
|
|
if number == 1:
|
|
owned = getattr(self.service._import_lock, "_is_owned", None)
|
|
assert owned is None or owned(), "Fixture delay is outside the actual service import lock"
|
|
started = time.perf_counter()
|
|
self.loop.call_soon_threadsafe(self.import_entered.set)
|
|
released = self.heavy_release.wait(timeout=args.hold_timeout)
|
|
self.holder_wait_ms = round((time.perf_counter() - started) * 1000, 3)
|
|
self.holder_timed_out = not released
|
|
if not released:
|
|
raise AssertionError("Heavy import gate timed out; interactive progress did not release it")
|
|
return self.original_store(session, **kwargs)
|
|
self.service._store_tournament_directory_profiles = store_inside_actual_import
|
|
async def countries_response(endpoint):
|
|
from hockey_data.client import Stat2TVResponse
|
|
if "countries" not in endpoint:
|
|
raise AssertionError("Unexpected external source requested: " + endpoint)
|
|
return Stat2TVResponse(content=COUNTRIES_XML, status_code=200,
|
|
content_type="application/xml", auth_mode="offline", url="offline://countries")
|
|
self.service.client.fetch_xml = countries_response
|
|
|
|
def worker_snapshot(self):
|
|
from hockey_data import async_work
|
|
borrowed = async_work._limiter.get().borrowed_tokens
|
|
frames = []
|
|
for thread_id, frame in sys._current_frames().items():
|
|
cursor = frame
|
|
while cursor is not None:
|
|
if cursor.f_code.co_name == "persist_countries" and cursor.f_code.co_filename.endswith("hockey_data/service.py"):
|
|
frames.append({"thread": thread_id, "function": cursor.f_code.co_name, "line": cursor.f_lineno})
|
|
break
|
|
cursor = cursor.f_back
|
|
return {"shared_db_workers_borrowed": borrowed, "import_worker_frames": frames,
|
|
"store_entries": self.store_calls, "import_tasks_started": self.import_calls,
|
|
"import_tasks_finished": self.import_results, "release_set": self.heavy_release.is_set()}
|
|
|
|
async def one_import(self):
|
|
self.import_calls += 1
|
|
if self.import_calls == self.args.waiters + 1:
|
|
self.waiters_started.set()
|
|
try:
|
|
result = await self.service.sync_country_directory()
|
|
assert result.get("ok") and result["countries"] == 1, result
|
|
self.import_results += 1
|
|
return result
|
|
except Exception as error:
|
|
self.import_errors.append(repr(error))
|
|
raise
|
|
|
|
async def interactive_request(self, name, kind, method, path, body):
|
|
assert not self.heavy_release.is_set()
|
|
started = time.perf_counter()
|
|
response = await self.http.request(method, path, headers=await self.headers(name), json=body)
|
|
assert response.status_code == 200, (kind, name, response.text)
|
|
assert not self.heavy_release.is_set(), (kind, "completed after release")
|
|
self.interactive_results.append({"account": name, "game": self.people[name]["game"],
|
|
"kind": kind, "status": response.status_code, "completed_before_release": True,
|
|
"milliseconds": round((time.perf_counter() - started) * 1000, 3)})
|
|
return response.json()
|
|
|
|
async def interactive_work(self):
|
|
await self.bump()
|
|
async def match_pipeline(i):
|
|
name = f"op{i}"
|
|
person = self.people[name]
|
|
game = person["game"]
|
|
await self.interactive_request(name, "timer_write", "PUT",
|
|
f"/api/hockey/games/{game}/control/timers", {"main_timer": {
|
|
"current_ms": self.timer_values[game], "running": False, "paused": True},
|
|
"penalty_board": {"penalties": [], "history": []}})
|
|
started = time.perf_counter()
|
|
await self.update_selection(name, "IMPORT-HELD")
|
|
assert not self.heavy_release.is_set()
|
|
self.interactive_results.append({"account": name, "game": game, "kind": "context_and_mapping",
|
|
"status": 200, "completed_before_release": True,
|
|
"milliseconds": round((time.perf_counter() - started) * 1000, 3)})
|
|
result = await self.interactive_request(name, "timer_transport_ack", "POST", "/api/hockey/vmix/sequence", {
|
|
"commands": [{"Function": "SuspendCountdown", "Input": "Shared Score", "SelectedName": "Clock.Text"}],
|
|
"device_id": person["device"], "session_token": self.tokens[name]})
|
|
assert result.get("ok"), result
|
|
await self.reject_viewer_write(f"view{i}")
|
|
response = await self.http.get(f"/api/hockey/games/{game}/control", headers=self.basic_headers(f"view{i}"))
|
|
assert response.status_code == 200, response.text
|
|
payload = response.json()
|
|
assert payload["control_access"]["game_id"] == game
|
|
assert payload["timers"]["main_timer"]["current_ms"] == self.timer_values[game]
|
|
assert not self.heavy_release.is_set()
|
|
await asyncio.gather(*(match_pipeline(i) for i in range(1, 4)))
|
|
assert (await self.tick()).get("ok")
|
|
self.verify_all_current()
|
|
await self.verify_context_scopes()
|
|
assert not self.heavy_release.is_set()
|
|
assert len(self.timer_command_frames) == 3
|
|
assert self.import_results == 0 and self.store_calls == 1
|
|
await self.health_probe()
|
|
self.before_release_verified = True
|
|
|
|
async def pressure(self):
|
|
await self.refresh_leases(force=True)
|
|
holder = asyncio.create_task(self.one_import())
|
|
waiters, background = [], []
|
|
try:
|
|
await asyncio.wait_for(self.import_entered.wait(), timeout=10)
|
|
waiters = [asyncio.create_task(self.one_import()) for _ in range(self.args.waiters)]
|
|
await asyncio.wait_for(self.waiters_started.wait(), timeout=10)
|
|
await asyncio.sleep(.05) # Let each public import reach its admission wait.
|
|
self.admission_observations["before_interactive"] = self.worker_snapshot()
|
|
assert self.admission_observations["before_interactive"]["shared_db_workers_borrowed"] == 1, self.admission_observations
|
|
assert not holder.done() and not any(task.done() for task in waiters)
|
|
background = [asyncio.create_task(self.poll_viewers()), asyncio.create_task(self.observe_event_loop())]
|
|
await asyncio.wait_for(self.interactive_work(), timeout=self.args.interactive_timeout)
|
|
self.admission_observations["after_interactive_before_release"] = self.worker_snapshot()
|
|
assert self.before_release_verified and not self.holder_timed_out
|
|
finally:
|
|
self.heavy_release.set()
|
|
self.stop_polling.set()
|
|
if background:
|
|
await asyncio.gather(*background, return_exceptions=True)
|
|
settled = await asyncio.wait_for(asyncio.gather(holder, *waiters, return_exceptions=True), timeout=60)
|
|
errors = [repr(item) for item in settled if isinstance(item, BaseException)]
|
|
if errors:
|
|
raise AssertionError({"import_task_errors": errors})
|
|
assert self.import_results == self.args.waiters + 1 and self.store_calls == self.args.waiters + 1
|
|
assert not self.import_errors and not self.holder_timed_out
|
|
from sqlalchemy import func, select
|
|
from hockey_data.async_work import run_db
|
|
from hockey_data.models import Country
|
|
def verify_country():
|
|
with self.db.session() as session:
|
|
assert session.scalar(select(func.count()).select_from(Country).where(Country.external_id == "191")) == 1
|
|
country = session.scalar(select(Country).where(Country.external_id == "191"))
|
|
assert country.iso2 == "RU" and country.iso3 == "RUS"
|
|
await run_db(verify_country)
|
|
await self.drain()
|
|
self.admission_observations["after_release_and_drain"] = self.worker_snapshot()
|
|
assert self.admission_observations["after_release_and_drain"]["shared_db_workers_borrowed"] == 0
|
|
|
|
async def run(self):
|
|
try:
|
|
for phase in ("bootstrap", "pressure"):
|
|
self.current_phase = phase
|
|
started = time.perf_counter()
|
|
print(f"import-admission phase={phase} started", file=sys.stderr, flush=True)
|
|
await asyncio.wait_for(getattr(self, phase)(), timeout=self.args.phase_timeout)
|
|
self.phase_results.append({"phase": phase, "ok": True, "seconds": round(time.perf_counter() - started, 3)})
|
|
print(f"import-admission phase={phase} passed", file=sys.stderr, flush=True)
|
|
self.assert_isolation()
|
|
assert self.before_release_verified and self.viewer_reads > 0
|
|
return self.report(True)
|
|
finally:
|
|
self.heavy_release.set()
|
|
self.stop_polling.set()
|
|
await self.hub.stop_auto_refresh()
|
|
await self.service.stop_live_refreshes()
|
|
await self.service.client.aclose()
|
|
await self.http.aclose()
|
|
self.db.engine.dispose()
|
|
|
|
def report(self, ok):
|
|
result = super().report(ok)
|
|
result.update(scenario="shared_import_admission", expected_negative_acks=0,
|
|
source="real sync_country_directory/persist_countries/import RLock",
|
|
imports_started=self.import_calls, imports_finished=self.import_results,
|
|
expected_imports=self.args.waiters + 1, store_entries=self.store_calls,
|
|
heavy_import_released_after_interactive=self.before_release_verified,
|
|
heavy_import_wait_ms=self.holder_wait_ms, heavy_import_timed_out=self.holder_timed_out,
|
|
import_errors=self.import_errors, admission_observations=self.admission_observations,
|
|
timer_command_frames=len(self.timer_command_frames), interactive_results=self.interactive_results)
|
|
result["limits"].append("Heavy import is deliberately held before its first SQL; database writer-lock contention is outside this admission test")
|
|
return result
|
|
|
|
return AdmissionScenario
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--project", type=Path, default=Path(__file__).resolve().parents[1],
|
|
help="Project root (defaults to the parent of tests/)")
|
|
modes = parser.add_mutually_exclusive_group()
|
|
modes.add_argument("--run", action="store_true")
|
|
modes.add_argument("--plan", action="store_true")
|
|
parser.add_argument("--waiters", type=int, default=12)
|
|
parser.add_argument("--fields", type=int, default=2)
|
|
parser.add_argument("--db-workers", type=int, default=4)
|
|
parser.add_argument("--interactive-timeout", type=float, default=5)
|
|
parser.add_argument("--hold-timeout", type=float, default=10)
|
|
parser.add_argument("--phase-timeout", type=float, default=180)
|
|
parser.add_argument("--ack-delay-ms", type=float, default=1)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
args.ticks = 1
|
|
if not (1 <= args.waiters <= 256 and 2 <= args.fields <= 200 and 2 <= args.db_workers <= 32):
|
|
parser.error("waiters 1..256, fields 2..200, db-workers 2..32")
|
|
if not (0 <= args.ack_delay_ms <= 1000 and 0 < args.interactive_timeout < args.hold_timeout < args.phase_timeout):
|
|
parser.error("Require 0 < interactive-timeout < hold-timeout < phase-timeout and ack-delay-ms 0..1000")
|
|
project = args.project.resolve()
|
|
if not (project / "tests/multi_match_harness.py").is_file():
|
|
parser.error("Project must contain the multi-match harness")
|
|
if not args.run:
|
|
print(json.dumps({"status": "prepared_not_run", "api_required": "async_work.run_import_db; country persist admitted before run_db",
|
|
"hold": "actual service._import_lock inside real sync_country_directory, before SQL",
|
|
"imports": 1 + args.waiters, "matches": 3, "agents": 6,
|
|
"must_finish_before_release": ["3 timer writes", "3 context writes",
|
|
"6 Agent SQL/Mapping updates", "3 acknowledged SuspendCountdown actions", "viewer reads and denied writes"],
|
|
"external_network": "disabled"}, indent=2))
|
|
return
|
|
sys.path.insert(0, str(project))
|
|
def denied(*_args, **_kwargs):
|
|
raise AssertionError("External sockets are disabled")
|
|
socket.socket.connect = socket.socket.connect_ex = socket.create_connection = denied
|
|
from hockey_data import async_work
|
|
if not hasattr(async_work, "run_import_db"):
|
|
parser.error("BUILD192 run_import_db API is not available; do not run the integration before it is ready")
|
|
os.environ["HOCKEY_DB_WORKERS"] = str(args.db_workers)
|
|
base_source = project / "tests/multi_match_harness.py"
|
|
baseline_sha = hashlib.sha256(base_source.read_bytes()).hexdigest()
|
|
base = load_base(project)
|
|
cls = build_scenario(base)
|
|
scenario = None
|
|
with tempfile.TemporaryDirectory(prefix="hockey-import-admission-") as temp:
|
|
directory = Path(temp)
|
|
os.environ["HOCKEY_UI_DATA_DIR"] = str(directory / "settings/local")
|
|
async def run():
|
|
nonlocal scenario
|
|
scenario = cls(project, directory, args)
|
|
return await scenario.run()
|
|
try:
|
|
report = asyncio.run(run())
|
|
except Exception as error:
|
|
report = scenario.report(False) if scenario else {"ok": False}
|
|
report.update(error=repr(error), traceback=traceback.format_exc())
|
|
report["reused_harness_sha256"] = baseline_sha
|
|
report["historical_build191_harness_sha256"] = base.BUILD191_HARNESS_SHA256
|
|
assert hashlib.sha256(base_source.read_bytes()).hexdigest() == baseline_sha, "Base harness was unexpectedly modified"
|
|
text = json.dumps(report, ensure_ascii=False, indent=2)
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(text + "\n", encoding="utf-8")
|
|
print(text)
|
|
raise SystemExit(0 if report["ok"] else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|