mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
60 lines
2.4 KiB
Python
60 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the small Agones allocation response surface used by the smoke gate."""
|
|
|
|
import json
|
|
import sys
|
|
from typing import Any
|
|
|
|
|
|
def validate_allocation(document: dict[str, Any]) -> tuple[str, int]:
|
|
status = document.get("status")
|
|
if not isinstance(status, dict) or status.get("state") != "Allocated":
|
|
raise ValueError(f"allocation state is {status.get('state') if isinstance(status, dict) else None!r}, expected 'Allocated'")
|
|
|
|
game_server = status.get("gameServer")
|
|
if not isinstance(game_server, dict):
|
|
raise ValueError("allocation did not return a GameServer")
|
|
metadata = game_server.get("metadata")
|
|
name = metadata.get("name") if isinstance(metadata, dict) else None
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError("allocation GameServer has no metadata.name")
|
|
|
|
game_status = game_server.get("status")
|
|
if not isinstance(game_status, dict):
|
|
raise ValueError("allocation GameServer has no status")
|
|
address = game_status.get("address")
|
|
if not isinstance(address, str) or not address.strip() or any(char.isspace() for char in address):
|
|
raise ValueError(f"allocation returned an invalid address: {address!r}")
|
|
if address in {"0.0.0.0", "::"}:
|
|
raise ValueError(f"allocation returned an unspecified address: {address!r}")
|
|
|
|
ports = game_status.get("ports")
|
|
if not isinstance(ports, list):
|
|
raise ValueError("allocation GameServer has no ports")
|
|
game_ports = [
|
|
entry.get("port")
|
|
for entry in ports
|
|
if isinstance(entry, dict) and entry.get("name") == "game"
|
|
]
|
|
if len(game_ports) != 1 or not isinstance(game_ports[0], int) or not 1 <= game_ports[0] <= 65535:
|
|
raise ValueError(f"allocation did not return exactly one valid named game port: {ports!r}")
|
|
return name, game_ports[0]
|
|
|
|
|
|
def main() -> int:
|
|
if len(sys.argv) != 2:
|
|
print(f"usage: {sys.argv[0]} allocation.json", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
with open(sys.argv[1], encoding="utf-8") as handle:
|
|
name, port = validate_allocation(json.load(handle))
|
|
except (OSError, ValueError, json.JSONDecodeError) as error:
|
|
print(f"8.49 allocation validation failed: {error}", file=sys.stderr)
|
|
return 1
|
|
print(f"8.49 PASS: Fleet became ready; GameServer {name} returned game UDP port {port}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|