mirror of
https://github.com/jcreek/CosmicClash.git
synced 2026-09-10 16:04:04 +00:00
Josh Creek
2c648514ba
test_contracts.py required operation ID `recordPlayerConnected`, but
openapi.json names that endpoint `claimPlayerConnection` — the accurate
name, since POST /servers/{id}/connect claims a connection lease and
returns a generation. Align the test on the document and assert the set
difference, so a future mismatch names the missing operation instead of
reporting "False is not true".
test_observability_manifests.py copied only two of the four files the
checker reads, so it died on a missing kustomization.yaml before ever
reaching the mutated namespace. Copy the full fixture, split the
namespace and scrape-path mutations into separate cases so either
defect produces its own diagnostic, and add an unmutated-copy case so a
broken fixture can't make the mutation cases pass vacuously.
Neither suite was invoked by any Make target or workflow, which is why
both could sit red. Add them, plus test_threat_model.py, to
verify_multiplayer_local.sh.
98 lines
4.7 KiB
Python
98 lines
4.7 KiB
Python
"""Dependency-free structural checks for the versioned control-plane contract."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).parent
|
|
|
|
|
|
class ContractTest(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.openapi = json.loads((ROOT / "openapi.json").read_text())
|
|
cls.events = json.loads((ROOT / "websocket-events.json").read_text())
|
|
cls.transitions = json.loads((ROOT / "state-transitions.json").read_text())
|
|
|
|
def test_openapi_is_versioned_and_has_core_surfaces(self):
|
|
self.assertEqual(self.openapi["openapi"], "3.1.0")
|
|
operations = {
|
|
operation["operationId"]
|
|
for path in self.openapi["paths"].values()
|
|
for operation in path.values()
|
|
if isinstance(operation, dict) and "operationId" in operation
|
|
}
|
|
# These are the operation IDs generated clients bind to, so a rename
|
|
# here is a breaking change for every consumer. Assert the difference
|
|
# rather than a bare subset check: a plain assertTrue reports only
|
|
# "False is not true" and hides which operation went missing.
|
|
required = {
|
|
"createSteamSession", "getProfile", "createQueueTicket",
|
|
"heartbeatQueueTicket", "cancelQueueTicket", "acceptProposal",
|
|
"declineProposal", "getAssignment", "registerServer",
|
|
"claimPlayerConnection", "submitMatchResult", "getRankedProfile",
|
|
}
|
|
self.assertEqual(set(), required - operations)
|
|
|
|
def test_ranked_profile_contract_is_authoritative_and_optional_season_metadata(self):
|
|
schema = self.openapi["components"]["schemas"]["RankedProfile"]
|
|
self.assertEqual(schema["required"], ["rating", "rd", "volatility", "ranked_games", "tier", "provisional"])
|
|
self.assertFalse(schema["additionalProperties"])
|
|
self.assertEqual(schema["properties"]["season_ends_at"]["format"], "date-time")
|
|
self.assertNotIn("access_token", json.dumps(schema).lower())
|
|
|
|
def test_mutations_require_idempotency_and_revision(self):
|
|
parameters = self.openapi["components"]["parameters"]
|
|
self.assertEqual(parameters["IdempotencyKey"]["name"], "Idempotency-Key")
|
|
self.assertTrue(parameters["IdempotencyKey"]["required"])
|
|
self.assertEqual(parameters["ExpectedRevision"]["name"], "If-Match-Revision")
|
|
for path, methods in self.openapi["paths"].items():
|
|
for method, operation in methods.items():
|
|
if method not in {"post", "delete", "put", "patch"} or "operationId" not in operation:
|
|
continue
|
|
if operation["operationId"] == "createSteamSession":
|
|
continue
|
|
refs = {item.get("$ref") for item in operation.get("parameters", [])}
|
|
self.assertIn("#/components/parameters/IdempotencyKey", refs, path)
|
|
|
|
def test_state_vocabulary_is_shared(self):
|
|
queue_states = self.openapi["components"]["schemas"]["QueueState"]["enum"]
|
|
websocket_states = self.events["$defs"]["stateChanged"]["allOf"][1]["properties"]["state"]["enum"]
|
|
self.assertEqual(queue_states, websocket_states)
|
|
self.assertIn("ASSIGNMENT_READY", queue_states)
|
|
self.assertIn("RESULT_PENDING", queue_states)
|
|
|
|
def test_events_have_revisioned_envelopes_and_no_credentials(self):
|
|
envelope = self.events["$defs"]["envelope"]
|
|
self.assertEqual(envelope["required"], ["event", "revision", "resource_id", "occurred_at"])
|
|
serialized = json.dumps(self.events).lower()
|
|
self.assertNotIn("access_token", serialized)
|
|
self.assertNotIn("web_api_ticket", serialized)
|
|
self.assertNotIn("relay_ticket", serialized)
|
|
|
|
def test_state_machine_has_explicit_recovery_and_terminal_edges(self):
|
|
for resource, states in self.transitions["resource_states"].items():
|
|
graph = self.transitions["transitions"][resource]
|
|
self.assertEqual(set(states), set(graph))
|
|
for state, targets in graph.items():
|
|
self.assertTrue(set(targets) <= set(states))
|
|
if state in {"COMPLETED", "CANCELLED", "EXPIRED", "FAILED"}:
|
|
self.assertEqual(targets, [], state)
|
|
|
|
queue = self.transitions["transitions"]["queue_ticket"]
|
|
self.assertIn("QUEUED", queue["PROPOSED"])
|
|
self.assertIn("QUEUED", queue["ACCEPTED"])
|
|
self.assertEqual(
|
|
self.transitions["mutation_rules"]["same_key_same_payload"],
|
|
"return_original_result_without_new_revision",
|
|
)
|
|
self.assertEqual(
|
|
self.transitions["mutation_rules"]["same_key_different_payload"],
|
|
"reject_conflict_without_state_change",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|