test(multiplayer): harden agones allocation gate

scripts/test_verify_agones_allocation_response.py

@@ -0,0 +1,64 @@
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))
+
+from verify_agones_allocation_response import validate_allocation
+
+
+def response(**overrides):
+    document = {
+        "status": {
+            "state": "Allocated",
+            "gameServer": {
+                "metadata": {"name": "cosmic-clash-game-abc"},
+                "status": {
+                    "address": "10.0.0.7",
+                    "ports": [{"name": "game", "port": 31001}],
+                },
+            },
+        }
+    }
+    document["status"].update(overrides)
+    return document
+
+
+class AgonesAllocationResponseTest(unittest.TestCase):
+    def test_accepts_allocated_game_server_with_named_udp_port(self):
+        self.assertEqual(validate_allocation(response()), ("cosmic-clash-game-abc", 31001))
+
+    def test_rejects_non_allocated_state(self):
+        with self.assertRaises(ValueError):
+            validate_allocation(response(state="Ready"))
+
+    def test_rejects_missing_identity_or_address(self):
+        missing_name = response()
+        missing_name["status"]["gameServer"]["metadata"] = {}
+        with self.assertRaises(ValueError):
+            validate_allocation(missing_name)
+
+        missing_address = response()
+        missing_address["status"]["gameServer"]["status"]["address"] = "0.0.0.0"
+        with self.assertRaises(ValueError):
+            validate_allocation(missing_address)
+
+    def test_rejects_ambiguous_or_invalid_game_ports(self):
+        duplicate = response()
+        duplicate["status"]["gameServer"]["status"]["ports"].append({"name": "game", "port": 31002})
+        with self.assertRaises(ValueError):
+            validate_allocation(duplicate)
+
+        wrong_name = response()
+        wrong_name["status"]["gameServer"]["status"]["ports"] = [{"name": "query", "port": 31001}]
+        with self.assertRaises(ValueError):
+            validate_allocation(wrong_name)
+
+        invalid_port = response()
+        invalid_port["status"]["gameServer"]["status"]["ports"][0]["port"] = 70000
+        with self.assertRaises(ValueError):
+            validate_allocation(invalid_port)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/verify_agones_allocation_response.py

@@ -0,0 +1,59 @@
+#!/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())

scripts/verify_kind_agones.sh

@@ -99,16 +99,4 @@ spec:
 EOF
 kubectl create -f "$work_dir/allocation.yaml" -o json > "$work_dir/allocation.json"
 
-python3 - "$work_dir/allocation.json" <<'PY'
-import json
-import sys
-
-doc = json.load(open(sys.argv[1], encoding="utf-8"))
-status = doc.get("status", {})
-if status.get("state") != "Allocated":
-    raise SystemExit(f"allocation state is {status.get('state')!r}, expected 'Allocated'")
-ports = status.get("gameServer", {}).get("status", {}).get("ports", [])
-if not ports or not any(p.get("port", 0) > 0 and p.get("port") != 7777 for p in ports):
-    raise SystemExit(f"allocation did not return a dynamic UDP port: {ports!r}")
-print("8.49 PASS: Fleet became ready and allocation returned a dynamic UDP port")
-PY
+python3 scripts/verify_agones_allocation_response.py "$work_dir/allocation.json"

scripts/verify_multiplayer_local.sh

@@ -32,5 +32,6 @@ python3 "$root_dir/server/security/test_compose_manifests.py"
 python3 "$root_dir/server/security/test_kubernetes_policies.py"
 python3 "$root_dir/server/security/test_supply_chain.py"
 python3 "$root_dir/scripts/verify_observability_manifests.py"
+python3 -m unittest "$root_dir/scripts/test_verify_agones_allocation_response.py"
 
 echo "LOCAL MULTIPLAYER GATE PASS"