test(device-plane): prove Device Manager rollback

infra/deploy-runner/test_device_manager_control_plane_artifacts.py

@@ -31,6 +31,29 @@ def load_runner():
 RUNNER = load_runner()
 
 
+def healthy_device_plane_inventory():
+    return {
+        "schemaVersion": "nodedc.device-plane.runtime-inventory.v1",
+        "composeProject": "nodedc-device-plane",
+        "services": [
+            {
+                "service": service,
+                "containerId": character * 64,
+                "imageId": f"sha256:{character * 64}",
+                "status": "running",
+                "running": True,
+                "health": "healthy",
+                "restartCount": 0,
+            }
+            for service, character in (
+                ("device-control-core", "a"),
+                ("device-gateway", "b"),
+                ("device-postgres", "c"),
+            )
+        ],
+    }
+
+
 class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
     def build(self, script, patch_id, artifact_dir):
         environment = os.environ.copy()
@@ -164,6 +187,109 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
             ],
         )
 
+    def test_apply_gate_checks_exact_services_core_contract_and_runtime_boundary(self):
+        entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES
+        services = ("device-control-core", "device-manager")
+        with (
+            mock.patch.object(
+                RUNNER,
+                "healthcheck_compose_service",
+            ) as service_health,
+            mock.patch.object(RUNNER, "healthcheck_url") as url_health,
+            mock.patch.object(
+                RUNNER,
+                "validate_device_manager_control_plane_runtime",
+            ) as runtime_acceptance,
+        ):
+            RUNNER.run_healthchecks("device-plane", entries, services)
+
+        self.assertEqual(
+            [call.args for call in service_health.call_args_list],
+            [
+                ("device-plane", "device-control-core"),
+                ("device-plane", "device-manager"),
+            ],
+        )
+        url_health.assert_called_once_with(
+            RUNNER.component_healthchecks(
+                "device-plane",
+                entries,
+                services,
+            )[0]
+        )
+        runtime_acceptance.assert_called_once_with()
+
+    def test_initial_install_rollback_removes_manager_and_restores_core_only(self):
+        entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES
+        missing = {
+            RUNNER.DEVICE_PLANE_MANAGER_COMPOSE_REL,
+            "services/device-manager",
+            RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_REL,
+        }
+        existing = [entry for entry in entries if entry not in missing]
+        with tempfile.TemporaryDirectory(
+            prefix="nodedc-device-manager-rollback-",
+        ) as directory:
+            backup = Path(directory) / "backup"
+            backup.mkdir()
+            (backup / "existing-files.txt").write_text(
+                "\n".join(existing) + "\n",
+                encoding="utf-8",
+            )
+            (backup / "missing-files.txt").write_text(
+                "\n".join(entry for entry in entries if entry in missing)
+                + "\n",
+                encoding="utf-8",
+            )
+            (backup / "runtime-before.json").write_text(
+                json.dumps(healthy_device_plane_inventory()),
+                encoding="utf-8",
+            )
+            with (
+                mock.patch.object(
+                    RUNNER,
+                    "stop_and_remove_compose_services",
+                ) as stop,
+                mock.patch.object(
+                    RUNNER,
+                    "restore_platform_overlay",
+                    return_value=len(entries),
+                ) as restore,
+                mock.patch.object(
+                    RUNNER,
+                    "run_component_runtime",
+                ) as restore_runtime,
+                mock.patch.object(
+                    RUNNER,
+                    "run_healthchecks",
+                ) as restore_health,
+            ):
+                result = RUNNER.rollback_device_plane_apply(
+                    Path(directory) / "live",
+                    backup,
+                    entries,
+                    "test-stamp",
+                    True,
+                    ("device-control-core", "device-manager"),
+                )
+
+        stop.assert_called_once_with("device-plane", ("device-manager",))
+        restore.assert_called_once()
+        restore_runtime.assert_called_once_with(
+            "device-plane",
+            existing,
+            ("device-control-core",),
+        )
+        restore_health.assert_called_once_with(
+            "device-plane",
+            existing,
+            ("device-control-core",),
+        )
+        self.assertEqual(
+            result,
+            f"source+runtime-restored:{len(entries)}",
+        )
+
 
 if __name__ == "__main__":
     unittest.main(verbosity=2)