Files
CosmicClash/server/security/test_fleet_manifests.py
T
Josh Creek 4f48f0a6a8 fix(kind): wait on the Fleet field that exists
The gate asserted `--for=jsonpath='{.status.ready}'=2`. An Agones Fleet's
status carries replicas, readyReplicas, reservedReplicas and
allocatedReplicas -- there is no `ready` -- so the wait could never match
however healthy the Fleet was.

It failed in the most misleading way available: as "the Fleet never
became ready", which sent three separate investigations after the game
server. Two of those found genuine bugs, but the gate would have stayed
red with both fixed.

The evidence is in the previous CI run's own dump, which the new
per-container diagnostics produced: both GameServers Ready and stable for
5m6s, and the Fleet reporting DESIRED 2 / CURRENT 2 / READY 2, while
kubectl wait timed out beside it. That same dump also confirms the health
fix in 0de97381 worked -- those GameServers had been churning every ~20s
before it.

Assert the corrected jsonpath in test_fleet_manifests.py and reject the
old one, alongside the build-by-default behaviour, so neither silently
regresses.
2026-09-05 21:55:20 +01:00

130 lines
6.1 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", "readOnlyRootFilesystem: true",
"allowPrivilegeEscalation: false", "fsGroup: 10001",
"name: HOME", "value: /run/cosmic-clash",
):
self.assertIn(hardening, fleet)
# Agones must assign its SDK service account so it can keep the token
# available to its injected sidecar while masking it from the game.
self.assertNotIn("serviceAccountName:", fleet)
self.assertNotIn("automountServiceAccountToken:", fleet)
for runtime in (
"ghcr.io/cosmic-clash/game-server@sha256:",
"--sdk-base-url=http://127.0.0.1:9358",
"--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)
game_server_egress = network.split("name: game-server-allowed-egress", 1)[-1].split("---", 1)[0]
for field in ("app.kubernetes.io/name: game-server", "port: 8080", "port: 443"):
self.assertIn(field, game_server_egress)
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("gameservers.namespaces[0]=cosmic-clash", runner)
for service in (
"agones.allocator.service.serviceType=ClusterIP",
"agones.ping.http.serviceType=ClusterIP",
"agones.ping.udp.serviceType=ClusterIP",
):
self.assertIn(service, runner)
# The Fleet's readiness field is readyReplicas; waiting on `.status.ready`
# silently never matches and reads as "the Fleet never became ready".
self.assertIn("jsonpath='{.status.readyReplicas}'=2", runner)
self.assertNotIn("jsonpath='{.status.ready}'", runner)
# Build by default, or a local rerun verifies whatever was tagged last.
self.assertIn("KIND_REUSE_GAME_SERVER_IMAGE", runner)
self.assertIn("cosmic-clash.io/capacity-type=on-demand", runner)
self.assertIn("topology.kubernetes.io/zone=kind-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()