feat(deploy): extend canon for managed EDP and Engine grants
This commit is contained in:
Codex
@@ -0,0 +1,147 @@
|
||||
#!/usr/bin/env python3
|
||||
import hashlib
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tarfile
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
BUILDER = SCRIPT_DIR / "build-engine-credential-sink-recovery-artifact.mjs"
|
||||
RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy"
|
||||
SOURCE = Path("/Volumes/docker/nodedc-deploy/applied/nodedc-engine-credential-sink-20260716-001.tgz")
|
||||
SOURCE_SHA256 = "0a96add05fe59db8f490927f66e07a84490474a7afb3ef7de51e1d6fd96f86a2"
|
||||
PATCH_ID = "engine-credential-sink-recovery-20990101-001"
|
||||
|
||||
|
||||
def load_runner():
|
||||
loader = importlib.machinery.SourceFileLoader(
|
||||
"nodedc_recovery_runner_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()
|
||||
|
||||
|
||||
class EngineCredentialSinkRecoveryArtifactTest(unittest.TestCase):
|
||||
def build(self, artifact_dir, patch_id=PATCH_ID, source=SOURCE):
|
||||
environment = os.environ.copy()
|
||||
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
|
||||
environment["NODEDC_ENGINE_CREDENTIAL_SINK_RECOVERY_SOURCE"] = str(source)
|
||||
result = subprocess.run(
|
||||
["node", str(BUILDER), patch_id],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=environment,
|
||||
)
|
||||
return json.loads(result.stdout)
|
||||
|
||||
def test_recovery_changes_only_manifest_id_and_passes_runner_policy(self):
|
||||
self.assertTrue(SOURCE.is_file())
|
||||
self.assertEqual(hashlib.sha256(SOURCE.read_bytes()).hexdigest(), SOURCE_SHA256)
|
||||
with tempfile.TemporaryDirectory(prefix="nodedc-engine-sink-recovery-") as directory:
|
||||
root = Path(directory)
|
||||
first_dir = root / "first"
|
||||
second_dir = root / "second"
|
||||
first_dir.mkdir()
|
||||
second_dir.mkdir()
|
||||
first = self.build(first_dir)
|
||||
second = self.build(second_dir)
|
||||
first_artifact = Path(first["artifact"])
|
||||
second_artifact = Path(second["artifact"])
|
||||
|
||||
self.assertEqual(first["sourceArtifactSha256"], SOURCE_SHA256)
|
||||
self.assertEqual(first["changed"], ["manifest.env:id"])
|
||||
self.assertEqual(first["sha256"], second["sha256"])
|
||||
self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes())
|
||||
|
||||
with tarfile.open(SOURCE, "r:gz") as old, tarfile.open(first_artifact, "r:gz") as new:
|
||||
old_files = old.extractfile("files.txt").read()
|
||||
new_files = new.extractfile("files.txt").read()
|
||||
self.assertEqual(old_files, new_files)
|
||||
self.assertEqual(
|
||||
new.extractfile("manifest.env").read().decode("utf-8"),
|
||||
f"id={PATCH_ID}\ncomponent=engine\ntype=app-overlay\n",
|
||||
)
|
||||
for relative_path, expected_sha256 in first["payloadSha256"].items():
|
||||
old_bytes = old.extractfile(f"payload/{relative_path}").read()
|
||||
new_bytes = new.extractfile(f"payload/{relative_path}").read()
|
||||
self.assertEqual(old_bytes, new_bytes)
|
||||
self.assertEqual(hashlib.sha256(new_bytes).hexdigest(), expected_sha256)
|
||||
|
||||
with tempfile.TemporaryDirectory(prefix="nodedc-recovery-runner-") as work:
|
||||
manifest, entries, _payload = RUNNER.load_artifact(
|
||||
first_artifact,
|
||||
Path(work),
|
||||
)
|
||||
self.assertEqual(manifest["id"], PATCH_ID)
|
||||
self.assertEqual(tuple(entries), RUNNER.ENGINE_CREDENTIAL_SINK_ARTIFACT_ENTRIES)
|
||||
self.assertEqual(RUNNER.component_services("engine", entries), ("nodedc-backend",))
|
||||
|
||||
def test_recovery_rejects_tampered_source_invalid_id_and_overwrite(self):
|
||||
with tempfile.TemporaryDirectory(prefix="nodedc-engine-sink-recovery-negative-") as directory:
|
||||
root = Path(directory)
|
||||
tampered = root / "tampered.tgz"
|
||||
shutil.copyfile(SOURCE, tampered)
|
||||
with tampered.open("ab") as stream:
|
||||
stream.write(b"tampered")
|
||||
|
||||
environment = os.environ.copy()
|
||||
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(root / "artifacts")
|
||||
environment["NODEDC_ENGINE_CREDENTIAL_SINK_RECOVERY_SOURCE"] = str(tampered)
|
||||
result = subprocess.run(
|
||||
["node", str(BUILDER), PATCH_ID],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=environment,
|
||||
)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("source_sha256_mismatch", result.stderr)
|
||||
|
||||
invalid = subprocess.run(
|
||||
["node", str(BUILDER), "engine-credential-sink-20260716-001"],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env={
|
||||
**os.environ,
|
||||
"NODEDC_DEPLOY_ARTIFACT_DIR": str(root / "invalid"),
|
||||
"NODEDC_ENGINE_CREDENTIAL_SINK_RECOVERY_SOURCE": str(SOURCE),
|
||||
},
|
||||
)
|
||||
self.assertNotEqual(invalid.returncode, 0)
|
||||
self.assertIn("fresh-recovery-patch-id", invalid.stderr)
|
||||
|
||||
artifact_dir = root / "duplicate"
|
||||
artifact_dir.mkdir()
|
||||
self.build(artifact_dir)
|
||||
duplicate = subprocess.run(
|
||||
["node", str(BUILDER), PATCH_ID],
|
||||
check=False,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env={
|
||||
**os.environ,
|
||||
"NODEDC_DEPLOY_ARTIFACT_DIR": str(artifact_dir),
|
||||
"NODEDC_ENGINE_CREDENTIAL_SINK_RECOVERY_SOURCE": str(SOURCE),
|
||||
},
|
||||
)
|
||||
self.assertNotEqual(duplicate.returncode, 0)
|
||||
self.assertIn("artifact_already_exists", duplicate.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user