feat(deploy): add Core channel forward upgrade

This commit is contained in:
Codex
2026-08-12 01:43:09 +03:00
parent c2d06bddc7
commit d8c0e9f3b1
4 changed files with 396 additions and 36 deletions
@@ -289,6 +289,35 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
self.assertEqual(checks[0]["expected_json"]["managementApi"], "enabled")
self.assertEqual(checks[0]["expected_json"]["commandTransport"], "disabled")
def test_edge_core_channel_upgrade_is_core_only_and_pins_bootstrap_018(self):
manifest, entries, names, result = self.assert_deterministic_artifact(
"build-device-edge-core-channel-bootstrap-artifact.mjs",
"device-edge-core-channel-upgrade-unit-001",
RUNNER.DEVICE_PLANE_EDGE_CORE_CHANNEL_UPGRADE_ENTRIES,
)
self.assertEqual(manifest["component"], "device-plane")
self.assertEqual(
RUNNER.component_services("device-plane", entries),
("device-control-core",),
)
self.assertEqual(result["services"], ["device-control-core"])
self.assertIn(
"payload/deployment/device-edge-core-channel-upgrade-v1.json",
names,
)
descriptor = RUNNER.expected_device_plane_edge_core_channel_upgrade_descriptor(
"device-edge-core-channel-upgrade-unit-001"
)
self.assertEqual(descriptor["action"], "upgrade")
self.assertEqual(
descriptor["bootstrapPredecessor"]["patchId"],
"device-edge-core-channel-bootstrap-20260812-018",
)
self.assertEqual(
descriptor["endpointPolicy"],
"public-ipv4-standard-https-tcp-443-only",
)
def test_release_v2_keeps_release_v1_predecessor_contract_immutable(self):
predecessor = device_manager_release_v1_descriptor(
release_id="device-manager-release-20260811-010",
@@ -492,6 +521,24 @@ class DeviceManagerControlPlaneArtifactsTest(unittest.TestCase):
allow_invalid_unexported_recovery=True
)
with (
mock.patch.object(
RUNNER,
"ensure_platform_runtime_secret",
),
mock.patch.object(
RUNNER,
"ensure_device_edge_channel_core_identity",
) as ensure_edge_identity,
):
RUNNER.prepare_component_runtime(
"device-plane",
RUNNER.DEVICE_PLANE_EDGE_CORE_CHANNEL_UPGRADE_ENTRIES,
)
ensure_edge_identity.assert_called_once_with(
allow_invalid_unexported_recovery=False
)
def test_apply_gate_checks_exact_services_core_contract_and_runtime_boundary(self):
entries = RUNNER.DEVICE_PLANE_MANAGER_RELEASE_V1_SUCCESSOR_ENTRIES
services = ("device-control-core", "device-manager")