ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
calltoplay
lanspread/crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
T
ddidderr 4a1b08db98 test(peer-cli): verify protocol 8 transfer lifecycle 
Replace legacy metadata and relay expectations with current protocol-8 JSONL
assertions. Scenarios now bind every source to authenticated PeerId and exact
ContentId, prove typed attempt lifecycle order, and fence cancellation,
quarantine, rollback, republishing, and peer-departure outcomes against vacuous
success.

Isolated topologies distinguish direct author pulls from relay and ambient mDNS
substitution. The run log records focused diagnostics and the final fresh-image
S1-S49 acceptance result without presenting Docker-host throughput as a
representative external-LAN measurement.

Test Plan:
- `LANSPREAD_S37_MIN_MIB_PER_S=100 just peer-cli-tests` -- passed S1-S49
- S37 -- passed 2,147,483,656 bytes in 17 chunks at 551.10 MiB/s
- `python3 -m py_compile crates/lanspread-peer-cli/scripts/run_extended_scenarios.py` -- passed
- `ruff check --select F,E9 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py` -- passed
- `git diff --cached --check` -- passed
2026-08-10 14:00:03 +02:00

4214 lines
171 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Run the peer-cli scenarios S1-S49 through Docker."""
from __future__ import annotations
import argparse
import base64
import binascii
import hashlib
import json
import os
import queue
import shlex
import shutil
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable
REPO = Path(__file__).resolve().parents[3]
RUN_ROOT = REPO / ".lanspread-peer-cli" / "extended-scenarios"
IMAGE = "lanspread-peer-cli:dev"
NETWORK = "lanspread"
CONTAINER_PREFIX = "lanspread-peer-cli-ext"
CATALOG_DB = "/app/game.db"
CATALOG_MANIFESTS = "/app/manifests"
FIXTURES = REPO / "crates" / "lanspread-peer-cli" / "fixtures"
CATALOG_PROFILES = REPO / "crates" / "lanspread-peer-cli" / "catalogs"
CHUNK_SIZE = 128 * 1024 * 1024
# Runtime liveness needs the 90-second stale threshold plus a 20-second ping
# tick before a departed peer and its remote state are removed. Keep a
# conservative harness margin; graceful shutdown intentionally sends no
# protocol Goodbye.
PEER_DEPARTURE_TIMEOUT_SECONDS = 125
CATALOG_VERSIONS = {
"alienswarm": "20190317",
"bf1942": "20160130",
"bfbc2": "20210416",
"cnc4": "20170204",
"cnctw": "20160128",
"cod5": "20160920",
"cod6": "20200315",
"coh": "20200907",
"css": "20240623",
"ggoo": "20200721",
}
PERF_GAME_ID = "bf1942"
PERF_GAME_VERSION = CATALOG_VERSIONS[PERF_GAME_ID]
PERF_GAME_SIZE = 2 * 1024 * 1024 * 1024
IGNORED_DIFF_NAMES = {".lanspread", ".lanspread.json", "local"}
class ScenarioError(RuntimeError):
pass
@dataclass
class LineWaiter:
seen: int = 0
@dataclass
class Peer:
runner: "Runner"
name: str
network: str = NETWORK
games_dir: Path | None = None
readonly_games: bool = False
tmpfs_size: str | None = None
fixtures: list[str] = field(default_factory=list)
extra_args: list[str] = field(default_factory=list)
catalog_dir: Path | None = None
process: subprocess.Popen[str] | None = None
output: list[dict[str, Any]] = field(default_factory=list)
raw_output: list[str] = field(default_factory=list)
events: queue.Queue[dict[str, Any]] = field(default_factory=queue.Queue)
condition: threading.Condition = field(default_factory=threading.Condition)
request_index: int = 0
ready_addr: str | None = None
peer_id: str | None = None
@property
def container_name(self) -> str:
return f"{CONTAINER_PREFIX}-{self.runner.run_id}-{self.name}"
@property
def host_games_dir(self) -> Path:
if self.games_dir is not None:
return self.games_dir
return self.runner.games_root / self.name
@property
def host_state_dir(self) -> Path:
return self.runner.state_root / self.name
def start(self) -> "Peer":
self.host_state_dir.mkdir(parents=True, exist_ok=True)
if self.tmpfs_size is None:
self.host_games_dir.mkdir(parents=True, exist_ok=True)
command = [
"docker",
"run",
"--rm",
"--init",
"--network",
self.network,
"--name",
self.container_name,
"-i",
"-v",
f"{self.host_state_dir}:/state",
]
if self.tmpfs_size is None:
mode = "ro" if self.readonly_games else "rw"
command.extend(["-v", f"{self.host_games_dir}:/games:{mode}"])
else:
command.extend(["--tmpfs", f"/games:size={self.tmpfs_size}"])
catalog_db = CATALOG_DB
catalog_manifests = CATALOG_MANIFESTS
if self.catalog_dir is not None:
if not (self.catalog_dir / "game.db").is_file():
raise ScenarioError(f"catalog profile has no game.db: {self.catalog_dir}")
if not (self.catalog_dir / "manifests").is_dir():
raise ScenarioError(f"catalog profile has no manifests/: {self.catalog_dir}")
command.extend(["-v", f"{self.catalog_dir}:/scenario-catalog:ro"])
catalog_db = "/scenario-catalog/game.db"
catalog_manifests = "/scenario-catalog/manifests"
command.extend(
[
IMAGE,
"--name",
self.name,
"--games-dir",
"/games",
"--state-dir",
"/state",
"--catalog-db",
catalog_db,
"--manifests-dir",
catalog_manifests,
]
)
for fixture in self.fixtures:
command.extend(["--fixture", fixture])
command.extend(self.extra_args)
self.runner.log(f"start {self.name}: {' '.join(command)}")
self.process = subprocess.Popen(
command,
cwd=REPO,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
threading.Thread(target=self._read_output, daemon=True).start()
self.wait_ready()
return self
def _read_output(self) -> None:
assert self.process is not None
assert self.process.stdout is not None
for line in self.process.stdout:
stripped = line.rstrip("\r\n")
with self.condition:
self.raw_output.append(stripped)
try:
parsed = json.loads(stripped)
except json.JSONDecodeError:
parsed = {"type": "raw", "line": stripped}
self.output.append(parsed)
self.condition.notify_all()
self.events.put(parsed)
def wait_ready(self) -> None:
line = self.wait_for(
lambda item: item.get("type") == "event"
and item.get("event") == "local-peer-ready",
timeout=20,
description=f"{self.name} local-peer-ready",
)
data = line["data"]
self.ready_addr = data["addr"]
self.peer_id = data["peer_id"]
def send(
self,
payload: dict[str, Any],
*,
expect_error: bool = False,
timeout: float = 20,
) -> dict[str, Any]:
assert self.process is not None
assert self.process.stdin is not None
self.request_index += 1
request_id = f"{self.name}-{self.request_index}"
payload = dict(payload)
payload["id"] = request_id
line = json.dumps(payload, separators=(",", ":"))
self.process.stdin.write(line + "\n")
self.process.stdin.flush()
response = self.wait_for(
lambda item: item.get("id") == request_id
and item.get("type") in {"result", "error"},
timeout=timeout,
description=f"{self.name} response to {payload.get('cmd')}",
)
if response["type"] == "error":
if expect_error:
return response
raise ScenarioError(f"{self.name} command failed: {response}")
if expect_error:
raise ScenarioError(f"{self.name} command unexpectedly succeeded: {response}")
return response
def wait_for(
self,
predicate: Callable[[dict[str, Any]], bool],
*,
timeout: float,
description: str,
waiter: LineWaiter | None = None,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
with self.condition:
if waiter is None:
start = 0
else:
start = waiter.seen
while True:
for index in range(start, len(self.output)):
item = self.output[index]
if predicate(item):
if waiter is not None:
waiter.seen = index + 1
return item
start = len(self.output)
if waiter is not None:
waiter.seen = start
remaining = deadline - time.monotonic()
if remaining <= 0:
tail = "\n".join(self.raw_output[-20:])
raise ScenarioError(
f"timed out waiting for {description} on {self.name}\n{tail}"
)
self.condition.wait(remaining)
def list_games(self) -> dict[str, Any]:
return self.send({"cmd": "list-games"})["data"]
def list_peers(self) -> list[dict[str, Any]]:
return self.send({"cmd": "list-peers"})["data"]["peers"]
def status(self) -> dict[str, Any]:
return self.send({"cmd": "status"})["data"]
def call_to_play_events(self) -> list[dict[str, Any]]:
return self.send({"cmd": "list-call-to-play"})["data"]["view"]["events"]
def publish_call_to_play(self, intent: dict[str, Any]) -> dict[str, Any]:
return self.send(
{"cmd": "publish-call-to-play", "intent": intent}
)["data"]["receipt"]
def set_call_to_play_display_name(self, display_name: str) -> None:
self.send(
{
"cmd": "set-call-to-play-display-name",
"display_name": display_name,
}
)
def ready_endpoint(self) -> dict[str, str]:
if self.ready_addr is None or self.peer_id is None:
raise ScenarioError(f"{self.name} is not ready")
return {"peer_id": self.peer_id, "addr": self.ready_addr}
def connect_to(self, other: "Peer") -> None:
self.send({"cmd": "connect", **other.ready_endpoint()})
self.send({"cmd": "wait-peers", "count": 1, "timeout_ms": 10000})
def connect_to_at(self, other: "Peer", addr: str, peer_count: int) -> None:
if other.peer_id is None:
raise ScenarioError(f"{other.name} has no ready peer identity")
self.send({"cmd": "connect", "peer_id": other.peer_id, "addr": addr})
self.send({"cmd": "wait-peers", "count": peer_count, "timeout_ms": 10000})
def shutdown(self) -> None:
if self.process is None:
return
if self.process.poll() is None:
try:
self.send({"cmd": "shutdown"}, timeout=8)
except Exception:
self.kill()
try:
self.process.wait(timeout=8)
except subprocess.TimeoutExpired:
self.kill()
def shutdown_gracefully_for_proof(self) -> None:
if self.process is None or self.process.poll() is not None:
raise ScenarioError(
f"{self.name} was not running at the strict graceful-shutdown fence"
)
result = self.send({"cmd": "shutdown"}, timeout=15)
if (
result.get("command") != "shutdown"
or result.get("data", {}).get("stopped") is not True
):
raise ScenarioError(
f"{self.name} returned an invalid graceful-shutdown proof: {result}"
)
try:
exit_code = self.process.wait(timeout=8)
except subprocess.TimeoutExpired as error:
raise ScenarioError(
f"{self.name} did not exit after its graceful-shutdown proof"
) from error
if exit_code != 0:
raise ScenarioError(
f"{self.name} exited {exit_code} after its graceful-shutdown proof"
)
def kill(self) -> None:
subprocess.run(
["docker", "rm", "-f", self.container_name],
cwd=REPO,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
if self.process is not None:
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
def docker_exec(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
return subprocess.run(
["docker", "exec", self.container_name, *args],
cwd=REPO,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
check=check,
)
class Runner:
def __init__(self, selected: set[str] | None = None, build_image: bool = False) -> None:
self.selected = selected
self.build_image = build_image
self.run_id = str(int(time.time()))
self.state_root = RUN_ROOT / "state"
self.games_root = RUN_ROOT / "games"
self.fixture_root = RUN_ROOT / "fixtures"
self.current_peers: list[Peer] = []
self.current_networks: list[str] = []
self.results: list[tuple[str, str]] = []
def log(self, message: str) -> None:
print(message, flush=True)
def run(self) -> None:
self.prepare()
scenarios: list[tuple[str, Callable[[], str]]] = [
("S1", self.s1_startup_scan),
("S2", self.s2_direct_connect_handshake),
("S3", self.s3_remote_aggregation),
("S4", self.s4_single_source_download),
("S5", self.s5_auto_install_download),
("S6", self.s6_manual_install_uninstall),
("S7", self.s7_duplicate_source_download),
("S8", self.s8_catalog_file_shape_failover),
("S9", self.s9_missing_game),
("S10", self.s10_shutdown_cleanup),
("S11", self.s11_same_identity_reconnect),
("S12", self.s12_transfer_serving_gates),
("S13", self.s13_exact_transfer_equality),
("S14", self.s14_large_multi_peer_chunking),
("S15", self.s15_three_way_version_skew),
("S16", self.s16_catalog_fanout_with_stale),
("S17", self.s17_catalog_byte_quarantine),
("S18", self.s18_redundant_source_drop),
("S19", self.s19_sole_source_drop),
("S20", self.s20_receiver_write_failure),
("S21", self.s21_add_game_propagation),
("S22", self.s22_remove_game_propagation),
("S23", self.s23_version_bump_propagation),
("S24", self.s24_two_clients_one_source),
("S25", self.s25_two_downloads_one_client),
("S26", self.s26_duplicate_download_rejection),
("S27", self.s27_self_connect_rejection),
("S28", self.s28_reconnect_generation_unit),
("S29", self.s29_empty_peer_participates),
("S30", self.s30_mesh_aggregation),
("S31", self.s31_bootstrapped_peer_source),
("S32", self.s32_reinstall_after_uninstall),
("S33", self.s33_install_after_mutation),
("S34", self.s34_many_small_files),
("S35", self.s35_unknown_game_filtered),
("S36", self.s36_catalog_singleton),
("S37", self.s37_single_source_download_throughput),
("S38", self.s38_first_play_launch_settings),
("S39", self.s39_streamed_install_local_only),
("S40", self.s40_streamed_receiver_not_source),
("S41", self.s41_solid_archive_streamed_install),
("S42", self.s42_streamed_install_retries_next_source),
("S43", self.s43_streamed_install_rejects_installed_game),
("S44", self.s44_corrupt_stream_rolls_back),
("S45", self.s45_sender_disconnect_mid_stream),
("S46", self.s46_receiver_cancel_mid_stream),
("S47", self.s47_multi_archive_streams_in_sorted_order),
("S48", self.s48_call_to_play_replication_and_late_join),
("S49", self.s49_terminal_call_to_play_late_join),
]
for scenario_id, scenario in scenarios:
if self.selected and scenario_id.lower() not in self.selected:
continue
self.cleanup_containers()
self.cleanup_scenario_networks()
self.current_peers = []
try:
self.log(f"\n== {scenario_id} ==")
evidence = scenario()
self.results.append((scenario_id, evidence))
self.log(f"{scenario_id} PASS: {evidence}")
finally:
self.stop_peers()
if not self.results:
raise ScenarioError("no scenarios selected")
self.log("\nSummary:")
for scenario_id, evidence in self.results:
self.log(f"- {scenario_id}: {evidence}")
def prepare(self) -> None:
reset_run_root()
self.state_root.mkdir(parents=True, exist_ok=True)
self.games_root.mkdir(parents=True, exist_ok=True)
self.fixture_root.mkdir(parents=True, exist_ok=True)
self.cleanup_containers()
self.cleanup_scenario_networks()
if self.build_image:
run(["just", "peer-cli-image"], "build peer-cli image")
run(["just", "peer-cli-net"], "prepare peer-cli docker network")
def cleanup_containers(self) -> None:
result = subprocess.run(
["docker", "ps", "-a", "--format", "{{.Names}}"],
cwd=REPO,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=True,
)
names = [
name
for name in result.stdout.splitlines()
if name.startswith(f"{CONTAINER_PREFIX}-")
]
if names:
subprocess.run(
["docker", "rm", "-f", *names],
cwd=REPO,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
def stop_peers(self) -> None:
for peer in reversed(self.current_peers):
peer.shutdown()
self.cleanup_containers()
self.cleanup_scenario_networks()
def create_scenario_network(self, name: str, *, internal: bool = False) -> str:
network = f"{CONTAINER_PREFIX}-{self.run_id}-{name}"
command = ["docker", "network", "create"]
if internal:
command.append("--internal")
command.append(network)
run(command, f"create {name} network")
self.current_networks.append(network)
return network
def cleanup_scenario_networks(self) -> None:
for network in reversed(self.current_networks):
subprocess.run(
["docker", "network", "rm", network],
cwd=REPO,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
self.current_networks = []
def connect_peer_to_network(self, peer: Peer, network: str) -> str:
run(
["docker", "network", "connect", network, peer.container_name],
f"attach {peer.name} to {network}",
)
inspect = run(
[
"docker",
"inspect",
"--format",
f"{{{{(index .NetworkSettings.Networks {json.dumps(network)}).IPAddress}}}}",
peer.container_name,
],
f"inspect {peer.name} address on {network}",
)
address = inspect.stdout.strip()
if not address or peer.ready_addr is None:
raise ScenarioError(f"{peer.name} has no address on {network}")
_, port = peer.ready_addr.rsplit(":", 1)
return f"{address}:{port}"
def peer(
self,
name: str,
*,
network: str = NETWORK,
games_dir: Path | None = None,
readonly_games: bool = False,
tmpfs_size: str | None = None,
fixtures: list[str] | None = None,
extra_args: list[str] | None = None,
catalog_dir: Path | None = None,
) -> Peer:
peer = Peer(
runner=self,
name=name,
network=network,
games_dir=games_dir,
readonly_games=readonly_games,
tmpfs_size=tmpfs_size,
fixtures=fixtures or [],
extra_args=extra_args or [],
catalog_dir=catalog_dir,
).start()
self.current_peers.append(peer)
return peer
def fixture_catalog(
self,
name: str,
game_root: Path,
*,
streamed_install: bool,
) -> Path:
output = self.fixture_root / f"{name}-catalog"
command = ["just", "fixture-catalog", str(output), str(game_root)]
if not streamed_install:
command = [
"just",
"fixture-download-only-catalog",
str(output),
game_root.name,
str(game_root),
]
run(command, f"publish {name} fixture catalog")
return output
def s1_startup_scan(self) -> str:
alpha = self.peer("s1-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
games = alpha.list_games()["local"]
expected = {"alienswarm", "bf1942", "ggoo"}
seen = {game["id"] for game in games}
if not expected.issubset(seen):
raise ScenarioError(f"fixture-alpha games missing: expected {expected}, saw {seen}")
for game in games:
if game["id"] in expected:
assert_game_state(game, downloaded=True, installed=False, availability="Ready")
return "fixture-alpha emitted ready local games alienswarm, bf1942, and ggoo"
def s2_direct_connect_handshake(self) -> str:
alpha = self.peer("s2-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s2-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
alpha.connect_to(bravo)
bravo.connect_to(alpha)
# Poll for library convergence rather than reading once: wait-peers only
# guarantees peer presence, and the mDNS path can upsert a peer with an
# empty library before its snapshot arrives.
bravo_content_id = catalog_content_id(bravo, "bfbc2")
wait_peer_has_game(alpha, bravo.peer_id, "bfbc2", bravo_content_id)
wait_remote_library_content(
alpha, "bfbc2", bravo_content_id, peer_count=1
)
peers = alpha.list_peers()
if len(peers) != 1 or peers[0]["peer_id"] == alpha.peer_id:
raise ScenarioError(f"bad alpha peers after connect: {peers}")
if peers[0]["game_count"] != 4:
raise ScenarioError(f"expected bravo game_count=4, got {peers}")
# Peer state is pulled by the dialing side; prove each direction with an
# explicit authenticated endpoint.
alpha_content_id = catalog_content_id(alpha, "alienswarm")
bravo_view = wait_peer_has_game(
bravo, alpha.peer_id, "alienswarm", alpha_content_id
)
wait_remote_library_content(
bravo, "alienswarm", alpha_content_id, peer_count=1
)
if bravo_view["game_count"] != 3:
raise ScenarioError(f"expected alpha game_count=3 on bravo, got {bravo_view}")
return "alpha<->bravo exchanged libraries: bravo had 4 games on alpha, alpha had 3 on bravo"
def s3_remote_aggregation(self) -> str:
alpha = self.peer("s3-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s3-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s3-client")
connect_many(client, [alpha, bravo])
expected_counts = {
"ggoo": 2,
"alienswarm": 1,
"bf1942": 1,
"bfbc2": 1,
"cnc4": 1,
"cnctw": 1,
}
for game_id, peer_count in expected_counts.items():
wait_remote_game(client, game_id, peer_count=peer_count)
return "empty client aggregated alpha/bravo with ggoo peer_count=2 and unique games peer_count=1"
def s4_single_source_download(self) -> str:
bravo = self.peer("s4-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s4-client")
connect_many(client, [bravo])
content_id = catalog_content_id(bravo, "bfbc2")
wait_peer_has_game(client, bravo.peer_id, "bfbc2", content_id)
wait_remote_library_content(client, "bfbc2", content_id, peer_count=1)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("download-begin", "bfbc2"), timeout=20, description="begin bfbc2", waiter=waiter)
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="finish bfbc2", waiter=waiter)
assert_chunk_authority(client, "bfbc2", content_id, {bravo.peer_id})
game = wait_local_game(client, "bfbc2", downloaded=True, installed=False)
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
if (client.host_games_dir / "bfbc2" / "local").exists():
raise ScenarioError("bfbc2 local/ exists after install=false")
return (
"bfbc2 exact ContentId matched manifest, raw peer snapshot, remote view, "
f"and every authenticated chunk; install=false state installed={game['installed']}"
)
def s5_auto_install_download(self) -> str:
bravo = self.peer("s5-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s5-client")
connect_many(client, [bravo])
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "cnctw"})
client.wait_for(event_is("download-finished", "cnctw"), timeout=60, description="finish cnctw", waiter=waiter)
client.wait_for(event_is("install-finished", "cnctw"), timeout=30, description="install cnctw", waiter=waiter)
wait_local_game(client, "cnctw", downloaded=True, installed=True)
if not (client.host_games_dir / "cnctw" / "local" / "fixture-payload.txt").is_file():
raise ScenarioError("cnctw install payload missing")
diff_game_dirs(FIXTURES / "fixture-bravo" / "cnctw", client.host_games_dir / "cnctw")
return "cnctw auto-installed, local fixture payload existed, root diff matched excluding local metadata"
def s6_manual_install_uninstall(self) -> str:
bravo = self.peer("s6-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s6-client")
connect_many(client, [bravo])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="finish bfbc2", waiter=waiter)
client.send({"cmd": "install", "game_id": "bfbc2"})
client.wait_for(event_is("install-finished", "bfbc2"), timeout=30, description="install bfbc2", waiter=waiter)
wait_local_game(client, "bfbc2", downloaded=True, installed=True)
client.send({"cmd": "uninstall", "game_id": "bfbc2"})
client.wait_for(event_is("uninstall-finished", "bfbc2"), timeout=30, description="uninstall bfbc2", waiter=waiter)
wait_local_game(client, "bfbc2", downloaded=True, installed=False)
if (client.host_games_dir / "bfbc2" / "local").exists():
raise ScenarioError("bfbc2 local/ remained after uninstall")
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
return "manual install/uninstall toggled installed state, removed local/, preserved downloaded root"
def s7_duplicate_source_download(self) -> str:
alpha = self.peer("s7-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s7-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s7-client")
connect_many(client, [alpha, bravo])
content_id = catalog_content_id(client, "ggoo")
wait_peer_has_game(client, alpha.peer_id, "ggoo", content_id)
wait_peer_has_game(client, bravo.peer_id, "ggoo", content_id)
wait_remote_library_content(client, "ggoo", content_id, peer_count=2)
wait_remote_game(client, "ggoo", peer_count=2)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "ggoo", "install": False})
client.wait_for(event_is("download-finished", "ggoo"), timeout=60, description="finish ggoo", waiter=waiter)
diff_game_dirs(FIXTURES / "fixture-alpha" / "ggoo", client.host_games_dir / "ggoo")
# The fixtures are byte-identical, so the diff alone is source-agnostic.
# Prove the duplicate-source path: download committed exactly once, every
# chunk came from the validated 2-peer set, both peers actually served,
# and nothing was fetched twice.
if count_events(client, "download-finished", "ggoo") != 1:
raise ScenarioError("ggoo did not finish exactly once")
assert_chunk_authority(client, "ggoo", content_id, {alpha.peer_id, bravo.peer_id})
if chunk_sources(client, "ggoo") != {alpha.peer_id, bravo.peer_id}:
raise ScenarioError(
f"expected both validated sources to serve ggoo, got {chunk_sources(client, 'ggoo')}"
)
assert_no_duplicate_chunks(client, "ggoo")
return "ggoo downloaded once, served from both validated sources with no duplicate chunks"
def s8_catalog_file_shape_failover(self) -> str:
game_id = "ggoo"
source_a_dir = self.fixture_root / "s8-source-a"
source_b_dir = self.fixture_root / "s8-source-b"
copy_game(game_id, source_a_dir)
copy_game(game_id, source_b_dir)
source_a = self.peer("s8-source-a", games_dir=source_a_dir)
source_b = self.peer("s8-source-b", games_dir=source_b_dir)
bad, good = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
# Production distributes the first catalog entry to the first source in
# authenticated (PeerId, address) order. Corrupt that source only after
# startup so this scenario deterministically exercises archive admission
# failure and retry regardless of Docker address or connection order.
with (bad.host_games_dir / game_id / f"{game_id}.eti").open("ab") as handle:
handle.write(b"conflict")
client = self.peer("s8-client")
connect_many(client, [bad, good])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, bad.peer_id, game_id, content_id)
wait_peer_has_game(client, good.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, game_id, peer_count=2, version=CATALOG_VERSIONS[game_id])
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(
event_is("download-finished", game_id),
timeout=60,
description="ggoo oversized-source failover",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
)
# The bad source's version.ini is still exact catalog content and is a
# valid successful chunk. Only its oversized archive must be rejected.
assert_chunk_authority(client, game_id, content_id, {bad.peer_id, good.peer_id})
archive_totals = chunk_totals(client, game_id, f"{game_id}.eti")
if set(archive_totals) != {good.peer_id}:
raise ScenarioError(
"oversized source supplied a successful archive chunk instead of failing over: "
f"{archive_totals}"
)
assert_no_duplicate_chunks(client, game_id)
diff_game_dirs(good.host_games_dir / game_id, client.host_games_dir / game_id)
bad_only = self.peer("s8-oversized-only-client")
connect_many(bad_only, [bad])
wait_peer_has_game(bad_only, bad.peer_id, game_id, content_id)
wait_remote_library_content(bad_only, game_id, content_id, peer_count=1)
wait_remote_game(bad_only, game_id, peer_count=1)
failed_start = len(bad_only.output)
failed_waiter = LineWaiter(failed_start)
bad_only.send({"cmd": "download", "game_id": game_id, "install": False})
bad_only.wait_for(
event_is("download-failed", game_id),
timeout=60,
description="oversized-only ggoo failure",
waiter=failed_waiter,
)
assert_download_status_trace(
bad_only,
game_id,
start=failed_start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(bad_only, failed_start, "download-finished", game_id)
assert_local_absent(bad_only, game_id)
assert_not_exists(bad_only.host_games_dir / game_id / "version.ini")
assert_not_exists(bad_only.host_games_dir / game_id / "local")
return (
"oversized source failed exact catalog-file admission and the honest source "
"completed; an oversized-only receiver published no version.ini or local/"
)
def s9_missing_game(self) -> str:
game_id = "cod6"
client = self.peer("s9-client")
start = len(client.output)
waiter = LineWaiter(start)
queued = client.send(
{"cmd": "download", "game_id": game_id, "install": False}
)
if queued.get("data", {}).get("queued") is not True:
raise ScenarioError(f"missing-source download was not queued: {queued}")
client.wait_for(
event_is("download-failed", game_id),
timeout=20,
description="asynchronous no-source download failure",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(client, start, "download-begin", game_id)
assert_no_event_since(client, start, "download-finished", game_id)
active_events = [
item
for item in client.output[start:]
if item.get("type") == "event"
and item.get("event") == "active-operations-changed"
and any(
operation.get("game_id") == game_id
for operation in item.get("data", {}).get("active_operations", [])
)
]
if active_events:
raise ScenarioError(
f"missing-source download entered active operations: {active_events}"
)
assert_no_active(client, game_id)
assert_local_absent(client, game_id)
game_root = client.host_games_dir / game_id
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "local")
assert_not_exists(game_root)
return (
"missing-source command queued, then emitted download-failed without begin, "
"active-operation, version.ini, local/, or game-root mutation"
)
def s10_shutdown_cleanup(self) -> str:
alpha = self.peer("s10-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s10-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
bravo.shutdown()
wait_departure_topology(
alpha,
expected_peer_ids=set(),
expected_remote_game_ids=set(),
phase="S10 bravo shutdown",
)
return (
"bravo shutdown converged through pinned liveness to an exact empty peer "
"set and remote-library view"
)
def s11_same_identity_reconnect(self) -> str:
alpha = self.peer("s11-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo_dir = FIXTURES / "fixture-bravo"
bravo = self.peer("s11-bravo", games_dir=bravo_dir, readonly_games=True)
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
first_peer = alpha.list_peers()[0]
first_addr = first_peer["addr"]
first_id = first_peer["peer_id"]
bravo.shutdown()
wait_departure_topology(
alpha,
expected_peer_ids=set(),
expected_remote_game_ids=set(),
phase="S11 old bravo generation shutdown",
)
bravo = self.peer("s11-bravo", games_dir=bravo_dir, readonly_games=True)
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
peers = alpha.list_peers()
# The real invariant is a single peer entry reusing the same identity (no
# duplicate). The listener address is an OS-assigned ephemeral port: it
# almost always differs across restarts, but asserting it MUST change
# tests the kernel's port allocator, not the peer, and can fail spuriously
# if the same port is reused. So we only require the identity to be stable.
if len(peers) != 1:
raise ScenarioError(f"expected one bravo peer after reconnect, got {peers}")
if peers[0]["peer_id"] != first_id:
raise ScenarioError(f"bravo peer id changed: {first_id} -> {peers[0]['peer_id']}")
changed = "new" if peers[0]["addr"] != first_addr else "same"
return f"bravo reused peer id {first_id} as a single entry at {changed} address {peers[0]['addr']}"
def s12_transfer_serving_gates(self) -> str:
output = run_just_test()
required = [
"wrong_identity_path_or_range_never_authorizes_an_entry",
"chunk_boundaries_are_exact_including_empty_files",
"admission_rejects_every_ineligible_v8_case_before_transfer_registration",
"stream_install_admission_separates_identity_capability_and_valid_payload",
]
missing = [name for name in required if f"{name} ... ok" not in output]
if missing:
raise ScenarioError(f"S12 unit proof tests did not run-and-pass: {missing}")
return "just test passed including catalog/sentinel/active/local-path serve gate tests"
def s13_exact_transfer_equality(self) -> str:
bravo = self.peer("s13-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
alpha = self.peer("s13-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
client = self.peer("s13-client")
connect_many(client, [bravo, alpha])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="bfbc2 finish", waiter=waiter)
client.send({"cmd": "download", "game_id": "alienswarm", "install": False})
client.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="alienswarm finish", waiter=waiter)
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", client.host_games_dir / "alienswarm")
return "small bfbc2 and large alienswarm transfers both diffed cleanly against sources"
def s14_large_multi_peer_chunking(self) -> str:
game_id = PERF_GAME_ID
source_dir = self.fixture_root / "s14-alpha"
# Four 128 MiB chunks so the balance assertion is meaningful: with two
# peers a fair split is 2+2 chunks (diff 0) and a 3+1 imbalance would
# exceed one CHUNK_SIZE. A 2-chunk file could never trip the check.
file_size = CHUNK_SIZE * 4
create_large_sparse_game(source_dir / game_id, size=file_size)
catalog_dir = self.fixture_catalog(
"s14", source_dir / game_id, streamed_install=False
)
alpha = self.peer("s14-alpha", games_dir=source_dir, catalog_dir=catalog_dir)
stage = self.peer("s14-stage", catalog_dir=catalog_dir)
connect_many(stage, [alpha])
waiter = LineWaiter(len(stage.output))
stage.send({"cmd": "download", "game_id": game_id, "install": False})
stage.wait_for(event_is("download-finished", game_id), timeout=180, description="stage finish", waiter=waiter)
diff_game_dirs(source_dir / game_id, stage.host_games_dir / game_id)
client = self.peer("s14-client", catalog_dir=catalog_dir)
connect_many(client, [alpha, stage])
wait_remote_game(client, game_id, peer_count=2, version=PERF_GAME_VERSION)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-finished", game_id), timeout=180, description="client finish", waiter=waiter)
diff_game_dirs(source_dir / game_id, client.host_games_dir / game_id)
totals = chunk_totals(client, game_id, f"{game_id}.eti")
if len(totals) != 2:
raise ScenarioError(f"expected .eti chunks from exactly two peers, got {totals}")
if sum(totals.values()) != file_size:
raise ScenarioError(f"chunk bytes {sum(totals.values())} != file size {file_size}: {totals}")
values = list(totals.values())
if max(values) - min(values) > CHUNK_SIZE:
raise ScenarioError(f"chunk totals not balanced within one chunk: {totals}")
return f"{game_id} ({file_size // (1024 * 1024)} MiB) split across two sources, balanced, diff matched: {totals}"
def s15_three_way_version_skew(self) -> str:
stale_dir = self.fixture_root / "s15-a"
wrong_content_dir = self.fixture_root / "s15-b"
exact_dir = self.fixture_root / "s15-c"
copy_game("cnc4", stale_dir, version="20150101")
copy_game("cnc4", wrong_content_dir, version=CATALOG_VERSIONS["cnc4"])
with (wrong_content_dir / "cnc4" / "cnc4.eti").open("ab") as archive:
archive.write(b"wrong-content-id")
wrong_catalog = self.fixture_catalog(
"s15-wrong-content",
wrong_content_dir / "cnc4",
streamed_install=False,
)
copy_game("cnc4", exact_dir, version=CATALOG_VERSIONS["cnc4"])
peers = [
self.peer("s15-a", games_dir=stale_dir),
self.peer(
"s15-b", games_dir=wrong_content_dir, catalog_dir=wrong_catalog
),
self.peer("s15-c", games_dir=exact_dir),
]
client = self.peer("s15-client")
connect_many(client, peers)
expected_content_id = catalog_content_id(client, "cnc4")
wrong_content_id = catalog_content_id(peers[1], "cnc4")
if wrong_content_id == expected_content_id:
raise ScenarioError("modified s15 fixture did not change ContentId")
wait_peer_without_game(client, peers[0].peer_id, "cnc4")
wait_peer_has_game(client, peers[1].peer_id, "cnc4", wrong_content_id)
wait_peer_has_game(client, peers[2].peer_id, "cnc4", expected_content_id)
wait_remote_library_content(
client, "cnc4", wrong_content_id, peer_count=1
)
wait_remote_library_content(
client, "cnc4", expected_content_id, peer_count=1
)
wait_remote_game(client, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "cnc4", "install": False})
client.wait_for(event_is("download-finished", "cnc4"), timeout=60, description="cnc4 finish", waiter=waiter)
assert_chunk_authority(
client, "cnc4", expected_content_id, {peers[2].peer_id}
)
diff_game_dirs(peers[2].host_games_dir / "cnc4", client.host_games_dir / "cnc4")
return (
"stale-version peer advertised nothing; wrong-ContentId peer stayed in the raw "
"view but was excluded; every verified chunk came from the exact catalog peer"
)
def s16_catalog_fanout_with_stale(self) -> str:
specs = [
("s16-a", "20180101"),
("s16-b", CATALOG_VERSIONS["alienswarm"]),
("s16-c", CATALOG_VERSIONS["alienswarm"]),
]
peer_dirs = []
for name, version in specs:
game_dir = self.fixture_root / name
copy_game("alienswarm", game_dir, version=version)
# Two 128 MiB chunks so the .eti can actually fan out across the two
# catalog-version peers; the stock 120 MiB fixture is a single chunk
# that can only ever come from one source.
inflate_archive_sparse(game_dir / "alienswarm", "alienswarm", CHUNK_SIZE * 2)
peer_dirs.append((name, game_dir))
catalog_dir = self.fixture_catalog(
"s16", peer_dirs[1][1] / "alienswarm", streamed_install=False
)
peers = [
self.peer(name, games_dir=game_dir, catalog_dir=catalog_dir)
for name, game_dir in peer_dirs
]
client = self.peer("s16-client", catalog_dir=catalog_dir)
connect_many(client, peers)
wait_remote_game(client, "alienswarm", peer_count=2, version=CATALOG_VERSIONS["alienswarm"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "alienswarm", "install": False})
client.wait_for(event_is("download-finished", "alienswarm"), timeout=180, description="alienswarm finish", waiter=waiter)
assert_only_chunk_sources(client, "alienswarm", {peers[1].peer_id, peers[2].peer_id})
totals = chunk_totals(client, "alienswarm", "alienswarm.eti")
if peers[0].peer_id in totals:
raise ScenarioError(f"stale peer contributed chunks: {totals}")
if set(totals) != {peers[1].peer_id, peers[2].peer_id}:
raise ScenarioError(f"expected .eti to fan out across both B and C, got {totals}")
diff_game_dirs(peers[1].host_games_dir / "alienswarm", client.host_games_dir / "alienswarm")
return f"catalog-version B/C peers split alienswarm.eti while stale A contributed zero; totals={totals}"
def s17_catalog_byte_quarantine(self) -> str:
game_id = "cnc4"
source_a_dir = self.fixture_root / "s17-source-a"
source_b_dir = self.fixture_root / "s17-source-b"
copy_game(game_id, source_a_dir)
copy_game(game_id, source_b_dir)
source_a = self.peer("s17-source-a", games_dir=source_a_dir)
source_b = self.peer("s17-source-b", games_dir=source_b_dir)
bad, good = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
corrupt_file_same_length(bad.host_games_dir / game_id / f"{game_id}.eti")
client = self.peer("s17-client")
connect_many(client, [bad, good])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, bad.peer_id, game_id, content_id)
wait_peer_has_game(client, good.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, "cnc4", peer_count=2, version=CATALOG_VERSIONS["cnc4"])
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(
event_is("download-finished", game_id),
timeout=60,
description="cnc4 retry after corrupt source",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
expect_invalid_source_retry=True,
)
assert_chunk_authority(client, game_id, content_id, {bad.peer_id, good.peer_id})
archive_totals = chunk_totals(client, game_id, f"{game_id}.eti")
if set(archive_totals) != {good.peer_id}:
raise ScenarioError(
"corrupt source supplied a successful archive chunk instead of being quarantined: "
f"{archive_totals}"
)
assert_no_duplicate_chunks(client, game_id)
diff_game_dirs(good.host_games_dir / game_id, client.host_games_dir / game_id)
# A fresh receiver has no runtime quarantine history. With only the
# same-length corrupt source available, catalog verification must fail
# and must not publish either the sentinel or local install state.
bad_only = self.peer("s17-bad-only-client")
connect_many(bad_only, [bad])
wait_peer_has_game(bad_only, bad.peer_id, game_id, content_id)
wait_remote_library_content(bad_only, game_id, content_id, peer_count=1)
wait_remote_game(bad_only, game_id, peer_count=1)
failed_start = len(bad_only.output)
failed_waiter = LineWaiter(failed_start)
bad_only.send({"cmd": "download", "game_id": game_id, "install": False})
bad_only.wait_for(
event_is("download-failed", game_id),
timeout=60,
description="all-bad cnc4 catalog failure",
waiter=failed_waiter,
)
assert_download_status_trace(
bad_only,
game_id,
start=failed_start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(bad_only, failed_start, "download-finished", game_id)
assert_local_absent(bad_only, game_id)
assert_not_exists(bad_only.host_games_dir / game_id / "version.ini")
assert_not_exists(bad_only.host_games_dir / game_id / "local")
return (
"same-length corrupt source was quarantined and retried via the honest source; "
"a fresh all-bad receiver failed without version.ini or local/"
)
def s18_redundant_source_drop(self) -> str:
game_id = "bf1942"
source_a_dir = self.fixture_root / "s18-a"
source_b_dir = self.fixture_root / "s18-b"
# Multi-chunk sparse archive so BOTH peers have more assigned .eti chunks
# than the per-peer stream window; a single-window file could finish before
# the force-kill lands. We verify the download SURVIVES a mid-download
# source kill (every byte still arrives, no download-failed, diff matches).
# Waiting for the lower-ID source to fill the stream window makes the
# interruption non-vacuous; the final lower-ID-planned offset must then be
# completed by the survivor. Sparse zero bytes are identical, so both
# sources match the one publisher-derived catalog authority.
file_size = CHUNK_SIZE * 16
create_large_sparse_game(source_a_dir / game_id, size=file_size)
create_large_sparse_game(source_b_dir / game_id, size=file_size)
catalog_dir = self.fixture_catalog(
"s18", source_a_dir / game_id, streamed_install=False
)
source_a = self.peer("s18-a", games_dir=source_a_dir, catalog_dir=catalog_dir)
source_b = self.peer("s18-b", games_dir=source_b_dir, catalog_dir=catalog_dir)
client = self.peer("s18-client", catalog_dir=catalog_dir)
connect_many(client, [source_a, source_b])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, source_a.peer_id, game_id, content_id)
wait_peer_has_game(client, source_b.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, game_id, peer_count=2)
killed, survivor = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-begin", game_id), timeout=20, description="download begin", waiter=waiter)
active_streams = wait_outbound_transfer(
killed, game_id, minimum=4, timeout=30
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_no_event_since(client, start, "download-failed", game_id)
killed.kill()
client.wait_for(event_is("download-finished", game_id), timeout=180, description="download finish", waiter=waiter)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
)
# Scan the WHOLE download window (start..) for download-failed. The old
# assert_no_event reused a waiter already advanced past download-finished,
# so it only saw the empty tail and could never fire.
assert_no_event_since(client, start, "download-failed", game_id)
diff_game_dirs(survivor.host_games_dir / game_id, client.host_games_dir / game_id)
assert_chunk_authority(
client, game_id, content_id, {source_a.peer_id, source_b.peer_id}
)
totals = chunk_totals(client, game_id, f"{game_id}.eti")
survivor_id = survivor.peer_id
if survivor_id is None:
raise ScenarioError("surviving source has no authenticated PeerId")
if survivor_id not in totals:
raise ScenarioError(f"surviving source served no .eti chunks: {totals}")
if sum(totals.values()) != file_size:
raise ScenarioError(f"download did not deliver the whole archive ({sum(totals.values())} != {file_size}): {totals}")
# Exact sources are ordered by PeerId. The lower-ID peer receives every
# even archive chunk, so this last even offset was originally assigned to
# `killed`; seeing the survivor complete it proves retry, not just its own
# initial half of the plan.
retried_offset = file_size - (2 * CHUNK_SIZE)
if not any(
item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
and item["data"].get("relative_path") == f"{game_id}.eti"
and item["data"].get("offset") == retried_offset
and item["data"].get("peer_id") == survivor_id
for item in client.output[start:]
):
raise ScenarioError(
"survivor did not retry the killed source's final planned archive chunk: "
f"offset={retried_offset} totals={totals}"
)
# We deliberately do NOT assert the exact per-source split. The killed
# source can serve a chunk or two before `docker rm -f` lands (a fast-LAN
# race), so requiring totals == {survivor: file_size} would be flaky. The
# robust proof of redundancy is that a source died mid-download yet every
# byte still arrived (sum == file_size plus the diff), the survivor served
# part of it, and no download-failed was emitted.
survivor_bytes = totals[survivor_id]
return (
f"lower-PeerId source killed with {active_streams} outbound streams; "
f"all {file_size} bytes delivered "
f"({survivor_bytes} from the survivor), no download-failed; diff matched; bytes={totals}"
)
def s19_sole_source_drop(self) -> str:
game_id = "bf1942"
source_dir = self.fixture_root / "s19-source"
# More chunks than the per-peer stream window ensure pending work remains
# when the force-kill interrupts four positively observed active streams.
create_large_sparse_game(source_dir / game_id, size=CHUNK_SIZE * 8)
catalog_dir = self.fixture_catalog(
"s19", source_dir / game_id, streamed_install=False
)
source = self.peer("s19-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s19-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, game_id, peer_count=1)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-begin", game_id), timeout=20, description="download begin", waiter=waiter)
active_streams = wait_outbound_transfer(
source, game_id, minimum=4, timeout=30
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_no_event_since(client, start, "download-failed", game_id)
# Forceful kill drops positively active streams without a departure signal.
source.kill()
terminal = client.wait_for(
event_is("download-failed", game_id),
timeout=120,
description="sole-source drop terminal failure",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_not_exists(client.host_games_dir / game_id / "version.ini")
wait_no_active(client, game_id)
assert_local_absent(client, game_id)
return (
f"sole-source forceful drop with {active_streams} outbound streams -> "
f"{terminal['event']}; version.ini absent; no ready row; no active op"
)
def s20_receiver_write_failure(self) -> str:
source_dir = self.fixture_root / "s20-source"
copy_game("alienswarm", source_dir)
source = self.peer("s20-source", games_dir=source_dir)
client = self.peer("s20-client", tmpfs_size="32m")
connect_many(client, [source])
wait_remote_game(client, "alienswarm", peer_count=1)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "alienswarm", "install": False})
client.wait_for(event_is("download-failed", "alienswarm"), timeout=90, description="download failed", waiter=waiter)
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal="download-failed",
reason="operation-failed",
)
client.docker_exec("test", "!", "-e", "/games/alienswarm/version.ini")
assert_no_active(client, "alienswarm")
return "32m tmpfs receiver emitted download-failed; /games/alienswarm/version.ini absent; active operations empty"
def s21_add_game_propagation(self) -> str:
alpha = self.peer("s21-alpha")
bravo_dir = self.fixture_root / "s21-bravo"
bravo_dir.mkdir(parents=True, exist_ok=True)
bravo = self.peer("s21-bravo", games_dir=bravo_dir)
connect_bidirectional(alpha, bravo)
assert len(alpha.list_peers()) == 1
stage_game_drop(bravo_dir, "cod5")
game = wait_remote_game(alpha, "cod5", peer_count=1)
return f"alpha saw {game['id']} from the existing bravo peer with peer_count={game['peer_count']}"
def s22_remove_game_propagation(self) -> str:
alpha = self.peer("s22-alpha")
bravo_dir = self.fixture_root / "s22-bravo"
copy_game("cod5", bravo_dir)
bravo = self.peer("s22-bravo", games_dir=bravo_dir)
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "cod5", peer_count=1)
shutil.rmtree(bravo_dir / "cod5")
wait_remote_absent(alpha, "cod5")
peers = alpha.list_peers()
if len(peers) != 1:
raise ScenarioError(f"expected bravo peer to remain, got {peers}")
return "alpha removed cod5 while keeping one bravo peer"
def s23_version_bump_propagation(self) -> str:
alpha = self.peer("s23-alpha")
bravo_dir = self.fixture_root / "s23-bravo"
copy_game("cnc4", bravo_dir, version="20160101")
bravo = self.peer("s23-bravo", games_dir=bravo_dir)
connect_bidirectional(alpha, bravo)
wait_remote_absent(alpha, "cnc4", timeout=5)
(bravo_dir / "cnc4" / "version.ini").write_text(CATALOG_VERSIONS["cnc4"], encoding="utf-8")
wait_remote_game(alpha, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
return "alpha observed stale cnc4 become catalog-version downloadable without reconnect"
def s24_two_clients_one_source(self) -> str:
source = self.peer("s24-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
c1 = self.peer("s24-client-a")
c2 = self.peer("s24-client-b")
connect_many(c1, [source])
connect_many(c2, [source])
waiter1 = LineWaiter(len(c1.output))
waiter2 = LineWaiter(len(c2.output))
c1.send({"cmd": "download", "game_id": "alienswarm", "install": False})
c2.send({"cmd": "download", "game_id": "alienswarm", "install": False})
c1.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="client-a finish", waiter=waiter1)
c2.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="client-b finish", waiter=waiter2)
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", c1.host_games_dir / "alienswarm")
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", c2.host_games_dir / "alienswarm")
wait_local_game(c1, "alienswarm", downloaded=True, installed=False)
wait_local_game(c2, "alienswarm", downloaded=True, installed=False)
# Responsiveness: the source still answers and still advertises its games.
if not any(g["id"] == "alienswarm" for g in source.list_games()["local"]):
raise ScenarioError("source no longer advertises alienswarm after serving two clients")
return "two concurrent clients finished alienswarm install=false; both diffs matched; source still responsive"
def s25_two_downloads_one_client(self) -> str:
source = self.peer("s25-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s25-client")
connect_many(client, [source])
start = len(client.output)
bfbc2_waiter = LineWaiter(start)
cnctw_waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.send({"cmd": "download", "game_id": "cnctw", "install": False})
client.wait_for(
event_is("download-finished", "bfbc2"),
timeout=60,
description="bfbc2 finish",
waiter=bfbc2_waiter,
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="cnctw finish",
waiter=cnctw_waiter,
)
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
diff_game_dirs(FIXTURES / "fixture-bravo" / "cnctw", client.host_games_dir / "cnctw")
wait_local_game(client, "bfbc2", downloaded=True, installed=False)
wait_local_game(client, "cnctw", downloaded=True, installed=False)
return "bfbc2 and cnctw concurrent downloads both finished install=false and diffed cleanly"
def s26_duplicate_download_rejection(self) -> str:
game_id = "bf1942"
source_dir = self.fixture_root / "s26-source"
# A large sparse archive keeps the first download active long enough that
# the duplicate request is guaranteed to race against an in-progress
# operation rather than a near-instant 3 MB transfer that may already be
# finished by the time the second command is read.
create_large_sparse_game(source_dir / game_id, size=CHUNK_SIZE * 2)
catalog_dir = self.fixture_catalog(
"s26", source_dir / game_id, streamed_install=False
)
source = self.peer("s26-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s26-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, game_id, peer_count=1)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": game_id, "install": False})
active = client.wait_for(
lambda item: item.get("type") == "event"
and item.get("event") == "active-operations-changed"
and any(
op.get("game_id") == game_id
for op in item.get("data", {}).get("active_operations", [])
),
timeout=20,
description="active operation",
waiter=waiter,
)
# Verify the operation kind, not just its presence (the only scenario that
# inspects the active_operations 'operation' field).
op = next(o for o in active["data"]["active_operations"] if o.get("game_id") == game_id)
if op.get("operation") != "Downloading":
raise ScenarioError(f"expected Downloading active operation, got {op}")
err = client.send(
{"cmd": "download", "game_id": game_id, "install": False},
expect_error=True,
)
if "operation already in progress" not in err["error"]:
raise ScenarioError(f"unexpected duplicate error: {err}")
client.wait_for(event_is("download-finished", game_id), timeout=180, description="first download finish", waiter=waiter)
diff_game_dirs(source_dir / game_id, client.host_games_dir / game_id)
return f"duplicate rejected while Downloading active ('{err['error']}'); first download diff matched"
def s27_self_connect_rejection(self) -> str:
alpha = self.peer("s27-alpha")
err = alpha.send(
{"cmd": "connect", **alpha.ready_endpoint()}, expect_error=True
)
if "cannot connect peer to itself" not in err["error"]:
raise ScenarioError(f"unexpected self-connect error: {err}")
peers = alpha.list_peers()
if peers:
raise ScenarioError(f"self-connect created peers: {peers}")
alpha.status()
return f"self-connect errored '{err['error']}'; peer list stayed empty"
def s28_reconnect_generation_unit(self) -> str:
output = run_just_test()
test_name = (
"peer_db::tests::"
"reconnect_gets_new_generation_and_stale_removal_loses_authority"
)
if f"{test_name} ... ok" not in output:
raise ScenarioError(f"S28 unit proof did not run-and-pass in just test output:\n{output}")
return (
"`just test` passed including reconnect generation fencing against "
"stale removal"
)
def s29_empty_peer_participates(self) -> str:
source_network = self.create_scenario_network(
"s29-source", internal=True
)
source = self.peer(
"s29-alpha",
network=source_network,
games_dir=FIXTURES / "fixture-alpha",
readonly_games=True,
)
empty = self.peer("s29-empty")
observer = self.peer("s29-observer")
connect_bidirectional(observer, empty)
peers = observer.list_peers()
if (
{peer.get("peer_id") for peer in peers} != {empty.peer_id}
or len(peers) != 1
or peers[0]["game_count"] != 0
):
raise ScenarioError(f"expected empty peer with zero games, got {peers}")
self.connect_peer_to_network(empty, source_network)
empty.connect_to(source)
content_id = catalog_content_id(source, "alienswarm")
wait_peer_has_game(empty, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(empty, "alienswarm", content_id, peer_count=1)
wait_remote_game(empty, "alienswarm", peer_count=1)
waiter = LineWaiter(len(empty.output))
empty.send({"cmd": "download", "game_id": "alienswarm", "install": False})
empty.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="empty download finish", waiter=waiter)
assert_chunk_authority(empty, "alienswarm", content_id, {source.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", empty.host_games_dir / "alienswarm")
wait_peer_has_game(observer, empty.peer_id, "alienswarm", content_id)
wait_remote_library_content(observer, "alienswarm", content_id, peer_count=1)
wait_remote_game(observer, "alienswarm", peer_count=1)
assert_exact_peer_ids(observer, {empty.peer_id})
return (
"observer remained authenticated only to the formerly empty peer, then saw "
"its exact alienswarm ContentId with peer_count=1 after the peer downloaded "
"directly from an isolated source; raw snapshot, remote view, and chunks agreed"
)
def s30_mesh_aggregation(self) -> str:
dirs = []
specs = [
("s30-a", [("ggoo", CATALOG_VERSIONS["ggoo"]), ("bf1942", CATALOG_VERSIONS["bf1942"])]),
("s30-b", [("ggoo", CATALOG_VERSIONS["ggoo"]), ("cnc4", CATALOG_VERSIONS["cnc4"])]),
("s30-c", [("cnc4", CATALOG_VERSIONS["cnc4"]), ("cod5", CATALOG_VERSIONS["cod5"])]),
("s30-d", [("cnctw", CATALOG_VERSIONS["cnctw"]), ("coh", CATALOG_VERSIONS["coh"])]),
("s30-e", [("cnctw", CATALOG_VERSIONS["cnctw"]), ("bf1942", CATALOG_VERSIONS["bf1942"])]),
]
peers = []
for name, games in specs:
game_dir = self.fixture_root / name
for game_id, version in games:
copy_game(game_id, game_dir, version=version)
dirs.append(game_dir)
peers.append(self.peer(name, games_dir=game_dir))
client = self.peer("s30-client")
connect_many(client, peers)
expected = {
"ggoo": (2, CATALOG_VERSIONS["ggoo"]),
"bf1942": (2, CATALOG_VERSIONS["bf1942"]),
"cnc4": (2, CATALOG_VERSIONS["cnc4"]),
"cod5": (1, CATALOG_VERSIONS["cod5"]),
"cnctw": (2, CATALOG_VERSIONS["cnctw"]),
"coh": (1, CATALOG_VERSIONS["coh"]),
}
for game_id, (peer_count, version) in expected.items():
wait_remote_game(client, game_id, peer_count=peer_count, version=version)
game_rows = client.list_games()["remote"]
ids = [game["id"] for game in game_rows]
if len(ids) != len(set(ids)):
raise ScenarioError(f"duplicate game rows: {ids}")
if any(peer["peer_id"] == client.peer_id for peer in client.list_peers()):
raise ScenarioError("client listed itself as a peer")
return f"client aggregated {len(expected)} IDs from 5 peers with expected peer_count/catalog versions"
def s31_bootstrapped_peer_source(self) -> str:
game_id = "alienswarm"
source = self.peer("s31-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bootstrap = self.peer("s31-bootstrap")
connect_many(bootstrap, [source])
content_id = catalog_content_id(bootstrap, game_id)
wait_peer_has_game(bootstrap, source.peer_id, game_id, content_id)
waiter = LineWaiter(len(bootstrap.output))
bootstrap.send({"cmd": "download", "game_id": game_id, "install": False})
bootstrap.wait_for(event_is("download-finished", game_id), timeout=90, description="bootstrap finish", waiter=waiter)
assert_chunk_authority(bootstrap, game_id, content_id, {source.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / game_id, bootstrap.host_games_dir / game_id)
source.kill()
third = self.peer("s31-third")
connect_many(third, [bootstrap])
wait_peer_has_game(third, bootstrap.peer_id, game_id, content_id)
wait_remote_library_content(third, game_id, content_id, peer_count=1)
assert_exact_peer_ids(third, {bootstrap.peer_id})
waiter = LineWaiter(len(third.output))
third.send({"cmd": "download", "game_id": game_id, "install": False})
third.wait_for(event_is("download-finished", game_id), timeout=90, description="third finish", waiter=waiter)
assert_chunk_authority(third, game_id, content_id, {bootstrap.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / game_id, third.host_games_dir / game_id)
return (
"bootstrap downloaded exact content from the original source, then served "
"every verified chunk to a third peer after the original source was killed; "
"diff matched original"
)
def s32_reinstall_after_uninstall(self) -> str:
source = self.peer("s32-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s32-client")
connect_many(client, [source])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="download finish", waiter=waiter)
client.send({"cmd": "install", "game_id": "bfbc2"})
client.wait_for(event_is("install-finished", "bfbc2"), timeout=30, description="install finish", waiter=waiter)
client.send({"cmd": "uninstall", "game_id": "bfbc2"})
client.wait_for(event_is("uninstall-finished", "bfbc2"), timeout=30, description="uninstall finish", waiter=waiter)
before = len(client.output)
client.send({"cmd": "install", "game_id": "bfbc2"})
reinstall_waiter = LineWaiter(before)
client.wait_for(event_is("install-finished", "bfbc2"), timeout=30, description="reinstall finish", waiter=reinstall_waiter)
recent = client.output[before:]
if any(item.get("event") == "download-chunk-finished" for item in recent):
raise ScenarioError("reinstall produced transfer chunk events")
wait_local_game(client, "bfbc2", downloaded=True, installed=True)
if not (client.host_games_dir / "bfbc2" / "local").is_dir():
raise ScenarioError("local/ was not recreated")
return "reinstall recreated local/, local state installed=true, no transfer events during reinstall"
def s33_install_after_mutation(self) -> str:
source = self.peer("s33-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s33-client")
connect_many(client, [source])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="download finish", waiter=waiter)
client.docker_exec(
"sh",
"-c",
"printf 'mutated archive bytes\\n' > /games/bfbc2/bfbc2.eti",
)
client.send({"cmd": "install", "game_id": "bfbc2"})
client.wait_for(event_is("install-finished", "bfbc2"), timeout=30, description="install finish", waiter=waiter)
client.docker_exec(
"cmp",
"/games/bfbc2/bfbc2.eti",
"/games/bfbc2/local/fixture-payload.txt",
)
return "fixture installer installed current mutated archive bytes exactly"
def s34_many_small_files(self) -> str:
source_dir = self.fixture_root / "s34-source"
create_many_small_game(source_dir / "bf1942")
catalog_dir = self.fixture_catalog(
"s34", source_dir / "bf1942", streamed_install=True
)
source = self.peer("s34-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s34-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, "bf1942", peer_count=1)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bf1942", "install": False})
client.wait_for(event_is("download-finished", "bf1942"), timeout=60, description="download finish", waiter=waiter)
diff_game_dirs(source_dir / "bf1942", client.host_games_dir / "bf1942")
chunks = [
item
for item in client.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "bf1942"
]
# 20 small files + version.ini, each a single coherent chunk: exactly 21
# chunk events, 21 distinct relative paths, no splits and no duplicates.
if len(chunks) != 21:
raise ScenarioError(f"expected exactly 21 file chunks (20 files + version.ini), got {len(chunks)}")
assert_no_duplicate_chunks(client, "bf1942")
distinct_paths = {item.get("data", {}).get("relative_path") for item in chunks}
if len(distinct_paths) != 21:
raise ScenarioError(f"expected 21 distinct file paths, got {len(distinct_paths)}: {sorted(distinct_paths)}")
return f"20 small files plus version.ini each transferred as one coherent chunk; diff matched; chunk events={len(chunks)}"
def s35_unknown_game_filtered(self) -> str:
game_id = "cod2"
source = self.peer(
"s35-source",
games_dir=FIXTURES / "fixture-unknown",
readonly_games=True,
catalog_dir=CATALOG_PROFILES / "unknown",
)
client = self.peer("s35-client")
connect_many(client, [source])
# The source has an honest, source-only catalog for cod2. The client uses
# the default reduced profile, where cod2 is intentionally unknown.
content_id = catalog_content_id(source, game_id)
wait_peer_has_game(client, source.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=1)
wait_remote_absent(client, game_id)
err = client.send(
{"cmd": "download", "game_id": game_id, "install": False},
expect_error=True,
)
if "not in the local catalog" not in err["error"]:
raise ScenarioError(f"unexpected unknown game error: {err}")
assert_not_exists(client.host_games_dir / game_id)
return (
f"raw v8 availability retained unknown {game_id}/{content_id}, but the local "
f"catalog excluded it from list-games; download errored '{err['error']}'"
)
def s36_catalog_singleton(self) -> str:
peers = []
for index in range(5):
game_dir = self.fixture_root / f"s36-{index}"
version = CATALOG_VERSIONS["cnc4"] if index == 0 else "20160101"
copy_game("cnc4", game_dir, version=version)
peers.append(self.peer(f"s36-{index}", games_dir=game_dir))
client = self.peer("s36-client")
connect_many(client, peers)
content_id = catalog_content_id(client, "cnc4")
wait_peer_has_game(client, peers[0].peer_id, "cnc4", content_id)
for stale in peers[1:]:
wait_peer_without_game(client, stale.peer_id, "cnc4")
wait_remote_library_content(client, "cnc4", content_id, peer_count=1)
wait_remote_game(client, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "cnc4", "install": False})
client.wait_for(event_is("download-finished", "cnc4"), timeout=60, description="download finish", waiter=waiter)
assert_chunk_authority(client, "cnc4", content_id, {peers[0].peer_id})
diff_game_dirs(peers[0].host_games_dir / "cnc4", client.host_games_dir / "cnc4")
return (
"catalog manifest, raw availability, remote view, and every verified chunk used "
"one exact ContentId/source; four stale peers advertised no transferable game"
)
def s37_single_source_download_throughput(self) -> str:
source_dir = self.fixture_root / "s37-source"
create_large_sparse_game(source_dir / PERF_GAME_ID, size=PERF_GAME_SIZE)
catalog_dir = self.fixture_catalog(
"s37", source_dir / PERF_GAME_ID, streamed_install=False
)
source = self.peer("s37-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s37-client", catalog_dir=catalog_dir)
connect_many(client, [source])
content_id = catalog_content_id(client, PERF_GAME_ID)
wait_peer_has_game(client, source.peer_id, PERF_GAME_ID, content_id)
wait_remote_library_content(
client, PERF_GAME_ID, content_id, peer_count=1
)
wait_remote_game(client, PERF_GAME_ID, peer_count=1, version=PERF_GAME_VERSION)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": PERF_GAME_ID, "install": False})
finished = client.wait_for(
event_is("download-finished", PERF_GAME_ID),
timeout=300,
description=f"{PERF_GAME_ID} throughput download",
waiter=waiter,
)
assert_download_status_trace(
client,
PERF_GAME_ID,
start=start,
terminal="download-finished",
)
destination_archive = client.host_games_dir / PERF_GAME_ID / f"{PERF_GAME_ID}.eti"
if destination_archive.stat().st_size != PERF_GAME_SIZE:
raise ScenarioError(
f"downloaded archive size mismatch: {destination_archive.stat().st_size} != {PERF_GAME_SIZE}"
)
diff_game_dirs(
source_dir / PERF_GAME_ID,
client.host_games_dir / PERF_GAME_ID,
)
throughput = finished.get("data", {}).get("throughput")
if not throughput:
raise ScenarioError(f"download-finished did not include throughput: {finished}")
expected_bytes = PERF_GAME_SIZE + len(PERF_GAME_VERSION)
if int(throughput["bytes"]) != expected_bytes:
raise ScenarioError(
f"throughput byte count mismatch: {throughput['bytes']} != {expected_bytes}"
)
chunks = [
item["data"]
for item in client.output[start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == PERF_GAME_ID
]
if len(chunks) != 17 or int(throughput["chunks"]) != 17:
raise ScenarioError(
"expected exactly 17 verified chunks "
f"(16 archive + version.ini), got events={len(chunks)} throughput={throughput['chunks']}"
)
assert_chunk_authority(client, PERF_GAME_ID, content_id, {source.peer_id})
archive_path = f"{PERF_GAME_ID}.eti"
expected_paths = {archive_path, "version.ini"}
actual_paths = {item.get("relative_path") for item in chunks}
if actual_paths != expected_paths:
raise ScenarioError(
f"verified chunk paths are not exact: {actual_paths} != {expected_paths}"
)
archive_chunks = [item for item in chunks if item.get("relative_path") == archive_path]
version_chunks = [
item
for item in chunks
if item.get("relative_path") == "version.ini"
]
expected_archive_chunks = [
(archive_path, index * CHUNK_SIZE, CHUNK_SIZE) for index in range(16)
]
actual_archive_chunks = sorted(
(
item.get("relative_path"),
int(item["offset"]),
int(item["length"]),
)
for item in archive_chunks
)
if actual_archive_chunks != expected_archive_chunks:
raise ScenarioError(
"archive chunk path/offset/length tuples are not exact: "
f"{actual_archive_chunks}"
)
expected_version_chunk = [("version.ini", 0, 8)]
actual_version_chunks = [
(
item.get("relative_path"),
int(item["offset"]),
int(item["length"]),
)
for item in version_chunks
]
if len(PERF_GAME_VERSION.encode("utf-8")) != 8:
raise ScenarioError(f"performance fixture version is not 8 bytes: {PERF_GAME_VERSION!r}")
if actual_version_chunks != expected_version_chunk:
raise ScenarioError(f"version.ini was not one exact chunk: {version_chunks}")
assert_no_duplicate_chunks(client, PERF_GAME_ID)
terminal_events = [
item.get("event")
for item in client.output[start:]
if item.get("type") == "event"
and item.get("data", {}).get("game_id") == PERF_GAME_ID
and item.get("event")
in {"download-finished", "download-failed"}
]
if terminal_events != ["download-finished"]:
raise ScenarioError(
f"expected exactly one successful terminal event: {terminal_events}"
)
game = wait_local_game(
client,
PERF_GAME_ID,
downloaded=True,
installed=False,
)
assert_game_state(
game,
downloaded=True,
installed=False,
availability="Ready",
)
wait_no_active(client, PERF_GAME_ID)
# The byte count alone never exercises the rate math. Validate the rate
# fields are positive and mutually consistent so a units/divisor bug
# (MiB vs MB, off-by-1000, zero duration) cannot slip through.
if throughput["duration_ms"] <= 0:
raise ScenarioError(f"throughput duration_ms not positive: {throughput}")
if throughput["mib_per_s"] <= 0 or throughput["mbit_per_s"] <= 0:
raise ScenarioError(f"throughput rates not positive: {throughput}")
derived_mib = (int(throughput["bytes"]) / 1_048_576) / (throughput["duration_ms"] / 1000.0)
if abs(derived_mib - throughput["mib_per_s"]) > max(1.0, derived_mib * 0.02):
raise ScenarioError(
f"mib_per_s {throughput['mib_per_s']} disagrees with bytes/duration {derived_mib}"
)
# mbit_per_s = bytes*8/s/1e6; mib_per_s = bytes/s/1048576 -> ratio is fixed.
ratio = throughput["mbit_per_s"] / throughput["mib_per_s"]
expected_ratio = 1_048_576 * 8 / 1_000_000 # 8.388608
if abs(ratio - expected_ratio) > 0.01:
raise ScenarioError(
f"mbit/mib ratio {ratio} != expected {expected_ratio}: {throughput}"
)
minimum_mib_per_s = float(os.environ.get("LANSPREAD_S37_MIN_MIB_PER_S", "100"))
if throughput["mib_per_s"] < minimum_mib_per_s:
raise ScenarioError(
f"throughput {throughput['mib_per_s']:.2f} MiB/s is below configured "
f"minimum {minimum_mib_per_s:.2f} MiB/s"
)
return (
f"{PERF_GAME_ID} {format_bytes(PERF_GAME_SIZE)} single-source download: "
f"{throughput['mib_per_s']:.2f} MiB/s, "
f"{throughput['mbit_per_s']:.2f} Mbit/s, "
f"{throughput['duration_ms'] / 1000.0:.3f}s, exactly 17 verified chunks; "
f"host={os.uname().nodename}, storage={RUN_ROOT}, profile=dynamic-s37, "
f"date={time.strftime('%Y-%m-%d')}"
)
def s38_first_play_launch_settings(self) -> str:
client_dir = self.fixture_root / "s38-client"
copy_game("css", client_dir)
client = self.peer(
"s38-client",
games_dir=client_dir,
extra_args=["--unrar", "/usr/local/bin/unrar"],
)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "install", "game_id": "css"})
client.wait_for(
event_is("install-finished", "css"),
timeout=30,
description="css install",
waiter=waiter,
)
wait_local_game(client, "css", downloaded=True, installed=True)
marker = client.host_state_dir / "games" / "css" / "launch_settings_applied"
if marker.exists():
raise ScenarioError("launch settings marker existed before first play")
local_root = client.host_games_dir / "css" / "local"
account_file = local_root / "profiles" / "local" / "account_name.txt"
language_file = local_root / "profiles" / "local" / "language.txt"
ini_file = (
local_root
/ "engine"
/ "bin"
/ "win64"
/ "steam_settings"
/ "SmartSteamEmu.ini"
)
for path in [account_file, language_file, ini_file]:
if not path.is_file():
raise ScenarioError(f"expected installed launch settings file: {path}")
if account_file.read_bytes() != b"stubaccount":
raise ScenarioError("installed account_name.txt did not contain the exact stub bytes")
if language_file.read_bytes() != b"english":
raise ScenarioError("installed language.txt did not contain the exact stub bytes")
expected_ini = (
b"[Settings]\r\n"
b"AppId = 240\r\n"
b"PersonaName = stubplayer\r\n"
b"Language = english\r\n"
)
if ini_file.read_bytes() != expected_ini:
raise ScenarioError(
"installed SmartSteamEmu.ini did not contain the exact CRLF stub bytes"
)
first = client.send(
{
"cmd": "play",
"game_id": "css",
"username": "Lan Hero",
"language": "german",
}
)["data"]["outcome"]
expected_first = {
"already_applied": False,
"account_name_written": True,
"language_written": True,
"persona_name_written": True,
}
if first != expected_first:
raise ScenarioError(f"unexpected first play outcome: {first}")
if not marker.is_file():
raise ScenarioError("launch settings marker was not written after first play")
if account_file.read_text(encoding="utf-8") != "Lan Hero":
raise ScenarioError("account_name.txt was not stamped with username")
if language_file.read_text(encoding="utf-8") != "german":
raise ScenarioError("language.txt was not stamped with language")
stamped_ini = ini_file.read_bytes()
if b"PersonaName = Lan Hero\r\n" not in stamped_ini:
raise ScenarioError("PersonaName was not stamped with CRLF preserved")
if b"AppId = 240\r\n" not in stamped_ini or b"Language = english\r\n" not in stamped_ini:
raise ScenarioError("SmartSteamEmu.ini sibling lines were not preserved")
client.docker_exec(
"sh",
"-c",
"printf resetaccount > /games/css/local/profiles/local/account_name.txt",
)
client.docker_exec(
"sh",
"-c",
"printf resetlang > /games/css/local/profiles/local/language.txt",
)
client.docker_exec(
"sh",
"-c",
"printf '[Settings]\\r\\nAppId = 240\\r\\n"
"PersonaName = resetplayer\\r\\nLanguage = english\\r\\n' > "
"/games/css/local/engine/bin/win64/steam_settings/SmartSteamEmu.ini",
)
second = client.send(
{
"cmd": "play",
"game_id": "css",
"username": "Second User",
"language": "french",
}
)["data"]["outcome"]
expected_second = {
"already_applied": True,
"account_name_written": False,
"language_written": False,
"persona_name_written": False,
}
if second != expected_second:
raise ScenarioError(f"unexpected second play outcome: {second}")
if account_file.read_text(encoding="utf-8") != "resetaccount":
raise ScenarioError("second play rewrote account_name.txt despite marker")
if language_file.read_text(encoding="utf-8") != "resetlang":
raise ScenarioError("second play rewrote language.txt despite marker")
if b"PersonaName = resetplayer\r\n" not in ini_file.read_bytes():
raise ScenarioError("second play rewrote PersonaName despite marker")
return "css first play stamped launch settings once; second play respected the marker"
def stream_install_cnctw(self, prefix: str) -> tuple[Peer, Peer]:
source_dir = self.fixture_root / f"{prefix}-bravo"
copy_game("cnctw", source_dir, version="20160128")
network = self.create_scenario_network(f"{prefix}-stream", internal=True)
source = self.peer(
f"{prefix}-bravo",
network=network,
games_dir=source_dir,
)
client = self.peer(f"{prefix}-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-begin", "cnctw"),
timeout=20,
description="stream begin cnctw",
waiter=waiter,
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="stream finish cnctw",
waiter=waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="stream install cnctw",
waiter=waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=start,
content_id=content_id,
source_peer_id=source.peer_id,
source_addr=source.ready_addr,
)
return source, client
def s39_streamed_install_local_only(self) -> str:
source, client = self.stream_install_cnctw("s39")
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
game_root = client.host_games_dir / "cnctw"
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
expected = {
"bin/cnctw-payload.bin": unrar_entry_sha256(
source, "cnctw", "bin/cnctw-payload.bin"
),
"data/cnctw-assets.dat": unrar_entry_sha256(
source, "cnctw", "data/cnctw-assets.dat"
),
}
actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if actual != expected:
raise ScenarioError(f"streamed local payload hashes mismatched: {actual} != {expected}")
streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
expected_bytes = 3 * 1024 * 1024
if streamed_bytes != expected_bytes:
raise ScenarioError(
f"streamed byte count mismatch: {streamed_bytes} != {expected_bytes}"
)
wait_no_outbound_transfer(source, "cnctw")
return (
"cnctw streamed into local/ only; root archive and version.ini absent; "
f"payload hashes={actual}; source outbound transfer drained"
)
def s40_streamed_receiver_not_source(self) -> str:
source, receiver = self.stream_install_cnctw("s40")
source.shutdown()
observer_network = self.create_scenario_network(
"s40-observer",
internal=True,
)
observer = self.peer("s40-observer", network=observer_network)
assert_exact_peer_ids(observer, set())
remote_view_start = len(observer.output)
receiver_addr = self.connect_peer_to_network(receiver, observer_network)
observer.connect_to_at(receiver, receiver_addr, 1)
assert_exact_peer_ids(observer, {receiver.peer_id})
wait_peer_without_game(observer, receiver.peer_id, "cnctw")
wait_remote_library_without_game(
observer,
"cnctw",
start=remote_view_start,
)
wait_remote_absent(observer, "cnctw", timeout=5)
start = len(observer.output)
waiter = LineWaiter(start)
queued = observer.send(
{"cmd": "download", "game_id": "cnctw", "install": False}
)
if queued.get("data", {}).get("queued") is not True:
raise ScenarioError(f"no-source cnctw download was not queued: {queued}")
observer.wait_for(
event_is("download-failed", "cnctw"),
timeout=20,
description="asynchronous no-source cnctw failure",
waiter=waiter,
)
assert_download_status_trace(
observer,
"cnctw",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(observer, start, "download-begin", "cnctw")
assert_no_event_since(observer, start, "download-finished", "cnctw")
active_events = [
item
for item in observer.output[start:]
if item.get("type") == "event"
and item.get("event") == "active-operations-changed"
and any(
operation.get("game_id") == "cnctw"
for operation in item.get("data", {}).get("active_operations", [])
)
]
if active_events:
raise ScenarioError(
f"no-source cnctw download entered active operations: {active_events}"
)
assert_no_active(observer, "cnctw")
assert_exact_peer_ids(observer, {receiver.peer_id})
wait_peer_without_game(observer, receiver.peer_id, "cnctw")
wait_remote_absent(observer, "cnctw", timeout=5)
assert_local_absent(observer, "cnctw")
assert_not_exists(observer.host_games_dir / "cnctw")
return (
"isolated observer roster changed from empty to exactly the stream-installed "
"receiver; raw/remote views omitted cnctw, and queued failure had no begin, "
"success, active-operation event, or local mutation"
)
def s41_solid_archive_streamed_install(self) -> str:
source_dir = self.fixture_root / "s41-solid-source"
source_game = source_dir / "cnctw"
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", source_game)
catalog_dir = CATALOG_PROFILES / "solid"
network = self.create_scenario_network("s41-stream", internal=True)
source = self.peer(
"s41-solid-source",
network=network,
games_dir=source_dir,
catalog_dir=catalog_dir,
)
assert_peer_rar_archive_solid(source, "cnctw")
client = self.peer(
"s41-solid-client",
network=network,
catalog_dir=catalog_dir,
)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="solid stream finish cnctw",
waiter=waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="solid stream install cnctw",
waiter=waiter,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{source.peer_id},
start=start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
game_root = client.host_games_dir / "cnctw"
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
expected = {
"bin/cnctw-solid-payload.bin": unrar_entry_sha256(
source, "cnctw", "bin/cnctw-solid-payload.bin"
),
"data/cnctw-solid-assets.dat": unrar_entry_sha256(
source, "cnctw", "data/cnctw-solid-assets.dat"
),
}
actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if actual != expected:
raise ScenarioError(
f"solid streamed payload hashes mismatched: {actual} != {expected}"
)
streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
expected_bytes = sum((game_root / "local" / rel).stat().st_size for rel in expected)
if streamed_bytes != expected_bytes:
raise ScenarioError(
f"solid streamed byte count mismatch: {streamed_bytes} != {expected_bytes}"
)
return (
"solid cnctw archive streamed through one local-only install; "
f"payload hashes={actual}, bytes={streamed_bytes}"
)
def s42_streamed_install_retries_next_source(self) -> str:
bad_dir = self.fixture_root / "s42-bad-source"
good_dir = self.fixture_root / "s42-good-source"
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", bad_dir / "cnctw")
copy_game("cnctw", good_dir, version="20160128")
network = self.create_scenario_network("s42-stream", internal=True)
# Learn and persist both installation identities before assigning the
# mismatching profile. Production orders sources by authenticated
# PeerId, not by Docker address or connection order.
source_a_probe = self.peer("s42-source-a", network=network)
source_b_probe = self.peer("s42-source-b", network=network)
lower_probe, upper_probe = sorted(
[source_a_probe, source_b_probe],
key=lambda peer: peer_id_sort_key(peer.peer_id),
)
lower_id = lower_probe.peer_id
upper_id = upper_probe.peer_id
lower_name = lower_probe.name
upper_name = upper_probe.name
source_a_probe.shutdown()
source_b_probe.shutdown()
bad = self.peer(lower_name, network=network, games_dir=bad_dir)
good = self.peer(upper_name, network=network, games_dir=good_dir)
if bad.peer_id != lower_id or good.peer_id != upper_id:
raise ScenarioError(
"S42 source identities changed across the profile restart: "
f"bad={bad.peer_id}/{lower_id}, good={good.peer_id}/{upper_id}"
)
assert_peer_rar_archive_solid(bad, "cnctw")
if peer_id_sort_key(bad.peer_id) >= peer_id_sort_key(good.peer_id):
raise ScenarioError(
"S42 requires the catalog-mismatched source to have the lower authenticated "
f"PeerId; bad={bad.peer_id}, good={good.peer_id}"
)
client = self.peer("s42-client", network=network)
connect_many(client, [bad, good])
assert_exact_peer_ids(client, {bad.peer_id, good.peer_id})
content_id = catalog_content_id(client, "cnctw")
source_content_ids = {
catalog_content_id(bad, "cnctw"),
catalog_content_id(good, "cnctw"),
}
if source_content_ids != {content_id}:
raise ScenarioError(
f"S42 sources did not share the client's exact ContentId: "
f"{source_content_ids} != {content_id}"
)
wait_peer_has_game(client, bad.peer_id, "cnctw", content_id)
wait_peer_has_game(client, good.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=2)
wait_remote_game(client, "cnctw", peer_count=2, version="20160128")
expected = {
"bin/cnctw-payload.bin": unrar_entry_sha256(
good, "cnctw", "bin/cnctw-payload.bin"
),
"data/cnctw-assets.dat": unrar_entry_sha256(
good, "cnctw", "data/cnctw-assets.dat"
),
}
if client.process is None:
raise ScenarioError("S42 client process was not retained")
client_pid = client.process.pid
first_start = len(client.output)
first_waiter = LineWaiter(first_start)
bad_first_edges = LineWaiter(len(bad.output))
good_first_edges = LineWaiter(len(good.output))
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
bad,
"cnctw",
waiter=bad_first_edges,
phase="first streamed install",
)
wait_outbound_transfer_cycle(
good,
"cnctw",
waiter=good_first_edges,
phase="first streamed install",
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="first retry stream finish cnctw",
waiter=first_waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="first retry stream install cnctw",
waiter=first_waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=first_start,
content_id=content_id,
source_peer_id=good.peer_id,
source_addr=good.ready_addr,
expect_invalid_source_retry=True,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{good.peer_id},
start=first_start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
game_root = client.host_games_dir / "cnctw"
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
first_actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if first_actual != expected:
raise ScenarioError(
f"first retry payload hashes mismatched: {first_actual} != {expected}"
)
first_streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output[first_start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
expected_bytes = 3 * 1024 * 1024
if first_streamed_bytes != expected_bytes:
raise ScenarioError(
f"first retry byte count mismatch: {first_streamed_bytes} != {expected_bytes}"
)
uninstall_start = len(client.output)
uninstall_waiter = LineWaiter(uninstall_start)
client.send({"cmd": "uninstall", "game_id": "cnctw"})
client.wait_for(
event_is("uninstall-finished", "cnctw"),
timeout=30,
description="uninstall before quarantine persistence retry",
waiter=uninstall_waiter,
)
wait_no_active(client, "cnctw")
assert_local_absent(client, "cnctw")
assert_not_exists(game_root / "local")
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
if client.process.poll() is not None or client.process.pid != client_pid:
raise ScenarioError("S42 client runtime changed before the quarantine retry")
assert_exact_peer_ids(client, {bad.peer_id, good.peer_id})
wait_peer_has_game(client, bad.peer_id, "cnctw", content_id)
wait_peer_has_game(client, good.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=2)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
assert_no_outbound_edge_during(
bad,
start=bad_first_edges.seen,
phase="first streamed install post-drain",
timeout=0.5,
)
assert_no_outbound_edge_during(
good,
start=good_first_edges.seen,
phase="first streamed install post-drain",
timeout=0.5,
)
second_bad_start = len(bad.output)
second_good_start = len(good.output)
good_second_edges = LineWaiter(second_good_start)
second_start = len(client.output)
second_waiter = LineWaiter(second_start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
good,
"cnctw",
waiter=good_second_edges,
phase="second streamed install",
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="quarantine persistence stream finish cnctw",
waiter=second_waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="quarantine persistence install finish cnctw",
waiter=second_waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=second_start,
content_id=content_id,
source_peer_id=good.peer_id,
source_addr=good.ready_addr,
)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
assert_no_outbound_edge_during(
good,
start=good_second_edges.seen,
phase="second streamed install post-drain",
timeout=0.5,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{good.peer_id},
start=second_start,
)
second_game = wait_local_game(
client,
"cnctw",
downloaded=False,
installed=True,
)
assert_game_state(
second_game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
second_actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if second_actual != expected:
raise ScenarioError(
f"second retry payload hashes mismatched: {second_actual} != {expected}"
)
second_streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output[second_start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
if second_streamed_bytes != expected_bytes:
raise ScenarioError(
f"second retry byte count mismatch: {second_streamed_bytes} != {expected_bytes}"
)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
bad.shutdown_gracefully_for_proof()
good.shutdown_gracefully_for_proof()
second_bad_edges = [
item
for item in bad.output[second_bad_start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if second_bad_edges:
raise ScenarioError(
"second streamed install admitted the quarantined bad source: "
f"{second_bad_edges}"
)
second_good_edges = [
item
for item in good.output[second_good_start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if len(second_good_edges) != 2:
raise ScenarioError(
"second streamed install did not emit exactly good admission/drain edges: "
f"{second_good_edges}"
)
return (
"first install admitted bad then good exact-content sources; after uninstall, "
"the same runtime skipped the quarantined bad source and admitted only good; "
f"good={good.peer_id}, bad={bad.peer_id}, bytes={second_streamed_bytes}"
)
def s43_streamed_install_rejects_installed_game(self) -> str:
_source, client = self.stream_install_cnctw("s43")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-failed", "cnctw"),
timeout=20,
description="already-installed stream rejection",
waiter=waiter,
)
assert_download_status_trace(
client,
"cnctw",
start=start,
terminal="download-failed",
reason="operation-failed",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(client, start, "install-finished", "cnctw")
assert_no_event_since(client, start, "download-finished", "cnctw")
wait_no_active(client, "cnctw")
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
return "already-installed cnctw rejected a second streamed install without state drift"
def s44_corrupt_stream_rolls_back(self) -> str:
source_dir = self.fixture_root / "s44-mismatched-source"
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", source_dir / "cnctw")
network = self.create_scenario_network("s44-stream", internal=True)
source = self.peer(
"s44-mismatched-source",
network=network,
games_dir=source_dir,
)
assert_peer_rar_archive_solid(source, "cnctw")
client = self.peer("s44-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
source_edges = LineWaiter(len(source.output))
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
source,
"cnctw",
waiter=source_edges,
phase="all-bad streamed install",
)
client.wait_for(
event_is("download-failed", "cnctw"),
timeout=30,
description="catalog-mismatched stream failed",
waiter=waiter,
)
assert_download_status_trace(
client,
"cnctw",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", "cnctw")
assert_no_event_since(client, start, "install-finished", "cnctw")
wait_no_active(client, "cnctw")
assert_failed_stream_left_no_local(client, "cnctw")
return (
"only source had a valid solid archive whose extracted output mismatched the "
"default catalog; integrity failure left no local install"
)
def s45_sender_disconnect_mid_stream(self) -> str:
source_dir = self.fixture_root / "s45-source"
copy_game("alienswarm", source_dir, version="20190317")
network = self.create_scenario_network("s45-stream", internal=True)
source = self.peer("s45-source", network=network, games_dir=source_dir)
client = self.peer("s45-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "alienswarm")
wait_peer_has_game(client, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(client, "alienswarm", content_id, peer_count=1)
wait_remote_game(client, "alienswarm", peer_count=1, version="20190317")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "alienswarm"})
first_chunk = client.wait_for(
event_is("download-chunk-finished", "alienswarm"),
timeout=30,
description="first alienswarm stream chunk before source drop",
waiter=waiter,
)
first_data = first_chunk.get("data", {})
if (
first_data.get("peer_id") != source.peer_id
or first_data.get("content_id") != content_id
):
raise ScenarioError(
f"sender-drop barrier used the wrong source/content: {first_data}"
)
source.kill()
client.wait_for(
event_is("download-failed", "alienswarm"),
timeout=60,
description="prompt sender transport failure",
waiter=waiter,
)
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", "alienswarm")
assert_no_event_since(client, start, "install-finished", "alienswarm")
assert_chunk_authority(
client,
"alienswarm",
content_id,
{source.peer_id},
start=start,
)
wait_no_active(client, "alienswarm")
assert_failed_stream_left_no_local(client, "alienswarm")
return (
"sender disconnect after first alienswarm chunk emitted prompt download-failed "
"and rolled back the stream before stale-peer liveness elapsed"
)
def s46_receiver_cancel_mid_stream(self) -> str:
source_dir = self.fixture_root / "s46-source"
copy_game("alienswarm", source_dir, version="20190317")
network = self.create_scenario_network("s46-stream", internal=True)
source = self.peer("s46-source", network=network, games_dir=source_dir)
client = self.peer("s46-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "alienswarm")
wait_peer_has_game(client, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(client, "alienswarm", content_id, peer_count=1)
wait_remote_game(client, "alienswarm", peer_count=1, version="20190317")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "alienswarm"})
first_chunk = client.wait_for(
event_is("download-chunk-finished", "alienswarm"),
timeout=30,
description="first alienswarm stream chunk before receiver cancel",
waiter=waiter,
)
first_data = first_chunk.get("data", {})
if (
first_data.get("peer_id") != source.peer_id
or first_data.get("content_id") != content_id
):
raise ScenarioError(
f"receiver-cancel barrier used the wrong source/content: {first_data}"
)
client.send({"cmd": "cancel-download", "game_id": "alienswarm"})
wait_no_active(client, "alienswarm", timeout=60)
assert_failed_stream_left_no_local(client, "alienswarm")
client.shutdown_gracefully_for_proof()
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal=None,
)
assert_no_event_since(client, start, "download-finished", "alienswarm")
assert_no_event_since(client, start, "download-failed", "alienswarm")
assert_no_event_since(client, start, "install-finished", "alienswarm")
assert_chunk_authority(
client,
"alienswarm",
content_id,
{source.peer_id},
start=start,
)
return "receiver cancel after first alienswarm chunk rolled back without failed event"
def s47_multi_archive_streams_in_sorted_order(self) -> str:
source_dir = self.fixture_root / "s47-source"
source_game = source_dir / "cnctw"
shutil.copytree(FIXTURES / "fixture-multi" / "cnctw", source_game)
catalog_dir = CATALOG_PROFILES / "multi"
network = self.create_scenario_network("s47-stream", internal=True)
source = self.peer(
"s47-source",
network=network,
games_dir=source_dir,
catalog_dir=catalog_dir,
)
client = self.peer(
"s47-client",
network=network,
catalog_dir=catalog_dir,
)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=30,
description="multi-archive stream finish",
waiter=waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="multi-archive stream install",
waiter=waiter,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{source.peer_id},
start=start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
game_root = client.host_games_dir / "cnctw"
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "a-first.eti")
assert_not_exists(game_root / "z-second.eti")
chunk_paths = streamed_chunk_paths(client, "cnctw")
expected_paths = [
"order/first.txt",
"order/second.txt",
]
if chunk_paths != expected_paths:
raise ScenarioError(f"multi-archive stream order mismatch: {chunk_paths}")
first = (game_root / "local" / "order" / "first.txt").read_text(encoding="utf-8")
second = (game_root / "local" / "order" / "second.txt").read_text(encoding="utf-8")
if first != "first archive payload\n" or second != "second archive payload\n":
raise ScenarioError(f"multi-archive payload mismatch: {first!r}, {second!r}")
return f"multi-archive cnctw streamed in sorted order: {chunk_paths}"
def s48_call_to_play_replication_and_late_join(self) -> str:
alice = self.peer("s48-alice")
bob = self.peer("s48-bob")
alice.set_call_to_play_display_name("Alice")
bob.set_call_to_play_display_name("Bob")
connect_bidirectional(alice, bob)
now = int(time.time() * 1000)
create = alice.publish_call_to_play({
"call_id": None,
"action": {
"Create": {
"game_id": "cnctw",
"max_players": 4,
"scheduled_for": None,
"deadline": now + 600_000,
}
},
})
call_id = create["call_id"]
wait_call_to_play_events(bob, {create["event_id"]})
rsvp = bob.publish_call_to_play({
"call_id": call_id,
"action": "Rsvp",
})
message = bob.publish_call_to_play({
"call_id": call_id,
"action": {
"SendMessage": {
"text": "I am in",
}
},
})
expected = {create["event_id"], rsvp["event_id"], message["event_id"]}
wait_call_to_play_events(alice, expected)
# Keep Alice<->Bob live on the ordinary network so Alice must retain
# Bob's slice throughout the no-relay proof. Charlie starts on an
# internal-only bridge shared with Alice: it has no route to Bob's
# ordinary-network address until Bob is explicitly attached below.
late_join_network = self.create_scenario_network(
"s48-late-join", internal=True
)
alice_late_join_addr = self.connect_peer_to_network(alice, late_join_network)
charlie = self.peer("s48-charlie", network=late_join_network)
charlie.set_call_to_play_display_name("Charlie")
charlie.connect_to_at(alice, alice_late_join_addr, 1)
wait_call_to_play_exact_events(alice, expected)
assert_exact_peer_ids(charlie, {alice.peer_id})
alice_only = wait_call_to_play_exact_events_from_exact_peers(
charlie,
{create["event_id"]},
{alice.peer_id},
)
wait_call_to_play_exact_events(alice, expected)
if any(event.get("id") in {rsvp["event_id"], message["event_id"]}
for event in alice_only):
raise ScenarioError(f"Alice relayed Bob-owned events to Charlie: {alice_only}")
bob_late_join_addr = self.connect_peer_to_network(bob, late_join_network)
charlie.connect_to_at(bob, bob_late_join_addr, 2)
events = wait_call_to_play_exact_events(charlie, expected)
if len(events) != 3:
raise ScenarioError(f"direct-author late-join view contains duplicates: {events}")
authors = {event["id"]: event["author_id"] for event in events}
if authors.get(create["event_id"]) != alice.peer_id or any(
authors.get(event_id) != bob.peer_id
for event_id in [rsvp["event_id"], message["event_id"]]
):
raise ScenarioError(f"direct-author attribution was not preserved: {events}")
bob.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
{create["event_id"]},
"Bob participant",
{alice.peer_id},
bob.peer_id,
)
alice.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
set(),
"Alice creator",
set(),
alice.peer_id,
)
return (
"late joiner pulled Alice and Bob directly; full-view replacement removed Bob's "
"participant slice on departure, then removed the call when Alice departed"
)
def s49_terminal_call_to_play_late_join(self) -> str:
alice = self.peer("s49-alice")
bob = self.peer("s49-bob")
alice.set_call_to_play_display_name("Alice")
bob.set_call_to_play_display_name("Bob")
connect_bidirectional(alice, bob)
now = int(time.time() * 1000)
create = alice.publish_call_to_play({
"call_id": None,
"action": {
"Create": {
"game_id": "cnctw",
"max_players": 4,
"scheduled_for": None,
"deadline": now + 600_000,
}
},
})
call_id = create["call_id"]
wait_call_to_play_events(bob, {create["event_id"]})
ready = bob.publish_call_to_play({
"call_id": call_id,
"action": {"Respond": {"ready_at": None}},
})
message = bob.publish_call_to_play({
"call_id": call_id,
"action": {
"SendMessage": {
"text": "Ready to launch",
}
},
})
participant_ids = {ready["event_id"], message["event_id"]}
wait_call_to_play_events(alice, {create["event_id"], *participant_ids})
start = alice.publish_call_to_play({
"call_id": call_id,
"action": "Start",
})
expected = {create["event_id"], *participant_ids, start["event_id"]}
wait_call_to_play_events(bob, expected)
late_join_network = self.create_scenario_network(
"s49-late-join", internal=True
)
alice_late_join_addr = self.connect_peer_to_network(alice, late_join_network)
charlie = self.peer("s49-charlie", network=late_join_network)
charlie.set_call_to_play_display_name("Charlie")
charlie.connect_to_at(alice, alice_late_join_addr, 1)
alice_ids = {create["event_id"], start["event_id"]}
wait_call_to_play_exact_events(alice, expected)
assert_exact_peer_ids(charlie, {alice.peer_id})
alice_only = wait_call_to_play_exact_events_from_exact_peers(
charlie,
alice_ids,
{alice.peer_id},
)
wait_call_to_play_exact_events(alice, expected)
if any(event.get("id") in participant_ids for event in alice_only):
raise ScenarioError(
f"Alice relayed Bob-owned terminal-call events to Charlie: {alice_only}"
)
bob_late_join_addr = self.connect_peer_to_network(bob, late_join_network)
charlie.connect_to_at(bob, bob_late_join_addr, 2)
events = wait_call_to_play_exact_events(charlie, expected)
if len(events) != 4:
raise ScenarioError(f"direct-author terminal late-join view is incomplete: {events}")
alice.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
set(),
"Alice creator",
{bob.peer_id},
alice.peer_id,
)
return (
"late joiner pulled terminal creator and participant slices directly; creator "
"departure removed the complete call despite the live participant author"
)
def run(command: list[str], description: str) -> subprocess.CompletedProcess[str]:
result = subprocess.run(
command,
cwd=REPO,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
if result.returncode != 0:
raise ScenarioError(f"{description} failed:\n{result.stdout}")
return result
def run_just_test() -> str:
env = os.environ.copy()
env["RUSTC_WRAPPER"] = ""
result = subprocess.run(
["just", "test"],
cwd=REPO,
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
if result.returncode != 0:
raise ScenarioError(f"just test failed:\n{result.stdout}")
return result.stdout
def reset_run_root() -> None:
if not RUN_ROOT.exists():
return
try:
shutil.rmtree(RUN_ROOT)
return
except PermissionError:
pass
subprocess.run(
[
"docker",
"run",
"--rm",
"-v",
f"{RUN_ROOT}:/cleanup",
"--entrypoint",
"/bin/sh",
"debian:bookworm-slim",
"-c",
"rm -rf /cleanup/* /cleanup/.[!.]* /cleanup/..?*",
],
cwd=REPO,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
shutil.rmtree(RUN_ROOT)
def copy_game(game_id: str, destination_games_dir: Path, *, version: str | None = None) -> None:
source = find_fixture_game(game_id)
destination = destination_games_dir / game_id
if destination.exists():
shutil.rmtree(destination)
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(source, destination)
version = version if version is not None else CATALOG_VERSIONS.get(game_id)
if version is not None:
(destination / "version.ini").write_text(version, encoding="utf-8")
def find_fixture_game(game_id: str) -> Path:
# Sorted so resolution is deterministic: several fixtures define the same
# game id (e.g. cnctw exists under fixture-bravo, fixture-multi and
# fixture-solid), and downstream assertions expect the fixture-bravo layout.
for fixture_dir in sorted(FIXTURES.iterdir()):
candidate = fixture_dir / game_id
if candidate.exists():
return candidate
raise ScenarioError(f"fixture game not found: {game_id}")
def stage_game_drop(destination_games_dir: Path, game_id: str) -> None:
source = find_fixture_game(game_id)
root = destination_games_dir / game_id
if root.exists():
shutil.rmtree(root)
root.mkdir(parents=True)
for child in source.iterdir():
if child.name == "version.ini":
continue
target = root / child.name
if child.is_dir():
shutil.copytree(child, target)
else:
shutil.copy2(child, target)
shutil.copy2(source / "version.ini", root / "version.ini")
def create_many_small_game(root: Path) -> None:
if root.exists():
shutil.rmtree(root)
root.mkdir(parents=True)
for index in range(20):
child = root / f"file-{index:02}.bin"
child.write_bytes(hashlib.sha256(f"small-{index}".encode()).digest() * 8)
(root / "version.ini").write_text(CATALOG_VERSIONS.get(root.name, "20250101"), encoding="utf-8")
def create_large_sparse_game(root: Path, *, size: int, version: str | None = None) -> None:
if root.exists():
shutil.rmtree(root)
root.mkdir(parents=True)
resolved = version if version is not None else CATALOG_VERSIONS.get(root.name, PERF_GAME_VERSION)
(root / "version.ini").write_text(resolved, encoding="utf-8")
archive = root / f"{root.name}.eti"
with archive.open("wb") as handle:
handle.truncate(size)
def inflate_archive_sparse(game_root: Path, game_id: str, size: int) -> None:
"""Replace a copied game's `.eti` with a sparse file of `size` bytes so the
archive spans multiple 128 MiB download chunks. Sparse zero bytes are
identical across copies, so one publisher-derived catalog manifest
authorizes every copy and a `diff` against any source still matches."""
archive = game_root / f"{game_id}.eti"
with archive.open("wb") as handle:
handle.truncate(size)
def corrupt_file_same_length(path: Path) -> None:
"""Flip one payload byte without changing the catalog-owned file shape."""
with path.open("r+b") as handle:
first = handle.read(1)
if not first:
raise ScenarioError(f"cannot corrupt empty fixture file: {path}")
handle.seek(0)
handle.write(bytes([first[0] ^ 0xFF]))
def sha256_file(path: Path) -> str:
hasher = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
hasher.update(chunk)
return hasher.hexdigest()
def unrar_entry_sha256(peer: Peer, game_id: str, relative_path: str) -> str:
command = (
f"unrar p -inul /games/{shlex.quote(game_id)}/{shlex.quote(game_id)}.eti "
f"{shlex.quote(relative_path)} | sha256sum"
)
output = peer.docker_exec("sh", "-c", command).stdout.strip()
if not output:
raise ScenarioError(f"empty sha256 output for {game_id}:{relative_path}")
return output.split()[0]
def assert_peer_rar_archive_solid(peer: Peer, game_id: str) -> None:
output = peer.docker_exec(
"unrar",
"lt",
"-cfg-",
f"/games/{game_id}/{game_id}.eti",
).stdout
for line in output.splitlines():
stripped = line.strip()
if stripped.startswith("Details:"):
if "solid" in stripped.lower():
return
raise ScenarioError(f"RAR archive is not solid: {game_id}")
raise ScenarioError(f"RAR archive details were not reported: {game_id}")
def peer_id_sort_key(peer_id: str | None) -> bytes:
if peer_id is None:
raise ScenarioError("cannot sort missing peer identity")
padded = peer_id.upper() + "=" * (-len(peer_id) % 8)
try:
decoded = base64.b32decode(padded)
except (binascii.Error, ValueError) as error:
raise ScenarioError(f"invalid peer identity {peer_id!r}") from error
if len(decoded) != 32:
raise ScenarioError(f"invalid peer identity length for {peer_id!r}")
return decoded
def format_bytes(size: int) -> str:
return f"{size / 1024 / 1024 / 1024:.2f} GiB"
def connect_many(client: Peer, peers: list[Peer]) -> None:
for peer in peers:
client.connect_to(peer)
client.send({"cmd": "wait-peers", "count": len(peers), "timeout_ms": 15000})
def connect_bidirectional(left: Peer, right: Peer) -> None:
left.connect_to(right)
right.connect_to(left)
wait_peer_visible(left, right.peer_id)
wait_peer_visible(right, left.peer_id)
def catalog_content_id(peer: Peer, game_id: str) -> str:
catalog_dir = peer.catalog_dir or CATALOG_PROFILES / "default"
manifest_path = catalog_dir / "manifests" / f"{game_id}.json"
try:
artifact = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise ScenarioError(
f"failed to load catalog manifest for {peer.name}/{game_id}: {manifest_path}: {error}"
) from error
content_id = artifact.get("content_id")
if (
not isinstance(content_id, str)
or len(content_id) != 64
or content_id.lower() != content_id
or any(char not in "0123456789abcdef" for char in content_id)
):
raise ScenarioError(
f"catalog manifest has invalid content_id for {peer.name}/{game_id}: {content_id!r}"
)
return content_id
def wait_remote_library_content(
peer: Peer,
game_id: str,
content_id: str,
*,
peer_count: int,
timeout: float = 20,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
last_view: list[dict[str, Any]] = []
while time.monotonic() < deadline:
latest = next(
(
item
for item in reversed(peer.output)
if item.get("type") == "event"
and item.get("event") == "remote-library-view"
),
None,
)
if latest is not None:
last_view = latest.get("data", {}).get("view", {}).get("games", [])
for row in last_view:
if (
row.get("game_id") == game_id
and row.get("content_id") == content_id
and row.get("peer_count") == peer_count
):
return row
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never received remote-library-view {game_id}/{content_id} "
f"peer_count={peer_count}; latest={last_view}"
)
def wait_remote_library_without_game(
peer: Peer,
game_id: str,
*,
start: int,
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_view: list[dict[str, Any]] | None = None
while time.monotonic() < deadline:
latest = next(
(
item
for item in reversed(peer.output[start:])
if item.get("type") == "event"
and item.get("event") == "remote-library-view"
),
None,
)
if latest is not None:
last_view = latest.get("data", {}).get("view", {}).get("games", [])
if all(row.get("game_id") != game_id for row in last_view):
return last_view
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never received a post-connect remote-library-view without "
f"{game_id}; latest={last_view}"
)
def wait_peer_visible(
observer: Peer,
peer_id: str | None,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
raise ScenarioError("cannot wait for a peer without peer_id")
deadline = time.monotonic() + timeout
last_peers: list[dict[str, Any]] = []
while time.monotonic() < deadline:
last_peers = observer.list_peers()
for peer in last_peers:
if peer.get("peer_id") == peer_id:
return peer
time.sleep(0.2)
raise ScenarioError(
f"{observer.name} never saw peer {peer_id}; peers={last_peers}"
)
def wait_remote_game(
peer: Peer,
game_id: str,
*,
peer_count: int | None = None,
version: str | None = None,
timeout: float = 20,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
last_rows: list[dict[str, Any]] = []
while time.monotonic() < deadline:
rows = peer.list_games()["remote"]
last_rows = rows
for row in rows:
if row["id"] != game_id:
continue
if peer_count is not None and row.get("peer_count") != peer_count:
continue
if version is not None and row.get("eti_game_version") != version:
continue
return row
time.sleep(0.4)
raise ScenarioError(
f"{peer.name} never saw remote {game_id} peer_count={peer_count} version={version}; rows={last_rows}"
)
def wait_remote_absent(peer: Peer, game_id: str, timeout: float = 20) -> None:
deadline = time.monotonic() + timeout
last_rows: list[dict[str, Any]] = []
while time.monotonic() < deadline:
rows = peer.list_games()["remote"]
last_rows = rows
if all(row["id"] != game_id for row in rows):
return
time.sleep(0.4)
raise ScenarioError(f"{peer.name} still lists remote {game_id}; rows={last_rows}")
def wait_departure_topology(
observer: Peer,
*,
expected_peer_ids: set[str | None],
expected_remote_game_ids: set[str],
phase: str,
) -> None:
if None in expected_peer_ids:
raise ScenarioError(f"{phase} cannot wait on a missing peer identity")
deadline = time.monotonic() + PEER_DEPARTURE_TIMEOUT_SECONDS
last_peers: list[dict[str, Any]] = []
last_remote: list[dict[str, Any]] = []
while time.monotonic() < deadline:
last_peers = observer.list_peers()
last_remote = observer.list_games()["remote"]
actual_peer_ids = {peer.get("peer_id") for peer in last_peers}
actual_remote_game_ids = {game.get("id") for game in last_remote}
if (
actual_peer_ids == expected_peer_ids
and actual_remote_game_ids == expected_remote_game_ids
):
return
time.sleep(0.4)
raise ScenarioError(
f"{phase} did not converge after the {PEER_DEPARTURE_TIMEOUT_SECONDS}s "
f"liveness allowance: expected peers={expected_peer_ids}, "
f"remote_games={expected_remote_game_ids}; peers={last_peers}, "
f"remote={last_remote}"
)
def wait_local_game(
peer: Peer,
game_id: str,
*,
downloaded: bool | None = None,
installed: bool | None = None,
timeout: float = 20,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
last_rows: list[dict[str, Any]] = []
while time.monotonic() < deadline:
rows = peer.list_games()["local"]
last_rows = rows
for row in rows:
if row["id"] != game_id:
continue
if downloaded is not None and row.get("downloaded") != downloaded:
continue
if installed is not None and row.get("installed") != installed:
continue
return row
time.sleep(0.4)
raise ScenarioError(
f"{peer.name} never reached local {game_id} downloaded={downloaded} installed={installed}; rows={last_rows}"
)
def wait_no_active(peer: Peer, game_id: str, timeout: float = 20) -> None:
deadline = time.monotonic() + timeout
last_active: list[dict[str, Any]] = []
while time.monotonic() < deadline:
active = peer.status()["active_operations"]
last_active = active
if all(item["game_id"] != game_id for item in active):
return
time.sleep(0.4)
raise ScenarioError(f"{peer.name} still has active operation for {game_id}: {last_active}")
def wait_outbound_transfer(
peer: Peer,
game_id: str,
*,
minimum: int = 1,
timeout: float = 20,
) -> int:
deadline = time.monotonic() + timeout
last_active: dict[str, int] = {}
while time.monotonic() < deadline:
active = peer.status()["active_outbound_transfers"]
last_active = active
count = int(active.get(game_id, 0))
if count >= minimum:
return count
time.sleep(0.02)
raise ScenarioError(
f"{peer.name} never reached {minimum} active outbound transfers for {game_id}: "
f"{last_active}"
)
def wait_no_outbound_transfer(peer: Peer, game_id: str, timeout: float = 20) -> None:
deadline = time.monotonic() + timeout
last_active: dict[str, int] = {}
while time.monotonic() < deadline:
active = peer.status()["active_outbound_transfers"]
last_active = active
if active.get(game_id, 0) == 0:
return
time.sleep(0.4)
raise ScenarioError(
f"{peer.name} still has outbound transfer for {game_id}: {last_active}"
)
def wait_outbound_transfer_cycle(
peer: Peer,
game_id: str,
*,
waiter: LineWaiter,
phase: str,
timeout: float = 30,
) -> None:
for edge in ["admission", "drain"]:
peer.wait_for(
event_is("outbound-transfer-count-changed"),
timeout=timeout,
description=f"{phase} {peer.name} outbound {edge} edge",
waiter=waiter,
)
wait_no_outbound_transfer(peer, game_id, timeout=timeout)
def assert_no_outbound_edge_during(
peer: Peer,
*,
start: int,
phase: str,
timeout: float = 1.0,
) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
edges = [
item
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if edges:
raise ScenarioError(
f"{phase} unexpectedly admitted an outbound transfer on {peer.name}: "
f"{edges}"
)
time.sleep(0.05)
def wait_call_to_play_events(
peer: Peer,
expected_ids: set[str],
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_events: list[dict[str, Any]] = []
while time.monotonic() < deadline:
events = peer.call_to_play_events()
last_events = events
if expected_ids <= {event.get("id") for event in events}:
return events
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never received Call to Play events {expected_ids}: {last_events}"
)
def wait_call_to_play_exact_events(
peer: Peer,
expected_ids: set[str],
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_events: list[dict[str, Any]] = []
while time.monotonic() < deadline:
events = peer.call_to_play_events()
last_events = events
if {event.get("id") for event in events} == expected_ids:
return events
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never converged to exact Call to Play replacement {expected_ids}: "
f"{last_events}"
)
def wait_call_to_play_departure_with_exact_remaining_peers(
peer: Peer,
expected_event_ids: set[str],
departed_author: str,
expected_remaining_peer_ids: set[str | None],
departed_peer_id: str | None,
) -> list[dict[str, Any]]:
if None in expected_remaining_peer_ids or departed_peer_id is None:
raise ScenarioError("cannot prove departure liveness with a missing identity")
allowed_peer_ids = {*expected_remaining_peer_ids, departed_peer_id}
deadline = time.monotonic() + PEER_DEPARTURE_TIMEOUT_SECONDS
last_events: list[dict[str, Any]] = []
last_peers: list[dict[str, Any]] = []
while time.monotonic() < deadline:
peers = peer.list_peers()
last_peers = peers
actual_peer_ids = {item.get("peer_id") for item in peers}
if not expected_remaining_peer_ids <= actual_peer_ids:
raise ScenarioError(
f"{departed_author} departure lost a peer that must remain live: "
f"expected {expected_remaining_peer_ids}, found {actual_peer_ids}; "
f"peers={peers}"
)
if not actual_peer_ids <= allowed_peer_ids:
raise ScenarioError(
f"{departed_author} departure observed unexpected authenticated peers: "
f"{actual_peer_ids}; peers={peers}"
)
events = peer.call_to_play_events()
last_events = events
if (
{event.get("id") for event in events} == expected_event_ids
and actual_peer_ids == expected_remaining_peer_ids
):
return events
time.sleep(0.2)
raise ScenarioError(
f"{departed_author} departure did not converge after the "
f"{PEER_DEPARTURE_TIMEOUT_SECONDS}s liveness allowance while peers "
f"{expected_remaining_peer_ids} stayed live: events={last_events}, "
f"peers={last_peers}"
)
def assert_exact_peer_ids(peer: Peer, expected_ids: set[str | None]) -> None:
if None in expected_ids:
raise ScenarioError("cannot assert an exact peer set with a missing identity")
peers = peer.list_peers()
actual_ids = {item.get("peer_id") for item in peers}
if actual_ids != expected_ids:
raise ScenarioError(
f"{peer.name} authenticated unexpected peers: {actual_ids} != "
f"{expected_ids}; peers={peers}"
)
def wait_call_to_play_exact_events_from_exact_peers(
peer: Peer,
expected_event_ids: set[str],
expected_peer_ids: set[str | None],
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_events: list[dict[str, Any]] = []
while time.monotonic() < deadline:
assert_exact_peer_ids(peer, expected_peer_ids)
events = peer.call_to_play_events()
last_events = events
if {event.get("id") for event in events} == expected_event_ids:
assert_exact_peer_ids(peer, expected_peer_ids)
return events
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never converged to exact Call to Play replacement "
f"{expected_event_ids} while restricted to peers {expected_peer_ids}: "
f"{last_events}"
)
def assert_game_state(
game: dict[str, Any],
*,
downloaded: bool,
installed: bool,
availability: str,
) -> None:
if (
game.get("downloaded") != downloaded
or game.get("installed") != installed
or game.get("availability") != availability
):
raise ScenarioError(
f"unexpected game state for {game.get('id')}: "
f"downloaded={game.get('downloaded')} installed={game.get('installed')} "
f"availability={game.get('availability')}"
)
def wait_peer_has_game(
observer: Peer,
peer_id: str | None,
game_id: str,
content_id: str,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
raise ScenarioError("cannot wait for a peer without peer_id")
deadline = time.monotonic() + timeout
last_peers: list[dict[str, Any]] = []
while time.monotonic() < deadline:
peers = observer.list_peers()
last_peers = peers
for peer in peers:
if peer.get("peer_id") != peer_id:
continue
if any(
game.get("game_id") == game_id
and game.get("content_id") == content_id
for game in peer.get("games", [])
):
return peer
time.sleep(0.4)
raise ScenarioError(
f"{observer.name} never saw peer {peer_id} advertise exact {game_id}/{content_id}; "
f"peers={last_peers}"
)
def wait_peer_without_game(
observer: Peer,
peer_id: str | None,
game_id: str,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
raise ScenarioError("cannot wait for a peer without peer_id")
deadline = time.monotonic() + timeout
last_peer: dict[str, Any] | None = None
while time.monotonic() < deadline:
for peer in observer.list_peers():
if peer.get("peer_id") != peer_id:
continue
last_peer = peer
if peer.get("library_revision") is None:
break
if all(game.get("game_id") != game_id for game in peer.get("games", [])):
return peer
time.sleep(0.4)
raise ScenarioError(
f"{observer.name} never observed an authoritative snapshot without {game_id} "
f"from {peer_id}; peer={last_peer}"
)
def assert_local_absent(peer: Peer, game_id: str) -> None:
rows = peer.list_games()["local"]
if any(
row["id"] == game_id and (row.get("downloaded") or row.get("installed"))
for row in rows
):
raise ScenarioError(f"{peer.name} advertises failed local {game_id}: {rows}")
def assert_no_active(peer: Peer, game_id: str) -> None:
status = peer.status()
active = status["active_operations"]
if any(item["game_id"] == game_id for item in active):
raise ScenarioError(f"{peer.name} still has active operation for {game_id}: {active}")
def assert_not_exists(path: Path) -> None:
if path.exists():
raise ScenarioError(f"expected path to be absent: {path}")
def assert_failed_stream_left_no_local(peer: Peer, game_id: str) -> None:
game_root = peer.host_games_dir / game_id
assert_local_absent(peer, game_id)
assert_not_exists(game_root / "local")
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / f"{game_id}.eti")
def assert_download_status_trace(
peer: Peer,
game_id: str,
*,
start: int,
terminal: str | None,
reason: str | None = None,
expect_begin: bool = True,
expect_verification: bool = True,
expect_invalid_source_retry: bool = False,
) -> str:
status_names = {
"download-begin",
"download-progress",
"download-activity-changed",
"download-finished",
"download-failed",
}
status_events = [
item
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") in status_names
and item.get("data", {}).get("game_id") == game_id
]
if not status_events:
raise ScenarioError(f"no attempt-keyed download status events for {game_id}")
attempt_ids = {
item.get("data", {}).get("attempt_id") for item in status_events
}
if len(attempt_ids) != 1:
raise ScenarioError(
f"download status crossed attempt IDs for {game_id}: {status_events}"
)
attempt_id = next(iter(attempt_ids))
if (
not isinstance(attempt_id, str)
or not attempt_id.isascii()
or not attempt_id.isdigit()
or (len(attempt_id) > 1 and attempt_id.startswith("0"))
):
raise ScenarioError(f"attempt ID is not a canonical decimal string: {attempt_id!r}")
begins = [item for item in status_events if item.get("event") == "download-begin"]
expected_begin_count = 1 if expect_begin else 0
if len(begins) != expected_begin_count:
raise ScenarioError(
f"unexpected begin count for {game_id}: {len(begins)} != {expected_begin_count}"
)
activities = [
item.get("data", {}).get("activity")
for item in status_events
if item.get("event") == "download-activity-changed"
]
expected_activities: list[str | None] = []
if expect_verification:
expected_activities.append("verifying-downloaded-chunks")
if expect_invalid_source_retry:
expected_activities.append("retrying-invalid-source")
expected_activities.append(None)
if activities != expected_activities:
raise ScenarioError(
f"verification activity trace for {game_id} was {activities}, "
f"expected {expected_activities}"
)
activity_indices = [
index
for index, item in enumerate(status_events)
if item.get("event") == "download-activity-changed"
]
if begins and activity_indices and status_events.index(begins[0]) >= activity_indices[0]:
raise ScenarioError(
f"verification activity preceded download begin for {game_id}: {status_events}"
)
terminals = [
item
for item in status_events
if item.get("event") in {"download-finished", "download-failed"}
]
expected_terminal_count = 0 if terminal is None else 1
if len(terminals) != expected_terminal_count:
raise ScenarioError(
f"unexpected terminal count for {game_id}: {terminals}"
)
if terminal is not None:
observed = terminals[0]
if observed.get("event") != terminal:
raise ScenarioError(
f"download terminal for {game_id} was {observed.get('event')}, expected {terminal}"
)
observed_reason = observed.get("data", {}).get("reason")
if observed_reason != reason:
raise ScenarioError(
f"download terminal reason for {game_id} was {observed_reason!r}, "
f"expected {reason!r}"
)
terminal_index = status_events.index(observed)
if begins and status_events.index(begins[0]) >= terminal_index:
raise ScenarioError(
f"download terminal preceded begin for {game_id}: {status_events}"
)
if activities and not any(
index < terminal_index
and item.get("event") == "download-activity-changed"
and item.get("data", {}).get("activity") is None
for index, item in enumerate(status_events)
):
raise ScenarioError(
f"download activity did not clear before terminal for {game_id}: {status_events}"
)
elif reason is not None:
raise ScenarioError("a reason cannot be expected without a terminal event")
observed_milestones: list[tuple[str, str | None]] = []
for item in status_events:
event = item.get("event")
data = item.get("data", {})
if event == "download-begin":
observed_milestones.append((event, None))
elif event == "download-activity-changed":
observed_milestones.append((event, data.get("activity")))
elif event in {"download-finished", "download-failed"}:
observed_milestones.append((event, data.get("reason")))
expected_milestones: list[tuple[str, str | None]] = []
if expect_begin:
expected_milestones.append(("download-begin", None))
if expect_verification:
expected_milestones.append(
("download-activity-changed", "verifying-downloaded-chunks")
)
if expect_invalid_source_retry:
expected_milestones.append(
("download-activity-changed", "retrying-invalid-source")
)
expected_milestones.append(("download-activity-changed", None))
if terminal is not None:
expected_milestones.append((terminal, reason))
if observed_milestones != expected_milestones:
raise ScenarioError(
f"download status milestones for {game_id} were {observed_milestones}, "
f"expected {expected_milestones}"
)
final_status = status_events[-1]
if terminal is not None:
final_status_is_expected = final_status.get("event") == terminal
else:
final_status_is_expected = (
final_status.get("event") == "download-activity-changed"
and final_status.get("data", {}).get("activity") is None
)
if not final_status_is_expected:
raise ScenarioError(
f"download status continued after settlement for {game_id}: {status_events}"
)
return attempt_id
def assert_stream_install_lifecycle(
peer: Peer,
game_id: str,
*,
start: int,
content_id: str,
source_peer_id: str | None,
source_addr: str | None,
expect_invalid_source_retry: bool = False,
) -> None:
if source_peer_id is None or source_addr is None:
raise ScenarioError("cannot prove streamed lifecycle without an exact source endpoint")
window = peer.output[start:]
failure_events = [
item
for item in window
if item.get("type") == "event"
and item.get("event")
in {"download-failed", "install-failed"}
and item.get("data", {}).get("game_id") == game_id
]
if failure_events:
raise ScenarioError(
f"successful streamed install emitted failure events for {game_id}: "
f"{failure_events}"
)
lifecycle: list[tuple[Any, ...]] = []
expected_chunks = {
"bin/cnctw-payload.bin": 2 * 1024 * 1024,
"data/cnctw-assets.dat": 1024 * 1024,
}
for item in window:
if item.get("type") != "event":
continue
event = item.get("event")
data = item.get("data", {})
if event == "active-operations-changed":
active_operations = data.get("active_operations", [])
matching = [
active
for active in active_operations
if active.get("game_id") == game_id
]
if matching:
if len(matching) != 1:
raise ScenarioError(
f"streamed install published duplicate active entries: {item}"
)
lifecycle.append(("active", matching[0].get("operation")))
elif active_operations == []:
lifecycle.append(("active", "empty"))
continue
if data.get("game_id") != game_id:
continue
if event == "download-chunk-finished":
path = data.get("relative_path")
if (
path not in expected_chunks
or int(data.get("offset", -1)) != 0
or int(data.get("length", -1)) != expected_chunks[path]
or data.get("content_id") != content_id
or data.get("peer_id") != source_peer_id
or data.get("peer_addr") != source_addr
):
raise ScenarioError(
f"streamed install emitted a non-authoritative verified chunk: {data}"
)
lifecycle.append(("chunk", path))
continue
if event == "download-activity-changed":
lifecycle.append(("activity", data.get("activity")))
continue
if event == "download-finished":
throughput = data.get("throughput", {})
if throughput.get("bytes") != 3 * 1024 * 1024 or throughput.get("chunks") != 2:
raise ScenarioError(
f"streamed install throughput was not exact: {throughput}"
)
if event in {
"download-begin",
"download-finished",
"install-finished",
"download-failed",
"install-failed",
}:
lifecycle.append((event,))
expected_lifecycle = [
("active", "Downloading"),
("download-begin",),
("activity", "verifying-downloaded-chunks"),
]
if expect_invalid_source_retry:
expected_lifecycle.append(("activity", "retrying-invalid-source"))
expected_lifecycle.extend([
("chunk", "bin/cnctw-payload.bin"),
("chunk", "data/cnctw-assets.dat"),
("activity", None),
("active", "Installing"),
("download-finished",),
("active", "empty"),
("install-finished",),
])
if lifecycle != expected_lifecycle:
raise ScenarioError(
f"streamed install lifecycle for {game_id} was not exact: "
f"{lifecycle} != {expected_lifecycle}"
)
assert_download_status_trace(
peer,
game_id,
start=start,
terminal="download-finished",
expect_invalid_source_retry=expect_invalid_source_retry,
)
def event_is(event: str, game_id: str | None = None) -> Callable[[dict[str, Any]], bool]:
def predicate(item: dict[str, Any]) -> bool:
if item.get("type") != "event" or item.get("event") != event:
return False
if game_id is None:
return True
return item.get("data", {}).get("game_id") == game_id
return predicate
def event_name_in(events: set[str], game_id: str | None = None) -> Callable[[dict[str, Any]], bool]:
def predicate(item: dict[str, Any]) -> bool:
if item.get("type") != "event" or item.get("event") not in events:
return False
if game_id is None:
return True
return item.get("data", {}).get("game_id") == game_id
return predicate
def assert_no_event_since(peer: Peer, start: int, event: str, game_id: str) -> None:
for item in peer.output[start:]:
if item.get("type") == "event" and item.get("event") == event:
if item.get("data", {}).get("game_id") == game_id:
raise ScenarioError(f"unexpected {event} for {game_id}: {item}")
def assert_only_chunk_sources(
peer: Peer,
game_id: str,
allowed_sources: set[str | None],
) -> None:
allowed = {source for source in allowed_sources if source is not None}
if not allowed:
raise ScenarioError("no allowed chunk sources supplied")
seen: set[str] = set()
for item in peer.output:
if item.get("type") != "event" or item.get("event") != "download-chunk-finished":
continue
data = item["data"]
if data.get("game_id") != game_id:
continue
source = data.get("peer_id")
seen.add(source)
if source not in allowed:
raise ScenarioError(f"unexpected chunk source for {game_id}: {data}")
if not seen:
raise ScenarioError(f"no chunk events recorded for {game_id}")
def assert_chunk_authority(
peer: Peer,
game_id: str,
content_id: str,
allowed_peer_ids: set[str | None],
*,
start: int = 0,
) -> None:
allowed = {peer_id for peer_id in allowed_peer_ids if peer_id is not None}
if not allowed:
raise ScenarioError("no authenticated chunk source identities supplied")
chunks = [
item["data"]
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
]
if not chunks:
raise ScenarioError(f"no verified chunk events recorded for {game_id}")
for chunk in chunks:
if chunk.get("content_id") != content_id:
raise ScenarioError(
f"chunk used wrong content identity for {game_id}: {chunk}"
)
if chunk.get("peer_id") not in allowed:
raise ScenarioError(
f"chunk used unauthenticated/unexpected source for {game_id}: {chunk}"
)
def streamed_chunk_paths(peer: Peer, game_id: str) -> list[str]:
return [
item["data"]["relative_path"]
for item in peer.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
]
def chunk_totals(peer: Peer, game_id: str, relative_path: str) -> dict[str, int]:
totals: dict[str, int] = {}
for item in peer.output:
if item.get("type") != "event" or item.get("event") != "download-chunk-finished":
continue
data = item["data"]
if data.get("game_id") != game_id or data.get("relative_path") != relative_path:
continue
totals[data["peer_id"]] = totals.get(data["peer_id"], 0) + int(data["length"])
return totals
def chunk_sources(peer: Peer, game_id: str) -> set[str]:
return {
item["data"]["peer_id"]
for item in peer.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
}
def assert_no_duplicate_chunks(peer: Peer, game_id: str) -> None:
seen: set[tuple[str | None, int]] = set()
for item in peer.output:
if item.get("type") != "event" or item.get("event") != "download-chunk-finished":
continue
data = item["data"]
if data.get("game_id") != game_id:
continue
key = (data.get("relative_path"), int(data.get("offset", 0)))
if key in seen:
raise ScenarioError(f"{peer.name} downloaded a duplicate chunk for {game_id}: {key}")
seen.add(key)
def count_events(peer: Peer, event: str, game_id: str) -> int:
return sum(
1
for item in peer.output
if item.get("type") == "event"
and item.get("event") == event
and item.get("data", {}).get("game_id") == game_id
)
def diff_game_dirs(source: Path, destination: Path) -> None:
source_manifest = manifest(source)
destination_manifest = manifest(destination)
if source_manifest != destination_manifest:
diff = run_diff(source, destination)
raise ScenarioError(
f"manifest mismatch between {source} and {destination}\n{diff}"
)
diff = run_diff(source, destination)
if diff:
raise ScenarioError(f"diff mismatch between {source} and {destination}\n{diff}")
def manifest(root: Path) -> dict[str, str]:
if not root.exists():
raise ScenarioError(f"missing manifest root: {root}")
entries: dict[str, str] = {}
for path in sorted(root.rglob("*")):
if not path.is_file():
continue
rel = path.relative_to(root)
if any(part in IGNORED_DIFF_NAMES for part in rel.parts):
continue
hasher = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
hasher.update(chunk)
entries[str(rel)] = hasher.hexdigest()
return entries
def run_diff(source: Path, destination: Path) -> str:
command = [
"diff",
"-r",
"-x",
".lanspread",
"-x",
".lanspread.json",
"-x",
"local",
str(source),
str(destination),
]
result = subprocess.run(
command,
cwd=REPO,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
return result.stdout if result.returncode else ""
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"scenarios",
nargs="*",
help="Scenario IDs to run, e.g. S18 S20. Defaults to all implemented scenarios.",
)
parser.add_argument(
"--build-image",
action="store_true",
help="Run `just peer-cli-image` before starting scenarios.",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
selected = {item.lower() for item in args.scenarios} if args.scenarios else None
try:
Runner(selected=selected, build_image=args.build_image).run()
except ScenarioError as error:
print(f"\nFAILED: {error}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
Reference in New Issue View Git Blame Copy Permalink