140 lines
5.3 KiB
Python
140 lines
5.3 KiB
Python
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy"
|
|
BUILDER_PATH = SCRIPT_DIR / "build-engine-l2-closed-loop-artifact.mjs"
|
|
ENGINE_ROOT = SCRIPT_DIR.parent.parent.parent / "NODEDC_ENGINE_INFRA"
|
|
PATCH_ID = "engine-l2-closed-loop-20991231-999"
|
|
SOURCE_COMMIT = "dda405cff27977622af0c218abb4d4420c408536"
|
|
|
|
|
|
def load_runner():
|
|
loader = importlib.machinery.SourceFileLoader(
|
|
"nodedc_engine_l2_closed_loop",
|
|
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 EngineL2ClosedLoopArtifactTest(unittest.TestCase):
|
|
def require_target_commit(self):
|
|
current = subprocess.run(
|
|
["git", "-C", str(ENGINE_ROOT), "rev-parse", "HEAD"],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
).stdout.strip()
|
|
if current != SOURCE_COMMIT:
|
|
self.skipTest("historical Engine L2 closed-loop source has advanced")
|
|
|
|
def build(self, artifact_dir):
|
|
self.require_target_commit()
|
|
environment = os.environ.copy()
|
|
environment["NODEDC_ENGINE_SOURCE_ROOT"] = str(ENGINE_ROOT)
|
|
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
|
|
completed = subprocess.run(
|
|
["node", str(BUILDER_PATH), PATCH_ID],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
env=environment,
|
|
)
|
|
return json.loads(completed.stdout)
|
|
|
|
def test_builder_is_commit_bound_and_byte_deterministic(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-closed-loop-") as directory:
|
|
root = Path(directory)
|
|
first = self.build(root / "first")
|
|
second = self.build(root / "second")
|
|
first_artifact = Path(first["artifact"])
|
|
second_artifact = Path(second["artifact"])
|
|
|
|
self.assertEqual(first["sourceCommit"], SOURCE_COMMIT)
|
|
self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes())
|
|
self.assertEqual(
|
|
first["sha256"],
|
|
hashlib.sha256(first_artifact.read_bytes()).hexdigest(),
|
|
)
|
|
self.assertEqual(first["services"], ["nodedc-backend", "app"])
|
|
self.assertEqual(first["mcpVersion"], "0.7.0")
|
|
|
|
def test_artifact_is_a_safe_exact_engine_overlay(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-engine-l2-overlay-") as directory:
|
|
root = Path(directory)
|
|
built = self.build(root / "artifact")
|
|
artifact = Path(built["artifact"])
|
|
|
|
with tarfile.open(artifact, "r:gz") as archive:
|
|
members = archive.getmembers()
|
|
for member in members:
|
|
self.assertIn(member.type, (tarfile.REGTYPE, tarfile.DIRTYPE))
|
|
self.assertEqual(member.uid, 0)
|
|
self.assertEqual(member.gid, 0)
|
|
self.assertEqual(member.mtime, 0)
|
|
self.assertNotIn(".DS_Store", member.name)
|
|
self.assertNotIn("__MACOSX", member.name)
|
|
archive.extractall(root / "extract", filter="data")
|
|
|
|
manifest, entries, payload = RUNNER.load_artifact(
|
|
artifact,
|
|
root / "loaded",
|
|
)
|
|
self.assertEqual(manifest["component"], "engine")
|
|
self.assertEqual(manifest["type"], "app-overlay")
|
|
self.assertEqual(tuple(entries), tuple(built["entries"]))
|
|
self.assertEqual(
|
|
RUNNER.component_services("engine", entries),
|
|
("nodedc-backend", "app"),
|
|
)
|
|
self.assertEqual(
|
|
RUNNER.component_healthchecks(
|
|
"engine",
|
|
entries,
|
|
("nodedc-backend", "app"),
|
|
),
|
|
(
|
|
"http://127.0.0.1:8080/",
|
|
"http://127.0.0.1:3001/health",
|
|
),
|
|
)
|
|
self.assertTrue(RUNNER.component_publish_dist("engine", entries))
|
|
self.assertEqual(RUNNER.component_builds("engine", entries), ())
|
|
|
|
payload_files = {
|
|
path.relative_to(payload).as_posix()
|
|
for path in payload.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
self.assertEqual(payload_files, set(built["targetSha256"]))
|
|
for relative_path, expected in built["targetSha256"].items():
|
|
self.assertEqual(
|
|
hashlib.sha256((payload / relative_path).read_bytes()).hexdigest(),
|
|
expected,
|
|
)
|
|
|
|
self.assertTrue(all("/tests/" not in path for path in payload_files))
|
|
self.assertTrue(all("/data/" not in path for path in payload_files))
|
|
self.assertTrue(all("/storage/" not in path for path in payload_files))
|
|
self.assertTrue(all("/logs/" not in path for path in payload_files))
|
|
self.assertTrue(all(not path.endswith(".env") for path in payload_files))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|