Files
CosmicClash/server/security/test_fleet_manifests.py
T

106 lines
4.8 KiB
Python

from pathlib import Path
import unittest
BASE = Path(__file__).parents[2] / "deploy" / "k8s"
ROOT = Path(__file__).parents[2]
class FleetManifestTest(unittest.TestCase):
def read(self, path):
return (BASE / path).read_text()
def test_base_fleet_selects_compatible_game_servers(self):
fleet = self.read("base/fleet.yaml")
for label in (
"cosmic-clash.io/region: EU", "cosmic-clash.io/build: build-1",
'cosmic-clash.io/protocol: "1"', "cosmic-clash.io/transport: enet",
"protocol: UDP", "containerPort: 7777", "replicas: 2",
):
self.assertIn(label, fleet)
for hardening in ("runAsNonRoot: true", "automountServiceAccountToken: false", "readOnlyRootFilesystem: true", "allowPrivilegeEscalation: false"):
self.assertIn(hardening, fleet)
for runtime in (
"ghcr.io/cosmic-clash/game-server@sha256:",
"--sdk-base-url=http://127.0.0.1:9357",
"--control-plane-url=http://control-plane.cosmic-clash.svc.cluster.local:8080",
"--protocol-version=1",
"/opt/cosmic-clash/CosmicClashServer.x86_64",
"--roster-path=/run/cosmic-clash/join-roster.json",
"--allocated-mode",
"--join-authorisations-key-file=/run/secrets/cosmic-clash/join-signing-key",
"fieldPath: metadata.annotations['cosmic-clash.io/image-digest']",
"secretName: cosmic-clash-game-server",
"emptyDir: {}",
):
self.assertIn(runtime, fleet)
for scheduling in (
"cosmic-clash.io/capacity-type: on-demand",
"topologyKey: topology.kubernetes.io/zone",
"whenUnsatisfiable: DoNotSchedule",
"maxSkew: 1",
):
self.assertIn(scheduling, fleet)
def test_autoscaler_preserves_ready_floor_and_owns_fleet(self):
autoscaler = self.read("base/fleet-autoscaler.yaml")
for field in (
"kind: FleetAutoscaler", "namespace: cosmic-clash",
"fleetName: cosmic-clash-game", "type: Buffer",
"minReady: 2", "maxReady: 6", "bufferSize: 2",
):
self.assertIn(field, autoscaler)
def test_pdb_protects_the_ready_floor_and_matches_game_servers(self):
pdb = self.read("base/game-server-pdb.yaml")
for field in (
"kind: PodDisruptionBudget", "apiVersion: policy/v1",
"namespace: cosmic-clash", "minAvailable: 2",
"app.kubernetes.io/name: game-server",
):
self.assertIn(field, pdb)
def test_eu_and_na_overlays_are_distinct_and_namespaced(self):
eu = self.read("overlays/eu/region.yaml")
na = self.read("overlays/na/region.yaml")
na_kustomization = self.read("overlays/na/kustomization.yaml")
self.assertIn("cosmic-clash.io/region: EU", eu)
self.assertIn("cosmic-clash.io/region: NA", na)
self.assertIn("path: /spec/template/spec/template/spec/containers/0/args/21", na_kustomization)
self.assertIn("value: --region=NA", na_kustomization)
self.assertNotEqual(eu, na)
self.assertEqual(na_kustomization.count("value: --region=NA"), 1)
self.assertEqual(na_kustomization.count("value: --region=EU"), 0)
for document in (eu, na):
self.assertIn("namespace: cosmic-clash", document)
def test_allocator_agones_rbac_is_in_the_game_server_namespace(self):
base = self.read("base/kustomization.yaml")
rbac = self.read("base/rbac.yaml")
self.assertNotIn("namespace: cosmic-clash", base)
self.assertNotIn("namespace: agones-system", rbac)
self.assertGreaterEqual(rbac.count("namespace: cosmic-clash"), 3)
self.assertIn("name: allocator", rbac)
def test_control_plane_service_and_game_server_egress_are_declared(self):
service = self.read("base/control-plane-service.yaml")
network = self.read("base/network-policies.yaml")
base = self.read("base/kustomization.yaml")
for field in ("kind: Service", "name: control-plane", "port: 8080", "targetPort: http"):
self.assertIn(field, service)
for field in ("name: game-server-allowed-egress", "app.kubernetes.io/name: game-server", "port: 8080"):
self.assertIn(field, network)
self.assertIn("control-plane-service.yaml", base)
def test_kind_runner_is_explicitly_separate_from_production_roster_flow(self):
runner = (ROOT / "scripts/verify_kind_agones.sh").read_text()
self.assertIn("Agones lifecycle smoke", runner)
self.assertIn("--control-plane-url=", runner)
self.assertIn("--allocated-mode", runner)
validator = (ROOT / "scripts/verify_agones_allocation_response.py").read_text()
self.assertIn("game UDP port", validator)
if __name__ == "__main__":
unittest.main()