681 lines
27 KiB
Python
681 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
PLATFORM_ROOT = SCRIPT_DIR.parent.parent
|
|
RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy"
|
|
|
|
|
|
def load_runner():
|
|
loader = importlib.machinery.SourceFileLoader(
|
|
"nodedc_device_manager_deploy_under_test",
|
|
str(RUNNER_PATH),
|
|
)
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
RUNNER = load_runner()
|
|
LAUNCHER_HUB_SERVICE_TRUST_UI_ENTRIES = (
|
|
"src/app/LauncherApp.tsx",
|
|
"src/styles/globals.css",
|
|
"src/widgets/admin-overlay/AdminOverlay.tsx",
|
|
)
|
|
|
|
|
|
def healthy_device_plane_inventory(*, include_manager=False):
|
|
services = [
|
|
("device-control-core", "a"),
|
|
("device-gateway", "b"),
|
|
("device-postgres", "c"),
|
|
]
|
|
if include_manager:
|
|
services.append(("device-manager", "d"))
|
|
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 services
|
|
],
|
|
}
|
|
|
|
|
|
def device_manager_release_descriptor(
|
|
release_id="device-manager-release-unit-001",
|
|
*,
|
|
action="activate",
|
|
predecessor_kind="reconciliation",
|
|
predecessor_patch="device-manager-reconciliation-unit-001",
|
|
predecessor_sha="a" * 64,
|
|
):
|
|
return {
|
|
"schemaVersion": "nodedc.device-plane.device-manager-release.v1",
|
|
"releaseId": release_id,
|
|
"action": action,
|
|
"predecessor": {
|
|
"kind": predecessor_kind,
|
|
"patchId": predecessor_patch,
|
|
"artifactSha256": predecessor_sha,
|
|
},
|
|
**RUNNER.expected_device_plane_manager_release_boundaries(),
|
|
}
|
|
|
|
|
|
class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
|
|
def build(self, script, patch_id, artifact_dir):
|
|
environment = os.environ.copy()
|
|
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
|
|
completed = subprocess.run(
|
|
["node", str(SCRIPT_DIR / script), patch_id],
|
|
cwd=PLATFORM_ROOT,
|
|
env=environment,
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return json.loads(completed.stdout)
|
|
|
|
def assert_deterministic_artifact(self, script, patch_id, expected_entries):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-device-manager-artifact-") as directory:
|
|
root = Path(directory)
|
|
first = self.build(script, patch_id, root / "first")
|
|
second = self.build(script, patch_id, root / "second")
|
|
first_artifact = Path(first["artifact"])
|
|
second_artifact = Path(second["artifact"])
|
|
self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes())
|
|
self.assertEqual(
|
|
first["sha256"],
|
|
hashlib.sha256(first_artifact.read_bytes()).hexdigest(),
|
|
)
|
|
self.assertEqual(tuple(first["entries"]), tuple(expected_entries))
|
|
extracted = root / "extracted"
|
|
extracted.mkdir()
|
|
manifest, entries, payload = RUNNER.load_artifact(first_artifact, extracted)
|
|
self.assertEqual(tuple(entries), tuple(expected_entries))
|
|
with tarfile.open(first_artifact, "r:gz") as archive:
|
|
members = archive.getmembers()
|
|
names = [member.name for member in members]
|
|
bytes_joined = b"\n".join(
|
|
archive.extractfile(member).read()
|
|
for member in members
|
|
if member.isfile()
|
|
)
|
|
self.assertFalse(any(Path(name).name.startswith("._") for name in names))
|
|
self.assertFalse(any("/node_modules/" in name or "/.git/" in name for name in names))
|
|
self.assertNotIn(b"-----BEGIN PRIVATE KEY-----", bytes_joined)
|
|
return manifest, entries, names, first
|
|
|
|
def test_platform_hub_trust_artifact_is_exact_and_build_free(self):
|
|
manifest, entries, _names, result = self.assert_deterministic_artifact(
|
|
"build-platform-device-core-hub-trust-artifact.mjs",
|
|
"platform-device-core-hub-trust-unit-001",
|
|
RUNNER.PLATFORM_DEVICE_CORE_HUB_TRUST_ENTRIES,
|
|
)
|
|
self.assertEqual(manifest["component"], "platform")
|
|
self.assertEqual(RUNNER.component_services("platform", entries), ("launcher",))
|
|
self.assertEqual(RUNNER.component_builds("platform", entries), ())
|
|
self.assertEqual(result["services"], ["launcher"])
|
|
|
|
def test_launcher_session_artifact_is_exact(self):
|
|
manifest, entries, _names, result = self.assert_deterministic_artifact(
|
|
"build-launcher-device-core-artifact.mjs",
|
|
"launcher-device-core-session-unit-001",
|
|
RUNNER.LAUNCHER_DEVICE_CORE_SESSION_ENTRIES,
|
|
)
|
|
self.assertEqual(manifest["component"], "launcher")
|
|
self.assertEqual(RUNNER.component_services("launcher", entries), ("launcher",))
|
|
self.assertEqual(len(RUNNER.component_builds("launcher", entries)), 1)
|
|
self.assertEqual(result["services"], ["launcher"])
|
|
|
|
def test_launcher_hub_service_trust_ui_artifact_is_exact(self):
|
|
manifest, entries, _names, result = self.assert_deterministic_artifact(
|
|
"build-launcher-hub-service-trust-ui-artifact.mjs",
|
|
"launcher-hub-service-trust-ui-unit-001",
|
|
LAUNCHER_HUB_SERVICE_TRUST_UI_ENTRIES,
|
|
)
|
|
self.assertEqual(manifest["component"], "launcher")
|
|
self.assertEqual(RUNNER.component_services("launcher", entries), ("launcher",))
|
|
self.assertEqual(len(RUNNER.component_builds("launcher", entries)), 1)
|
|
self.assertEqual(result["services"], ["launcher"])
|
|
|
|
def test_device_manager_artifact_selects_only_core_and_manager(self):
|
|
manifest, entries, names, result = self.assert_deterministic_artifact(
|
|
"build-device-manager-control-plane-artifact.mjs",
|
|
"device-manager-control-plane-unit-001",
|
|
RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES,
|
|
)
|
|
self.assertEqual(manifest["component"], "device-plane")
|
|
self.assertEqual(
|
|
RUNNER.component_services("device-plane", entries),
|
|
("device-control-core", "device-manager"),
|
|
)
|
|
builds = RUNNER.component_builds("device-plane", entries)
|
|
self.assertEqual(len(builds), 2)
|
|
self.assertIn(RUNNER.DEVICE_PLANE_CONTROL_CORE_IMAGE, builds[0][1])
|
|
self.assertIn(RUNNER.DEVICE_PLANE_MANAGER_IMAGE, builds[1][1])
|
|
self.assertIn("payload/services/device-manager/dist/index.html", names)
|
|
self.assertIn(
|
|
"payload/services/device-manager/server/device-manager-server.mjs",
|
|
names,
|
|
)
|
|
self.assertIn(
|
|
"payload/services/device-control-core/src/credential-reference.mjs",
|
|
names,
|
|
)
|
|
self.assertFalse(any(name.endswith(".test.mjs") for name in names))
|
|
self.assertEqual(result["services"], ["device-control-core", "device-manager"])
|
|
self.assertNotIn("device-postgres", result["services"])
|
|
self.assertIn("docker-compose.device-manager.yml", entries)
|
|
self.assertNotIn("docker-compose.device-plane.yml", entries)
|
|
compose = (
|
|
PLATFORM_ROOT / "device-plane/docker-compose.device-manager.yml"
|
|
).read_text(encoding="utf-8")
|
|
for required in (
|
|
'DEVICE_EDGE_CHANNEL_ENABLED: "true"',
|
|
"DEVICE_EDGE_CHANNEL_CORE_KEY_FILE: ",
|
|
"DEVICE_EDGE_CHANNEL_CORE_CERTIFICATE_FILE: ",
|
|
"DEVICE_EDGE_CHANNEL_TRUST_ROOT: ",
|
|
"name: nodedc-device-plane-egress",
|
|
):
|
|
self.assertIn(required, compose)
|
|
self.assertNotIn("PRIVATE KEY", compose)
|
|
checks = RUNNER.component_healthchecks("device-plane", entries, tuple(result["services"]))
|
|
self.assertEqual(checks[0]["expected_json"]["managementApi"], "enabled")
|
|
self.assertEqual(checks[0]["expected_json"]["discoveryIngest"], "enabled")
|
|
|
|
def test_public_route_artifact_is_last_and_proxy_only(self):
|
|
manifest, entries, _names, result = self.assert_deterministic_artifact(
|
|
"build-platform-device-manager-route-artifact.mjs",
|
|
"platform-device-manager-route-unit-001",
|
|
RUNNER.PLATFORM_DEVICE_MANAGER_PUBLIC_ROUTE_ENTRIES,
|
|
)
|
|
self.assertEqual(manifest["component"], "platform")
|
|
self.assertEqual(RUNNER.component_services("platform", entries), ("reverse-proxy",))
|
|
self.assertEqual(RUNNER.component_builds("platform", entries), ())
|
|
self.assertEqual(result["services"], ["reverse-proxy"])
|
|
|
|
def test_runner_creates_only_file_backed_runtime_secrets(self):
|
|
with mock.patch.object(RUNNER, "ensure_platform_runtime_secret") as ensure:
|
|
RUNNER.prepare_component_runtime(
|
|
"platform",
|
|
RUNNER.PLATFORM_DEVICE_CORE_HUB_TRUST_ENTRIES,
|
|
)
|
|
self.assertEqual(
|
|
[call.args[0] for call in ensure.call_args_list],
|
|
[RUNNER.PLATFORM_DEVICE_CORE_INTERNAL_TOKEN_FILE],
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"ensure_platform_runtime_secret",
|
|
) as ensure,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"ensure_device_edge_channel_core_identity",
|
|
) as ensure_edge_identity,
|
|
):
|
|
RUNNER.prepare_component_runtime(
|
|
"device-plane",
|
|
RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES,
|
|
)
|
|
self.assertEqual(
|
|
[call.args[0] for call in ensure.call_args_list],
|
|
[
|
|
RUNNER.DEVICE_PLANE_POSTGRES_PASSWORD_FILE,
|
|
RUNNER.DEVICE_PLANE_GATEWAY_CORE_TOKEN_FILE,
|
|
RUNNER.DEVICE_PLANE_IDENTIFIER_PEPPER_FILE,
|
|
RUNNER.DEVICE_PLANE_MANAGEMENT_CORE_TOKEN_FILE,
|
|
RUNNER.PLATFORM_DEVICE_CORE_INTERNAL_TOKEN_FILE,
|
|
],
|
|
)
|
|
ensure_edge_identity.assert_called_once_with()
|
|
|
|
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_with_grace",
|
|
) 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_activation_resolves_predecessor_from_descriptor_and_journal(self):
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="nodedc-device-manager-release-predecessor-",
|
|
) as directory:
|
|
root = Path(directory)
|
|
applied = root / "applied"
|
|
backups = root / "backups"
|
|
temporary = root / "tmp"
|
|
device_plane = root / "device-plane"
|
|
for path in (applied, backups, temporary, device_plane / "deployment"):
|
|
path.mkdir(parents=True)
|
|
predecessor_patch = "device-manager-reconciliation-unit-004"
|
|
artifact_name = f"nodedc-device-plane-{predecessor_patch}.tgz"
|
|
artifact = applied / artifact_name
|
|
artifact_bytes = b"reviewed-reconciliation-artifact"
|
|
artifact.write_bytes(artifact_bytes)
|
|
artifact_sha = hashlib.sha256(artifact_bytes).hexdigest()
|
|
descriptor = device_manager_release_descriptor(
|
|
predecessor_patch=predecessor_patch,
|
|
predecessor_sha=artifact_sha,
|
|
)
|
|
state_file = root / "applied.jsonl"
|
|
state_file.write_text(
|
|
json.dumps({
|
|
"id": predecessor_patch,
|
|
"artifact": artifact_name,
|
|
"backup_id": "predecessor-backup",
|
|
"component": "device-plane",
|
|
"sha256": artifact_sha,
|
|
"status": "ok",
|
|
}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
(backups / "predecessor-backup").mkdir()
|
|
baseline_backup = root / "failed-backup"
|
|
predecessor_descriptor = {"marker": "exact"}
|
|
marker = (
|
|
device_plane
|
|
/ RUNNER.DEVICE_PLANE_MANAGER_V2_RECONCILIATION_REL
|
|
)
|
|
marker.write_text(
|
|
json.dumps(predecessor_descriptor),
|
|
encoding="utf-8",
|
|
)
|
|
with (
|
|
mock.patch.object(RUNNER, "APPLIED_DIR", applied),
|
|
mock.patch.object(RUNNER, "BACKUPS_DIR", backups),
|
|
mock.patch.object(RUNNER, "TMP_DIR", temporary),
|
|
mock.patch.object(RUNNER, "STATE_FILE", state_file),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"component_root",
|
|
return_value=device_plane,
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_device_plane_manager_control_plane_payload",
|
|
return_value=descriptor,
|
|
) as payload,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"load_artifact",
|
|
return_value=(
|
|
{
|
|
"id": predecessor_patch,
|
|
"component": "device-plane",
|
|
"type": "app-overlay",
|
|
},
|
|
RUNNER.DEVICE_PLANE_MANAGER_V2_RECONCILIATION_ENTRIES,
|
|
root / "predecessor-payload",
|
|
),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_device_plane_manager_v2_reconciliation_payload",
|
|
return_value=predecessor_descriptor,
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_device_plane_manager_v2_reconciliation_backup",
|
|
return_value=baseline_backup,
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_device_plane_manager_v2_reconciled_baseline",
|
|
return_value={"accepted": True},
|
|
) as baseline,
|
|
):
|
|
result = (
|
|
RUNNER.validate_device_plane_manager_activation_predecessor(
|
|
root / "payload"
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
result["mode"],
|
|
"reconciled-manager-forward-activation",
|
|
)
|
|
payload.assert_called_once_with(root / "payload")
|
|
baseline.assert_called_once_with(
|
|
baseline_backup,
|
|
marker_installed=True,
|
|
)
|
|
|
|
def test_failed_patch_id_and_sha_are_globally_non_replayable(self):
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="nodedc-device-manager-failed-replay-",
|
|
) as directory:
|
|
failed_state = Path(directory) / "failed.jsonl"
|
|
failed_state.write_text(
|
|
json.dumps({
|
|
"id": "device-manager-release-failed-001",
|
|
"sha256": "b" * 64,
|
|
"status": "failed",
|
|
}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
with mock.patch.object(
|
|
RUNNER,
|
|
"FAILED_STATE_FILE",
|
|
failed_state,
|
|
):
|
|
with self.assertRaisesRegex(
|
|
RUNNER.DeployError,
|
|
"SHA is terminal failed",
|
|
):
|
|
RUNNER.reject_failed_artifact_replay(
|
|
{"id": "different"},
|
|
"b" * 64,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
RUNNER.DeployError,
|
|
"patch id is terminal failed",
|
|
):
|
|
RUNNER.reject_failed_artifact_replay(
|
|
{"id": "device-manager-release-failed-001"},
|
|
"c" * 64,
|
|
)
|
|
|
|
def test_upgrade_accepts_any_current_release_without_runner_patch_ids(self):
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="nodedc-device-manager-release-upgrade-",
|
|
) as directory:
|
|
root = Path(directory)
|
|
applied = root / "applied"
|
|
backups = root / "backups"
|
|
temporary = root / "tmp"
|
|
device_plane = root / "device-plane"
|
|
for path in (applied, backups, temporary, device_plane / "deployment"):
|
|
path.mkdir(parents=True)
|
|
predecessor_patch = "device-manager-release-arbitrary-041"
|
|
artifact_name = f"nodedc-device-plane-{predecessor_patch}.tgz"
|
|
artifact = applied / artifact_name
|
|
artifact_bytes = b"arbitrary-reviewed-device-manager-release"
|
|
artifact.write_bytes(artifact_bytes)
|
|
artifact_sha = hashlib.sha256(artifact_bytes).hexdigest()
|
|
predecessor_descriptor = device_manager_release_descriptor(
|
|
release_id=predecessor_patch,
|
|
)
|
|
candidate_descriptor = device_manager_release_descriptor(
|
|
release_id="device-manager-release-arbitrary-042",
|
|
action="upgrade",
|
|
predecessor_kind="release",
|
|
predecessor_patch=predecessor_patch,
|
|
predecessor_sha=artifact_sha,
|
|
)
|
|
installed = (
|
|
device_plane / RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_REL
|
|
)
|
|
installed.write_text(
|
|
json.dumps(predecessor_descriptor),
|
|
encoding="utf-8",
|
|
)
|
|
state_file = root / "applied.jsonl"
|
|
state_file.write_text(
|
|
json.dumps({
|
|
"id": predecessor_patch,
|
|
"artifact": artifact_name,
|
|
"backup_id": "arbitrary-release-backup",
|
|
"component": "device-plane",
|
|
"sha256": artifact_sha,
|
|
"status": "ok",
|
|
}) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
(backups / "arbitrary-release-backup").mkdir()
|
|
with (
|
|
mock.patch.object(RUNNER, "APPLIED_DIR", applied),
|
|
mock.patch.object(RUNNER, "BACKUPS_DIR", backups),
|
|
mock.patch.object(RUNNER, "TMP_DIR", temporary),
|
|
mock.patch.object(RUNNER, "STATE_FILE", state_file),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"component_root",
|
|
return_value=device_plane,
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"validate_device_plane_manager_control_plane_payload",
|
|
side_effect=(
|
|
candidate_descriptor,
|
|
predecessor_descriptor,
|
|
),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"load_artifact",
|
|
return_value=(
|
|
{
|
|
"id": predecessor_patch,
|
|
"component": "device-plane",
|
|
"type": "app-overlay",
|
|
},
|
|
RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES,
|
|
root / "predecessor-payload",
|
|
),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"healthcheck_compose_service",
|
|
) as health,
|
|
):
|
|
result = (
|
|
RUNNER.validate_device_plane_manager_activation_predecessor(
|
|
root / "candidate-payload"
|
|
)
|
|
)
|
|
|
|
self.assertEqual(result["mode"], "active-manager-forward-upgrade")
|
|
self.assertEqual(
|
|
[call.args for call in health.call_args_list],
|
|
[
|
|
("device-plane", "device-control-core"),
|
|
("device-plane", "device-manager"),
|
|
("device-plane", "device-postgres"),
|
|
],
|
|
)
|
|
|
|
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,
|
|
"healthcheck_compose_service_with_grace",
|
|
) as restore_health,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"component_healthchecks",
|
|
return_value=("core-health",),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"healthcheck_url",
|
|
) as restore_url_health,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"run_healthchecks",
|
|
) as generic_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",
|
|
"device-control-core",
|
|
)
|
|
restore_url_health.assert_called_once_with("core-health")
|
|
generic_health.assert_not_called()
|
|
self.assertEqual(
|
|
result,
|
|
f"source+runtime-restored:{len(entries)}",
|
|
)
|
|
|
|
def test_upgrade_rollback_restores_previous_core_and_manager_release(self):
|
|
entries = RUNNER.DEVICE_PLANE_MANAGER_CONTROL_PLANE_ENTRIES
|
|
with tempfile.TemporaryDirectory(
|
|
prefix="nodedc-device-manager-upgrade-rollback-",
|
|
) as directory:
|
|
backup = Path(directory) / "backup"
|
|
backup.mkdir()
|
|
(backup / "existing-files.txt").write_text(
|
|
"\n".join(entries) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
(backup / "missing-files.txt").write_text("", encoding="utf-8")
|
|
(backup / "runtime-before.json").write_text(
|
|
json.dumps(healthy_device_plane_inventory(include_manager=True)),
|
|
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),
|
|
),
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"run_component_runtime",
|
|
) as restore_runtime,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"healthcheck_compose_service_with_grace",
|
|
) as restore_health,
|
|
mock.patch.object(
|
|
RUNNER,
|
|
"component_healthchecks",
|
|
return_value=(),
|
|
),
|
|
):
|
|
result = RUNNER.rollback_device_plane_apply(
|
|
Path(directory) / "live",
|
|
backup,
|
|
entries,
|
|
"test-stamp",
|
|
True,
|
|
("device-control-core", "device-manager"),
|
|
)
|
|
|
|
stop.assert_not_called()
|
|
restore_runtime.assert_called_once_with(
|
|
"device-plane",
|
|
list(entries),
|
|
("device-control-core", "device-manager"),
|
|
)
|
|
self.assertEqual(
|
|
[call.args for call in restore_health.call_args_list],
|
|
[
|
|
("device-plane", "device-control-core"),
|
|
("device-plane", "device-manager"),
|
|
],
|
|
)
|
|
self.assertEqual(
|
|
result,
|
|
f"source+runtime-restored:{len(entries)}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|