Protect private bank configuration from drift reconciliation (Refs #152)

agent/tests/test_agent.py

@@ -1644,9 +1644,51 @@ class AgentTests(unittest.TestCase):
         self.assertEqual(self.agent.execute(self.operation("instance.remove"))["state"], "failed")
         self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))
 
-    def test_create_materializes_private_bank_config_from_instance_credential(self):
+    def test_private_bank_file_survives_stale_file_plans_without_snapshots(self):
+        self.create()
+        app = self.agent.instance_dir(INSTANCE, 1) / "app"
+        value = "configs/electric_cherry_bank.cfg"
+        target = app / value
+        secret = b'set tws_bank_token "ec_bank_private_regression"\n'
+        target.write_bytes(secret)
+        files = [{"id": 9, "path": value, "delete": True, "contentBase64": None}]
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.stage(INSTANCE, 1, {"stageId": "private", "files": files}, "stage")
+        marker = self.agent.instance_dir(INSTANCE, 1) / "stages/private/manifest.json"
+        # A stage written by a previous agent must be rejected on activation too.
+        module.atomic_json(marker, {"state": "staged", "files": files})
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.activate(INSTANCE, 1, "private", override_safety=True)
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.rebuild_before(INSTANCE, 1, "private", {"files": []})
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.restore_files(app, [{"path": value, "contentBase64": None}])
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.file_operation("files.read", INSTANCE, 1, {"path": value}, "read")
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.read_log(INSTANCE, 1, {"path": value})
+        manifest = self.agent.file_operation("files.manifest", INSTANCE, 1,
+            {"roots": ["configs"], "paths": [value], "ignoreNames": []}, "manifest")
+        self.assertNotIn(value, [item["path"] for item in manifest["files"]])
+        self.assertEqual(target.read_bytes(), secret)
+        self.assertNotIn("before", json.loads(marker.read_text()))
+        with self.assertRaisesRegex(AgentError, "Private bank"):
+            self.agent.initial_files({"files": [{"path": value, "contentBase64": base64.b64encode(secret).decode()}]})
+
+    def prepare_t6_library(self):
+        target = self.agent.library_dir("t6", self.library_id)
+        module.shutil.copytree(self.library, target)
+        manifest = json.loads((target / "manifest.json").read_text())
+        manifest["game"] = "t6"
+        module.atomic_json(target / "manifest.json", manifest)
+
+    def test_t6_create_materializes_private_bank_config(self):
+        self.prepare_t6_library()
+        self.test_create_materializes_private_bank_config_from_instance_credential(game="t6")
+
+    def test_create_materializes_private_bank_config_from_instance_credential(self, game="t5"):
         secret = "ec_bank_" + "a" * 43
-        payload = {"name": "server", "game": "t5", "libraryId": self.library_id,
+        payload = {"name": "server", "game": game, "libraryId": self.library_id,
                    "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960,
                    "bank": {"token": secret, "configPath": "configs/electric_cherry_bank.cfg"}}
         operation = self.operation("instance.create", payload, "create")
@@ -1699,9 +1741,14 @@ class AgentTests(unittest.TestCase):
         self.assertFalse(path.exists())
         self.assertEqual(self.docker.calls, calls)
 
-    def test_endpoint_upgrade_rebuild_retries_original_decisions_before_starting_public_bank(self):
+    def test_t6_rebuild_preserves_private_bank_configuration(self):
+        self.prepare_t6_library()
+        self.test_endpoint_upgrade_rebuild_retries_original_decisions_before_starting_public_bank(game="t6")
+
+    def test_endpoint_upgrade_rebuild_retries_original_decisions_before_starting_public_bank(self, game="t5"):
         secret = "ec_bank_" + "a" * 43
         payload = self.create()
+        payload["game"] = game
         payload["bank"] = {"token": secret, "configPath": "configs/electric_cherry_bank.cfg"}
         paths = [self.bank_journal(game=game, endpoint=endpoint)
                  for game in ("t5", "t6")