148 lines
6.0 KiB
Python
148 lines
6.0 KiB
Python
#!/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)
|