CosmicClash/scripts/fake_agones_provider.py

#!/usr/bin/env python3
"""Minimal deterministic Agones HTTP surface for the allocated Compose smoke."""
import json
from http.server import BaseHTTPRequestHandler, HTTPServer


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if "/gameservers" not in self.path:
            self.send_error(404)
            return
        body = {"items": [{"metadata": {"name": "allocator-ready-1", "labels": {
            "cosmic-clash.io/region": "EU", "cosmic-clash.io/build": "build-1",
            "cosmic-clash.io/protocol": "1", "cosmic-clash.io/transport": "enet"
        }}, "status": {"state": "Ready"}}]}
        self._json(body)

    def do_POST(self):
        if "/gameserverallocations" not in self.path:
            self.send_error(404)
            return
        length = int(self.headers.get("Content-Length", "0"))
        request = json.loads(self.rfile.read(length))
        selectors = request.get("spec", {}).get("selectors", [])
        labels = selectors[0].get("matchLabels", {}) if selectors else {}
        if labels.get("cosmic-clash.io/region") != "EU" or labels.get("cosmic-clash.io/transport") != "enet":
            self.send_error(422, "incompatible selector")
            return
        self._json({"status": {"state": "Allocated", "gameServerName": "allocator-ready-1",
                                "address": "127.0.0.1", "ports": [{"name": "default", "port": 31001}]}})

    def _json(self, body):
        encoded = json.dumps(body).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        self.wfile.write(encoded)

    def log_message(self, *_args):
        return


HTTPServer(("0.0.0.0", 8080), Handler).serve_forever()