Require revocable authorization for automatic lobby removal

agent/tests/test_agent.py

@@ -959,6 +959,61 @@ class AgentTests(unittest.TestCase):
         self.agent.rcon = lambda *_: {"ok": True, "players": 0}
         self.assertEqual(self.agent.execute(operation)["state"], "completed")
 
+    def autoscaling_operation(self, op_id="autoscaling-remove"):
+        return self.operation("instance.remove", {"autoscaling": {"policyId": "policy", "revision": 1, "expiresAt": 2000000}}, op_id)
+
+    def test_autoscaling_occupied_or_unknown_removal_never_retries_later(self):
+        self.create()
+        for index, players in enumerate((2, None)):
+            operation = self.autoscaling_operation(f"autoscaling-{index}")
+            self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count}
+            with patch.object(self.agent, "request", side_effect=AssertionError("Occupied removal requested authorization")):
+                result = self.agent.execute(operation)
+                self.assertEqual(result["state"], "failed")
+                self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"])
+                self.agent.rcon = lambda *_: {"ok": True, "players": 0}
+                self.assertEqual(self.agent.execute(operation)["state"], "failed")
+        self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls))
+
+    def test_autoscaling_removal_requires_fresh_one_shot_controller_authorization(self):
+        self.create()
+        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
+        with patch.object(self.agent, "request", return_value={"authorized": False}) as request:
+            self.assertEqual(self.agent.execute(self.autoscaling_operation("revoked"))["state"], "failed")
+            self.assertTrue(request.call_args.args[1].endswith("/revoked/authorize-removal"))
+        with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}):
+            self.assertEqual(self.agent.execute(self.autoscaling_operation("valid"))["state"], "completed")
+        self.assertEqual(sum(args[:2] == ("container", "rm") for args in self.docker.calls), 1)
+
+    def test_autoscaling_rechecks_players_after_authorization_and_rejects_expired_claim(self):
+        self.create()
+        with patch.object(self.agent, "rcon", side_effect=[{"ok": True, "players": 0}, {"ok": True, "players": 1}]), patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}):
+            self.assertEqual(self.agent.execute(self.autoscaling_operation("joined"))["state"], "failed")
+        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
+        with patch.object(self.agent, "request", return_value={"authorized": True, "validForMs": 5000}), patch.object(module.time, "monotonic", side_effect=[0, 6]):
+            self.assertEqual(self.agent.execute(self.autoscaling_operation("expired"))["state"], "failed")
+        self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls))
+
+    def test_autoscaling_restart_before_mutation_invalidates_instead_of_replaying(self):
+        self.create()
+        operation = self.autoscaling_operation()
+        self.agent.journal.accept(operation)
+        self.agent.journal.update(operation["id"], "running")
+        self.agent.journal.close()
+        self.agent = Agent(self.config, self.docker)
+        with patch.object(self.agent, "request", side_effect=AssertionError("Recovered command requested execution")):
+            result = self.agent.execute(operation)
+        self.assertEqual(result["state"], "failed")
+        self.assertTrue(json.loads(result["result"])["autoscalingInvalidated"])
+        self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls))
+
+    def test_autoscaling_cannot_use_admin_override(self):
+        self.create()
+        operation = self.autoscaling_operation()
+        operation["payload"]["safetyOverride"] = True
+        self.assertEqual(self.agent.execute(operation)["state"], "failed")
+        self.assertFalse(any(args[:2] == ("container", "stop") for args in self.docker.calls))
+
     def test_occupancy_hold_explains_local_rcon_failure_without_exposing_reply(self):
         self.create()
         operation = self.operation("instance.restart")