weichen
01b7799663
Why: An immutable npm lock caches tarballs but packed consumers also require resolver metadata before candidate execution. Evidence: SELF host replay passed 57 composition and CI credential cases; real npm 10 and 11 cold-cache consumers passed without lock changes. Impact: Acquire only installed exact-version npmjs metadata with clean config, preserve cache paths for offline execution, adopt the repaired Frontend source and record immutable CI fixtures. Refs: R6-R8, R14; dependency-workspace U4/U5; UI 63d289e and Frontend 5467517
701 lines
44 KiB
Python
701 lines
44 KiB
Python
"""Real Git/npm/Go fixtures for immutable product composition."""
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import zipfile
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "scripts/build-composition.py"
|
|
RDM_SCRIPTS = Path(os.environ.get("COMPOSITION_RDM_SOURCE", str(SCRIPT.parent.parent / ".migration/Agent-System-RDM")))
|
|
LEGACY_PRODUCT = Path(__file__).resolve().parent / "fixtures/rdm-legacy-product.sh"
|
|
|
|
|
|
class LocalNpmRegistry:
|
|
"""Real npm cache/registry traffic with no public-network dependency."""
|
|
name = "composition-cold-fixture"
|
|
|
|
def __init__(self, base, env):
|
|
self.base, self.env = base, env
|
|
self.requests = []
|
|
self.tarballs = {}
|
|
self.integrities = {}
|
|
for version in ("1.0.0", "1.0.1"):
|
|
payload = json.dumps({"name": self.name, "version": version,
|
|
"scripts": {"install": "node -e 'process.exit(91)'"}}).encode()
|
|
stream = io.BytesIO()
|
|
with tarfile.open(fileobj=stream, mode="w:gz") as archive:
|
|
entry = tarfile.TarInfo("package/package.json")
|
|
entry.size = len(payload)
|
|
archive.addfile(entry, io.BytesIO(payload))
|
|
self.tarballs[version] = stream.getvalue()
|
|
self.integrities[version] = "sha512-" + base64.b64encode(hashlib.sha512(stream.getvalue()).digest()).decode()
|
|
registry = self
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
registry.requests.append((self.path, self.headers.get("Authorization")))
|
|
if self.path == "/" + registry.name:
|
|
versions = {v: {"name": registry.name, "version": v, "dist": {
|
|
"tarball": registry.url + "/" + registry.name + "/-/" + v + ".tgz",
|
|
"integrity": registry.integrities[v]}} for v in registry.tarballs}
|
|
body = json.dumps({"name": registry.name, "dist-tags": {"latest": "1.0.1"}, "versions": versions}).encode()
|
|
content_type = "application/json"
|
|
else:
|
|
version = self.path.rsplit("/", 1)[-1].removesuffix(".tgz")
|
|
if version not in registry.tarballs:
|
|
self.send_error(404)
|
|
return
|
|
body, content_type = registry.tarballs[version], "application/octet-stream"
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
self.url = "http://127.0.0.1:" + str(self.server.server_port)
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.cache = base / "cold-npm-cache"
|
|
self.global_config = base / "empty-npm-globalrc"
|
|
self.global_config.write_text("")
|
|
self.home = base / "cold-home"
|
|
self.home.mkdir()
|
|
self.tools = base / "npm-tools"
|
|
self.tools.mkdir()
|
|
self.calls = base / "npm-calls.jsonl"
|
|
npm = shutil.which("npm")
|
|
wrapper = self.tools / "npm"
|
|
wrapper.write_text("#!" + sys.executable + "\n" +
|
|
"import json, os, sys\nfrom pathlib import Path\nargs = sys.argv[1:]\n" +
|
|
"with Path(" + repr(str(self.calls)) + ").open('a') as output:\n" +
|
|
" output.write(json.dumps({'args': args, 'credential_keys': [k for k in os.environ if any(w in k.upper() for w in ('TOKEN', 'SECRET', 'PASSWORD'))]}) + '\\n')\n" +
|
|
"if args[:2] == ['cache', 'add'] and args[2].startswith(" + repr(self.name + "@") + "):\n" +
|
|
" args = [a.replace('https://registry.npmjs.org', " + repr(self.url) + ") for a in args]\n" +
|
|
"if args[:1] == ['ci']: args.append('--offline')\n" +
|
|
"os.execv(" + repr(npm) + ", [" + repr(npm) + ", *args])\n")
|
|
wrapper.chmod(0o755)
|
|
self.env = env | {"HOME": str(self.home), "npm_config_cache": str(self.cache),
|
|
"PATH": str(self.tools) + os.pathsep + env["PATH"]}
|
|
# Seed only the locked tarball, exactly the state produced by npm ci.
|
|
tarball = base / "locked.tgz"
|
|
tarball.write_bytes(self.tarballs["1.0.0"])
|
|
subprocess.run([npm, "cache", "add", str(tarball), "--ignore-scripts", "--userconfig=/dev/null",
|
|
"--globalconfig=" + str(self.global_config)], env=self.env, check=True, capture_output=True)
|
|
|
|
def close(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join()
|
|
|
|
def manifests(self, source):
|
|
source.mkdir(exist_ok=True)
|
|
manifest = {"name": "fixture", "version": "1.0.0", "dependencies": {self.name: "1.0.0"}}
|
|
(source / "package.json").write_text(json.dumps(manifest))
|
|
lock = {"name": "fixture", "version": "1.0.0", "lockfileVersion": 3, "requires": True,
|
|
"packages": {"": manifest, "node_modules/" + self.name: {
|
|
"version": "1.0.0", "resolved": "https://registry.npmjs.org/" + self.name + "/-/1.0.0.tgz",
|
|
"integrity": self.integrities["1.0.0"]}}}
|
|
(source / "package-lock.json").write_text(json.dumps(lock))
|
|
return (source / "package-lock.json").read_bytes()
|
|
|
|
def consumer(self, directory, env=None):
|
|
directory.mkdir(exist_ok=True)
|
|
(directory / "package.json").write_text(json.dumps({"dependencies": {self.name: "1.0.0"}}))
|
|
return subprocess.run(["npm", "install", "--offline", "--ignore-scripts", "--no-audit", "--no-fund",
|
|
"--registry=" + self.url, "--userconfig=/dev/null", "--globalconfig=" + str(self.global_config)],
|
|
cwd=directory, env=env or self.env, capture_output=True, text=True)
|
|
|
|
|
|
@unittest.skipUnless(shutil.which("npm"), "npm required")
|
|
class NpmAcquisitionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="composition-cold-npm-")
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.base = Path(self.temporary.name)
|
|
self.registry = LocalNpmRegistry(self.base, os.environ.copy())
|
|
self.addCleanup(self.registry.close)
|
|
spec = importlib.util.spec_from_file_location("cold_npm_composer", SCRIPT)
|
|
self.composer = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.composer)
|
|
|
|
def test_locked_metadata_is_acquired_before_offline_consumer_without_credentials(self):
|
|
source = self.base / "source"
|
|
original = self.registry.manifests(source)
|
|
red = self.registry.consumer(self.base / "before")
|
|
self.assertNotEqual(red.returncode, 0)
|
|
self.assertIn("ENOTCACHED", red.stderr)
|
|
self.composer.npm_dependencies(source, {}, self.registry.env | {"GITEA_TOKEN": "synthetic-secret"})
|
|
self.assertEqual((source / "package-lock.json").read_bytes(), original)
|
|
green = self.registry.consumer(self.base / "after", self.composer.execution_environment(self.registry.env))
|
|
self.assertEqual(green.returncode, 0, green.stdout + green.stderr)
|
|
actual = json.loads((self.base / "after/node_modules" / self.registry.name / "package.json").read_text())
|
|
self.assertEqual(actual["version"], "1.0.0")
|
|
calls = [json.loads(line) for line in self.registry.calls.read_text().splitlines()]
|
|
metadata = [call for call in calls if call["args"][:2] == ["cache", "add"]]
|
|
self.assertTrue(metadata, "tarball acquisition alone does not populate offline resolver metadata")
|
|
self.assertEqual(metadata[0]["args"][2], self.registry.name + "@1.0.0")
|
|
self.assertFalse(metadata[0]["credential_keys"])
|
|
self.assertTrue(self.registry.requests)
|
|
self.assertTrue(all(auth is None for _, auth in self.registry.requests))
|
|
|
|
def test_unreviewed_tarball_source_is_refused_before_acquisition(self):
|
|
source = self.base / "source"
|
|
self.registry.manifests(source)
|
|
path = source / "package-lock.json"
|
|
lock = json.loads(path.read_text())
|
|
lock["packages"]["node_modules/" + self.registry.name]["resolved"] = self.registry.url + "/untrusted.tgz"
|
|
path.write_text(json.dumps(lock))
|
|
with self.assertRaisesRegex(self.composer.CompositionError, "reviewed public registry"):
|
|
self.composer.npm_dependencies(source, {}, self.registry.env | {"GITEA_TOKEN": "synthetic-secret"})
|
|
self.assertFalse(self.registry.requests)
|
|
self.assertFalse(self.registry.calls.exists())
|
|
|
|
|
|
@unittest.skipUnless(all(shutil.which(c) for c in ("git", "go", "node", "npm")), "Git/Go/Node/npm required")
|
|
class CompositionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="composition-test-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.base = Path(self.temp.name)
|
|
self.root = self.base / "dev"
|
|
self.backend = self.base / "backend"
|
|
self.frontend = self.base / "frontend"
|
|
self.out = self.base / "output"
|
|
self.env = os.environ.copy()
|
|
self.env.update(GOWORK="off", GOPROXY="off", GOSUMDB="off", GOTOOLCHAIN="local")
|
|
self.env.pop("GOFLAGS", None)
|
|
for repo in (self.root, self.backend, self.frontend):
|
|
repo.mkdir()
|
|
self.run_command(["git", "init", "-q", "-b", "main"], repo)
|
|
self.run_command(["git", "config", "user.email", "fixture@example.invalid"], repo)
|
|
self.run_command(["git", "config", "user.name", "Composition Fixture"], repo)
|
|
self.write(self.backend, "go.mod", "module example.com/backend\n\ngo 1.23\n")
|
|
self.write(self.backend, "web/dist/.gitkeep", "")
|
|
self.write(self.backend, "web/embed.go", '''package web
|
|
import "embed"
|
|
//go:embed all:dist
|
|
var Files embed.FS
|
|
func Page() string { b, _ := Files.ReadFile("dist/index.html"); return string(b) }
|
|
''')
|
|
self.write(self.backend, "cmd/agent-system/main.go", '''package main
|
|
import ("fmt"; "os"; "example.com/backend/web")
|
|
var version = "fixture-dev"
|
|
func main() { if len(os.Args)>1 && os.Args[1]=="version" {fmt.Print(version)} else {fmt.Print(web.Page())} }
|
|
''')
|
|
self.write(self.backend, "cmd/storage-migrate/main.go", '''package main
|
|
import "fmt"
|
|
func main() { fmt.Print("migration-tool") }
|
|
''')
|
|
self.write(self.backend, "web/embed_test.go", '''package web
|
|
import "testing"
|
|
func TestProduct(t *testing.T) { if Page() != "immutable-page" { t.Fatalf("unexpected page: %q", Page()) } }
|
|
''')
|
|
self.backend_sha = self.commit(self.backend)
|
|
self.write(self.frontend, "package.json", json.dumps({
|
|
"name": "composition-fixture", "version": "1.0.0", "private": True,
|
|
"scripts": {"build": "node build.cjs", "test": "node test.cjs", "postinstall": "node -e \"process.exit(91)\""},
|
|
}))
|
|
self.write(self.frontend, "build.cjs", '''const fs = require('node:fs');
|
|
if (process.env.GITEA_TOKEN || process.env.SSH_AUTH_SOCK || process.env.GIT_CONFIG_COUNT) process.exit(90);
|
|
fs.mkdirSync('dist/assets', { recursive: true });
|
|
fs.writeFileSync('dist/index.html', 'immutable-page');
|
|
fs.writeFileSync('dist/assets/main.js', 'console.log("fixed")');
|
|
''')
|
|
self.write(self.frontend, "test.cjs", '''const fs = require('node:fs');
|
|
if (fs.readFileSync('dist/index.html', 'utf8') !== 'immutable-page') process.exit(92);
|
|
''')
|
|
self.run_command(["npm", "install", "--package-lock-only", "--ignore-scripts", "--no-audit", "--no-fund"], self.frontend)
|
|
self.frontend_sha = self.commit(self.frontend)
|
|
self.manifest = {"schema": 1, "components": [
|
|
{"id": "Agent-System-Backend", "repo": str(self.backend), "path": ".workspace/Agent-System-Backend", "role": "application", "modules": [{"path": "example.com/backend", "directory": "."}], "target": "main"},
|
|
{"id": "Agent-System-Frontend", "repo": str(self.frontend), "path": ".workspace/Agent-System-Frontend", "role": "application", "modules": [], "target": "main"},
|
|
]}
|
|
self.lock = {"schema": 1, "components": {
|
|
"Agent-System-Backend": {"commit": self.backend_sha},
|
|
"Agent-System-Frontend": {"commit": self.frontend_sha},
|
|
}}
|
|
self.save_inputs()
|
|
self.root_sha = self.commit(self.root)
|
|
|
|
def write(self, repo, path, value):
|
|
destination = repo / path
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text(value)
|
|
|
|
def run_command(self, command, cwd=None, env=None, check=True):
|
|
result = subprocess.run(command, cwd=cwd, env=env or self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
|
if check:
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
return result
|
|
|
|
def commit(self, repo):
|
|
self.run_command(["git", "add", "."], repo)
|
|
self.run_command(["git", "commit", "-qm", "Fixture"], repo)
|
|
return self.run_command(["git", "rev-parse", "HEAD"], repo).stdout.strip()
|
|
|
|
def save_inputs(self):
|
|
self.write(self.root, "components.json", json.dumps(self.manifest))
|
|
self.write(self.root, "components.lock.json", json.dumps(self.lock))
|
|
|
|
def compose(self, target="build", extra=(), env=None):
|
|
return self.run_command([sys.executable, "-B", str(SCRIPT), "--root", str(self.root), "--target", target, "--output", str(self.out), *extra], env=env, check=False)
|
|
|
|
def acquisition_guard(self, env, *, mark_frontend=True):
|
|
"""Forward to real tools, refusing acquisition after candidate execution."""
|
|
marker = self.base / "candidate-executed"
|
|
tools = self.base / "guarded-tools"
|
|
tools.mkdir()
|
|
for name in ("git", "go", "npm"):
|
|
executable = shutil.which(name)
|
|
wrapper = tools / name
|
|
wrapper.write_text("#!" + sys.executable + "\n" + '''import os, pathlib, sys
|
|
args = sys.argv[1:]
|
|
name = pathlib.Path(sys.argv[0]).name
|
|
acquisition = (name == 'git' and args and args[0] in {'clone', 'fetch', 'archive'}
|
|
or name == 'go' and args and (args[0] == 'mod' or args[0] == 'list')
|
|
or name == 'npm' and args and (args[0] in {'ci', 'install'} or args[:2] == ['cache', 'add']))
|
|
if acquisition and pathlib.Path(os.environ['COMPOSITION_EXEC_MARKER']).exists():
|
|
print('acquisition-after-execution: ' + name + ' ' + ' '.join(args[:2]), file=sys.stderr)
|
|
sys.exit(86)
|
|
''' + "os.execv(" + repr(executable) + ", [" + repr(executable) + ", *args])\n")
|
|
wrapper.chmod(0o755)
|
|
if mark_frontend:
|
|
self.write(self.frontend, "build.cjs", (self.frontend / "build.cjs").read_text()
|
|
+ "\nrequire('node:fs').writeFileSync(process.env.COMPOSITION_EXEC_MARKER, 'frontend-executed')\n")
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
return env | {"PATH": str(tools) + os.pathsep + env["PATH"],
|
|
"COMPOSITION_EXEC_MARKER": str(marker), "GITEA_TOKEN": "synthetic-phase-token"}
|
|
|
|
def test_all_product_acquisition_precedes_candidate_execution(self):
|
|
self.test_rdm_vendored_remote_backend_embeds_assets_without_mutating_module_cache()
|
|
self.write(self.root, "plugins/example/plugin.json", '{"name":"fixed-plugin"}')
|
|
self.commit(self.root)
|
|
env = self.acquisition_guard(self.env | {"GOPROXY": (self.base / "proxy").as_uri(),
|
|
"GOMODCACHE": str(self.base / "module-cache")})
|
|
result = self.compose(extra=("--rdm-source", str(self.base / "rdm")), env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual((self.base / "candidate-executed").read_text(), "frontend-executed")
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system-rdm-product")]).stdout, "immutable-page")
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system-admin")]).stdout, "immutable-page")
|
|
self.assertEqual((self.out / "plugins/example/plugin.json").read_text(), '{"name":"fixed-plugin"}')
|
|
|
|
def test_fixed_commits_embed_assets_without_touching_developer_checkout(self):
|
|
# Reading working trees or allowing go.work/npm overlays would produce DIRTY or fail.
|
|
self.write(self.backend, "web/dist/index.html", "DIRTY")
|
|
self.write(self.backend, "cmd/agent-system/main.go", "invalid local Go")
|
|
self.write(self.frontend, "build.cjs", "throw new Error('dirty local frontend')")
|
|
self.write(self.root, "go.work", "invalid developer workspace")
|
|
self.write(self.root, ".workspace/Agent-System-Frontend/package.json", '{"dependencies":{"bad":"file:/missing"}}')
|
|
env = self.env | {"GITEA_TOKEN": "fixture-secret", "SSH_AUTH_SOCK": "/missing-agent", "GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "credential.helper", "GIT_CONFIG_VALUE_0": "bad-helper"}
|
|
result = self.compose(env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system")]).stdout, "immutable-page")
|
|
self.assertEqual(self.run_command([str(self.out / "storage-migrate")]).stdout, "migration-tool")
|
|
self.assertEqual((self.backend / "web/dist/index.html").read_text(), "DIRTY")
|
|
resolved = json.loads((self.out / "resolved-manifest.json").read_text())
|
|
self.assertEqual(resolved["root"]["commit"], self.root_sha)
|
|
self.assertEqual(resolved["applications"]["Agent-System-Backend"]["commit"], self.backend_sha)
|
|
self.assertEqual(resolved["applications"]["Agent-System-Frontend"]["commit"], self.frontend_sha)
|
|
self.assertEqual(resolved["artifacts"]["agent-system"]["sha256"], hashlib.sha256((self.out / "agent-system").read_bytes()).hexdigest())
|
|
self.assertEqual(resolved["go"]["backend"]["modules"][0]["Path"], "example.com/backend")
|
|
self.assertTrue(resolved["npm"]["lock_sha256"])
|
|
self.assertTrue(resolved["frontend"]["sha256"])
|
|
self.assertNotIn("fixture-secret", (self.out / "resolved-manifest.json").read_text())
|
|
|
|
def test_test_target_runs_real_frontend_and_go_tests(self):
|
|
result = self.compose("test")
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("example.com/backend/web", result.stdout)
|
|
self.assertEqual(json.loads((self.out / "resolved-manifest.json").read_text())["target"], "test")
|
|
|
|
def test_unused_transitive_zip_does_not_require_consumer_sum_rewrite(self):
|
|
proxy = self.base / "unused-proxy"
|
|
for name, mod, body in (
|
|
("unused", "module example.com/unused\n\ngo 1.16\n", "package unused\n"),
|
|
("used", "module example.com/used\n\ngo 1.16\nrequire example.com/unused v1.0.0\n", "package used\nfunc Value() int { return 7 }\n"),
|
|
):
|
|
directory = proxy / "example.com" / name / "@v"
|
|
directory.mkdir(parents=True)
|
|
(directory / "v1.0.0.mod").write_text(mod)
|
|
(directory / "v1.0.0.info").write_text('{"Version":"v1.0.0","Time":"2026-10-13T00:00:00Z"}')
|
|
(directory / "list").write_text("v1.0.0\n")
|
|
with zipfile.ZipFile(directory / "v1.0.0.zip", "w") as archive:
|
|
archive.writestr(f"example.com/{name}@v1.0.0/go.mod", mod)
|
|
archive.writestr(f"example.com/{name}@v1.0.0/value.go", body)
|
|
env = self.env | {"GOPROXY": proxy.as_uri(), "GOMODCACHE": str(self.base / "unused-cache")}
|
|
self.write(self.backend, "go.mod", "module example.com/backend\n\ngo 1.23\nrequire example.com/used v1.0.0\n")
|
|
self.write(self.backend, "web/value.go", 'package web\nimport "example.com/used"\nvar Value = used.Value()\n')
|
|
self.run_command(["go", "mod", "tidy"], self.backend, env=env)
|
|
sums = (self.backend / "go.sum").read_text()
|
|
self.assertNotIn("example.com/unused v1.0.0 h1:", sums)
|
|
self.lock["components"]["Agent-System-Backend"]["commit"] = self.commit(self.backend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose(env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual((self.backend / "go.sum").read_text(), sums)
|
|
graph = json.loads((self.out / "resolved-manifest.json").read_text())["go"]["backend"]["modules"]
|
|
self.assertIn("example.com/used", {module["Path"] for module in graph})
|
|
|
|
def test_developer_package_tarball_overrides_do_not_reach_formal_tests(self):
|
|
self.write(self.frontend, "test.cjs", "if (process.env.CONTRACTS_TARBALL || process.env.UI_TARBALL) process.exit(93)")
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose("test", env=self.env | {"CONTRACTS_TARBALL": "/tmp/dirty-contracts.tgz", "UI_TARBALL": "/tmp/dirty-ui.tgz"})
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
|
|
def test_test_target_checks_fixed_go_library_and_ui_sources(self):
|
|
library = self.base / "probe-library"
|
|
library.mkdir()
|
|
self.run_command(["git", "init", "-q", "-b", "main"], library)
|
|
self.run_command(["git", "config", "user.email", "fixture@example.invalid"], library)
|
|
self.run_command(["git", "config", "user.name", "Fixture"], library)
|
|
self.write(library, "go.mod", "module example.com/probe\n\ngo 1.23\n")
|
|
self.write(library, "probe.go", "package probe\nfunc Text() string { return \"\" }\n")
|
|
self.write(library, "probe_test.go", '''package probe
|
|
import ("testing"; "os")
|
|
func TestFixedLibrary(t *testing.T) {
|
|
if err:=os.WriteFile(os.Getenv("COMPOSITION_GO_MARKER"), []byte("fixed-go-test"), 0600); err!=nil {t.Fatal(err)}
|
|
if path:=os.Getenv("COMPOSITION_EXEC_MARKER"); path!="" { if err:=os.WriteFile(path, []byte("library-executed"), 0600); err!=nil {t.Fatal(err)} }
|
|
}
|
|
''')
|
|
library_sha = self.commit(library)
|
|
version = "v0.0.0-" + self.run_command(["git", "show", "-s", "--format=%cd", "--date=format-local:%Y%m%d%H%M%S", library_sha], library).stdout.strip() + "-" + library_sha[:12]
|
|
proxy = self.base / "probe-proxy/example.com/probe/@v"
|
|
proxy.mkdir(parents=True)
|
|
(proxy / (version + ".mod")).write_text((library / "go.mod").read_text())
|
|
(proxy / (version + ".info")).write_text(json.dumps({"Version": version, "Time": "2026-10-13T00:00:00Z", "Origin": {"VCS": "git", "URL": str(library), "Hash": library_sha}}))
|
|
(proxy / "list").write_text(version + "\n")
|
|
with zipfile.ZipFile(proxy / (version + ".zip"), "w") as archive:
|
|
for path in self.run_command(["git", "ls-files"], library).stdout.splitlines():
|
|
archive.write(library / path, "example.com/probe@" + version + "/" + path)
|
|
self.write(self.backend, "go.mod", f"module example.com/backend\n\ngo 1.23\nrequire example.com/probe {version}\n")
|
|
self.write(self.backend, "web/probe.go", 'package web\nimport "example.com/probe"\nvar _ = probe.Text()\n')
|
|
env = self.env | {"GOPROXY": (self.base / "probe-proxy").as_uri(), "GOMODCACHE": str(self.base / "probe-cache"),
|
|
"COMPOSITION_GO_MARKER": str(self.base / "go-tested"), "COMPOSITION_UI_MARKER": str(self.base / "ui-tested")}
|
|
self.run_command(["go", "mod", "download", "all"], self.backend, env=env)
|
|
self.lock["components"]["Agent-System-Backend"]["commit"] = self.commit(self.backend)
|
|
self.write(library, "probe_test.go", "invalid dirty library tests")
|
|
ui = self.base / "ui"
|
|
ui.mkdir()
|
|
self.run_command(["git", "init", "-q", "-b", "main"], ui)
|
|
self.run_command(["git", "config", "user.email", "fixture@example.invalid"], ui)
|
|
self.run_command(["git", "config", "user.name", "Fixture"], ui)
|
|
self.write(ui, "package.json", json.dumps({"name": "@agent-system/ui", "version": "1.0.0", "scripts": {"test": "node test.cjs", "typecheck": "node -e \"process.exit(0)\""}}))
|
|
self.write(ui, "test.cjs", "require('node:fs').writeFileSync(process.env.COMPOSITION_UI_MARKER, 'fixed-ui-test')")
|
|
self.run_command(["npm", "install", "--package-lock-only", "--ignore-scripts", "--no-audit", "--no-fund"], ui)
|
|
ui_sha = self.commit(ui)
|
|
package = json.loads((self.frontend / "package.json").read_text())
|
|
package["dependencies"] = {"@agent-system/ui": "git+" + ui.as_uri() + "#" + ui_sha}
|
|
self.write(self.frontend, "package.json", json.dumps(package))
|
|
self.run_command(["npm", "install", "--package-lock-only", "--ignore-scripts", "--no-audit", "--no-fund"], self.frontend)
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.write(ui, "test.cjs", "throw Error('dirty ui tests')")
|
|
self.manifest["components"].extend([
|
|
{"id": "Probe", "repo": str(library), "path": ".workspace/Probe", "role": "go-library", "modules": [{"path": "example.com/probe", "directory": "."}], "target": "main"},
|
|
{"id": "Agent-System-UI", "repo": ui.as_uri(), "path": ".workspace/Agent-System-UI", "role": "npm-library", "modules": [], "target": "main"},
|
|
])
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
env = self.acquisition_guard(env, mark_frontend=False)
|
|
result = self.compose("test", env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertTrue((self.base / "go-tested").is_file(), "resolved Go provider tests were not executed")
|
|
self.assertEqual((self.base / "go-tested").read_text(), "fixed-go-test")
|
|
self.assertTrue((self.base / "ui-tested").is_file(), "fixed npm UI provider tests were not executed")
|
|
self.assertEqual((self.base / "ui-tested").read_text(), "fixed-ui-test")
|
|
self.assertIn("go vet", result.stdout)
|
|
|
|
def test_version_and_plugin_assets_come_from_committed_root(self):
|
|
self.write(self.root, "plugins/example/plugin.json", '{"name":"immutable-plugin"}')
|
|
self.root_sha = self.commit(self.root)
|
|
self.write(self.root, "plugins/example/plugin.json", '{"name":"dirty-plugin"}')
|
|
result = self.compose()
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
resolved = json.loads((self.out / "resolved-manifest.json").read_text())
|
|
self.assertIn("version", resolved)
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system"), "version"]).stdout, resolved["version"])
|
|
self.assertIn(self.root_sha[:12], resolved["version"])
|
|
self.assertIn(self.backend_sha[:12], resolved["version"])
|
|
self.assertIn(self.frontend_sha[:12], resolved["version"])
|
|
self.assertEqual((self.out / "plugins/example/plugin.json").read_text(), '{"name":"immutable-plugin"}')
|
|
self.assertTrue(resolved["assets"]["plugins"]["sha256"])
|
|
|
|
def test_npm_local_dependency_overlay_is_rejected(self):
|
|
lock = json.loads((self.frontend / "package-lock.json").read_text())
|
|
lock["packages"]["node_modules/local"] = {"resolved": "file:../dirty"}
|
|
self.write(self.frontend, "package-lock.json", json.dumps(lock))
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("local", result.stdout)
|
|
self.assertFalse((self.out / "agent-system").exists())
|
|
|
|
def test_frontend_dist_symlink_cannot_import_assets_from_outside_source(self):
|
|
external = self.base / "external-assets"
|
|
self.write(external, "index.html", "escaped-page")
|
|
self.write(external, "assets/main.js", "escaped-code")
|
|
self.write(self.frontend, "build.cjs", "require('node:fs').symlinkSync(" + json.dumps(str(external)) + ", 'dist', 'dir')")
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("symlink", result.stdout)
|
|
self.assertFalse((self.out / "agent-system").exists())
|
|
|
|
def test_project_test_mode_refuses_missing_boundary_gate(self):
|
|
self.write(self.root, "Makefile", "test-governance:\n\ttrue\n")
|
|
self.commit(self.root)
|
|
result = self.compose("test")
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("lint-boundaries", result.stdout)
|
|
|
|
def test_fixed_root_governance_and_boundaries_receive_proven_source_map(self):
|
|
self.write(self.root, "Makefile", "test-governance:\n\tpython3 -B scripts/map-check.py governance\n")
|
|
self.write(self.root, "scripts/lint-boundaries.sh", "#!/bin/sh\npython3 -B scripts/map-check.py boundaries\n")
|
|
self.write(self.root, "scripts/check-pins.sh", "#!/bin/sh\npython3 -B scripts/map-check.py pins\n")
|
|
self.write(self.root, "scripts/map-check.py", '''import json, os, pathlib, sys
|
|
root = pathlib.Path.cwd()
|
|
assert not (root / '.git').exists(), 'gates must run from the immutable archive'
|
|
mapping = json.loads(pathlib.Path(os.environ['COMPOSITION_SOURCE_MAP']).read_text())
|
|
lock = json.loads((root / 'components.lock.json').read_text())
|
|
for name in ('Agent-System-Backend', 'Agent-System-Frontend'):
|
|
assert mapping['components'][name]['commit'] == lock['components'][name]['commit']
|
|
assert (pathlib.Path(mapping['components']['Agent-System-Backend']['path']) / 'web/dist/index.html').read_text() == 'immutable-page'
|
|
with pathlib.Path(os.environ['COMPOSITION_GATE_MARKER']).open('a') as output: output.write(sys.argv[1]+'\\n')
|
|
''')
|
|
self.commit(self.root)
|
|
# Dirty gate code is excluded, just like dirty application code.
|
|
self.write(self.root, "scripts/map-check.py", "raise Exception('dirty governance')")
|
|
marker = self.base / "gates-run"
|
|
result = self.compose("test", extra=("--source-map", str(self.base / "source-map.json")), env=self.env | {"COMPOSITION_GATE_MARKER": str(marker)})
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(marker.read_text(), "boundaries\npins\ngovernance\n")
|
|
self.assertEqual(json.loads((self.base / "source-map.json").read_text())["components"]["Agent-System-Backend"]["commit"], self.backend_sha)
|
|
|
|
def test_missing_frontend_assets_fails_before_publishing_binary(self):
|
|
self.write(self.frontend, "build.cjs", "require('node:fs').mkdirSync('dist', {recursive:true});")
|
|
self.lock["components"]["Agent-System-Frontend"]["commit"] = self.commit(self.frontend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("index.html", result.stdout)
|
|
self.assertFalse((self.out / "agent-system").exists())
|
|
|
|
def test_dirty_root_lock_is_rejected(self):
|
|
self.lock["components"]["Agent-System-Backend"]["commit"] = "a" * 40
|
|
self.save_inputs()
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
|
|
def test_dirty_root_composer_cannot_publish_under_committed_identity(self):
|
|
tools = self.root / "scripts"
|
|
(tools / "workspace").mkdir(parents=True)
|
|
shutil.copy2(SCRIPT, tools / "build-composition.py")
|
|
shutil.copy2(SCRIPT.parent / "workspace/transport.py", tools / "workspace/transport.py")
|
|
self.commit(self.root)
|
|
composer = tools / "build-composition.py"
|
|
composer.write_text(composer.read_text().replace(
|
|
'(destination / ".gitkeep").touch()',
|
|
'(destination / ".gitkeep").touch(); (destination / "index.html").write_text("dirty-tool-page")'))
|
|
result = self.run_command([sys.executable, str(composer), "--root", str(self.root),
|
|
"--target", "build", "--output", str(self.out)], check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("tool", result.stdout)
|
|
self.assertFalse((self.out / "agent-system").exists())
|
|
|
|
def test_dirty_root_transport_is_refused_before_import(self):
|
|
tools = self.root / "scripts"
|
|
(tools / "workspace").mkdir(parents=True)
|
|
shutil.copy2(SCRIPT, tools / "build-composition.py")
|
|
shutil.copy2(SCRIPT.parent / "workspace/transport.py", tools / "workspace/transport.py")
|
|
self.commit(self.root)
|
|
marker = self.base / "transport-executed"
|
|
(tools / "workspace/transport.py").write_text(
|
|
"from pathlib import Path\nPath(" + repr(str(marker)) + ").write_text('executed')\n")
|
|
result = self.run_command([sys.executable, str(tools / "build-composition.py"), "--root", str(self.root),
|
|
"--target", "build", "--output", str(self.out)], check=False)
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertFalse(marker.exists(), "dirty transport executed before source identity validation")
|
|
self.assertIn("committed", result.stdout)
|
|
|
|
def test_locked_missing_revision_is_rejected(self):
|
|
self.lock["components"]["Agent-System-Backend"]["commit"] = "a" * 40
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertFalse((self.out / "agent-system").exists())
|
|
|
|
def test_source_path_escape_is_rejected(self):
|
|
self.manifest["components"][0]["path"] = "../outside"
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("path", result.stdout)
|
|
|
|
def test_local_go_replace_is_rejected(self):
|
|
self.write(self.backend, "go.mod", "module example.com/backend\n\ngo 1.23\nreplace example.com/dirty => /missing/local\n")
|
|
self.lock["components"]["Agent-System-Backend"]["commit"] = self.commit(self.backend)
|
|
self.save_inputs()
|
|
self.commit(self.root)
|
|
result = self.compose()
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("replace", result.stdout)
|
|
|
|
def test_rdm_vendored_remote_backend_embeds_assets_without_mutating_module_cache(self):
|
|
rdm = self.base / "rdm"
|
|
rdm.mkdir()
|
|
self.run_command(["git", "init", "-q", "-b", "main"], rdm)
|
|
self.run_command(["git", "config", "user.email", "fixture@example.invalid"], rdm)
|
|
self.run_command(["git", "config", "user.name", "Fixture"], rdm)
|
|
version = "v0.0.0-" + self.run_command(["git", "show", "-s", "--format=%cd", "--date=format-local:%Y%m%d%H%M%S", self.backend_sha], self.backend).stdout.strip() + "-" + self.backend_sha[:12]
|
|
proxy = self.base / "proxy/example.com/backend/@v"
|
|
proxy.mkdir(parents=True)
|
|
(proxy / (version + ".mod")).write_text((self.backend / "go.mod").read_text())
|
|
(proxy / (version + ".info")).write_text(json.dumps({"Version": version, "Time": "2026-10-13T00:00:00Z"}))
|
|
(proxy / "list").write_text(version + "\n")
|
|
with zipfile.ZipFile(proxy / (version + ".zip"), "w") as archive:
|
|
paths = self.run_command(["git", "ls-files"], self.backend).stdout.splitlines()
|
|
for path in paths:
|
|
archive.write(self.backend / path, "example.com/backend@" + version + "/" + path)
|
|
self.write(rdm, "go.mod", f"module example.com/rdm\n\ngo 1.23\nrequire example.com/backend {version}\n")
|
|
self.write(rdm, "main.go", 'package main\nimport ("fmt"; "example.com/backend/web")\nfunc main() { fmt.Print(web.Page()) }\n')
|
|
module_cache = self.base / "module-cache"
|
|
env = self.env | {"GOPROXY": (self.base / "proxy").as_uri(), "GOMODCACHE": str(module_cache)}
|
|
self.run_command(["go", "mod", "download", "all"], rdm, env=env)
|
|
rdm_sha = self.commit(rdm)
|
|
cached_asset = module_cache / ("example.com/backend@" + version) / "web/dist/.gitkeep"
|
|
self.assertTrue(cached_asset.exists())
|
|
result = self.compose(extra=("--rdm-source", str(rdm)), env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system-rdm-product")]).stdout, "immutable-page")
|
|
self.assertFalse(cached_asset.with_name("index.html").exists())
|
|
resolved = json.loads((self.out / "resolved-manifest.json").read_text())
|
|
self.assertEqual(resolved["applications"]["Agent-System-RDM"]["commit"], rdm_sha)
|
|
self.assertTrue(resolved["go"]["rdm"]["vendor_modules_sha256"])
|
|
self.assertTrue(resolved["go"]["rdm"]["vendor_backend_source_sha256"])
|
|
self.assertEqual(resolved["go"]["rdm"]["modules"][1]["Version"], version)
|
|
|
|
@unittest.skipUnless((RDM_SCRIPTS / "scripts/build-product.sh").is_file(), "external RDM source was not supplied; host must replay the actual wrapper")
|
|
def test_versioned_rdm_product_entry_uses_fixed_composer(self):
|
|
# Reuse the real module fixture first; its independent embed test remains
|
|
# mandatory even when an external RDM wrapper is unavailable.
|
|
self.test_rdm_vendored_remote_backend_embeds_assets_without_mutating_module_cache()
|
|
rdm = self.base / "rdm"
|
|
version = self.run_command(["go", "list", "-m", "-f", "{{.Version}}", "example.com/backend"], rdm,
|
|
env=self.env | {"GOPROXY": (self.base / "proxy").as_uri(), "GOMODCACHE": str(self.base / "module-cache")}).stdout.strip()
|
|
env = self.env | {"GOPROXY": (self.base / "proxy").as_uri(), "GOMODCACHE": str(self.base / "module-cache")}
|
|
|
|
# The versioned RDM entry invokes the same fixed Dev composer, even when
|
|
# local RDM Go source is dirty. No API-only fallback is allowed.
|
|
self.write(rdm, "go.mod", f"module git-ssh.u1680832.nyat.app/weichen/Agent-System-RDM.git\n\ngo 1.23\nrequire example.com/backend {version}\n")
|
|
shutil.copy2(RDM_SCRIPTS / "Makefile", rdm / "Makefile")
|
|
(rdm / "scripts").mkdir()
|
|
shutil.copy2(RDM_SCRIPTS / "scripts/build-product.sh", rdm / "scripts/build-product.sh")
|
|
self.commit(rdm)
|
|
self.write(rdm, "main.go", "invalid local RDM")
|
|
(self.root / "scripts/workspace").mkdir(parents=True)
|
|
shutil.copy2(SCRIPT, self.root / "scripts/build-composition.py")
|
|
shutil.copy2(SCRIPT.parent / "workspace/transport.py", self.root / "scripts/workspace/transport.py")
|
|
self.commit(self.root)
|
|
result = self.run_command(["make", "product", "COMPOSITION_ROOT=" + str(self.root)], rdm, env=env, check=False)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(self.run_command([str(rdm / "bin/agent-system-rdm-product")]).stdout, "immutable-page")
|
|
self.assertTrue((rdm / "bin/agent-system-admin").exists())
|
|
self.assertEqual(json.loads((rdm / "bin/PRODUCT_MANIFEST.json").read_text())["kind"], "agent-system-composition/1")
|
|
|
|
def test_modified_go_module_cache_is_rejected_before_product_publication(self):
|
|
self.test_rdm_vendored_remote_backend_embeds_assets_without_mutating_module_cache()
|
|
cached = next((self.base / "module-cache/example.com").glob("backend@*/web/embed.go"))
|
|
cached.chmod(0o644)
|
|
cached.write_text(cached.read_text() + "\n// unauthorized cache mutation\n")
|
|
result = self.compose(extra=("--rdm-source", str(self.base / "rdm")),
|
|
env=self.env | {"GOPROXY": (self.base / "proxy").as_uri(), "GOMODCACHE": str(self.base / "module-cache")})
|
|
self.assertNotEqual(result.returncode, 0, result.stdout)
|
|
self.assertIn("modified", result.stdout)
|
|
|
|
def test_rdm_product_and_admin_follow_its_selected_backend_not_dev_source_lock(self):
|
|
self.test_rdm_vendored_remote_backend_embeds_assets_without_mutating_module_cache()
|
|
selected_embed = (self.backend / "web/embed.go").read_text().replace("return string(b)", 'return "selected-rdm:" + string(b)')
|
|
self.write(self.backend, "web/embed.go", selected_embed)
|
|
selected_sha = self.commit(self.backend)
|
|
selected_version = "v0.0.0-" + self.run_command(["git", "show", "-s", "--format=%cd", "--date=format-local:%Y%m%d%H%M%S", selected_sha], self.backend).stdout.strip() + "-" + selected_sha[:12]
|
|
proxy = self.base / "proxy/example.com/backend/@v"
|
|
(proxy / (selected_version + ".mod")).write_text((self.backend / "go.mod").read_text())
|
|
(proxy / (selected_version + ".info")).write_text(json.dumps({"Version": selected_version, "Time": "2026-10-13T00:00:00Z", "Origin": {"VCS": "git", "URL": str(self.backend), "Hash": selected_sha}}))
|
|
with zipfile.ZipFile(proxy / (selected_version + ".zip"), "w") as archive:
|
|
for path in self.run_command(["git", "ls-files"], self.backend).stdout.splitlines():
|
|
archive.write(self.backend / path, "example.com/backend@" + selected_version + "/" + path)
|
|
rdm = self.base / "rdm"
|
|
self.write(rdm, "go.mod", f"module example.com/rdm\n\ngo 1.23\nrequire example.com/backend {selected_version}\n")
|
|
env = self.env | {"GOPROXY": (self.base / "proxy").as_uri(), "GOMODCACHE": str(self.base / "module-cache")}
|
|
self.run_command(["go", "mod", "download", "all"], rdm, env=env)
|
|
self.commit(rdm)
|
|
result = self.compose(extra=("--rdm-source", str(rdm)), env=env)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system")]).stdout, "immutable-page")
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system-rdm-product")]).stdout, "selected-rdm:immutable-page")
|
|
self.assertEqual(self.run_command([str(self.out / "agent-system-admin")]).stdout, "selected-rdm:immutable-page")
|
|
resolved = json.loads((self.out / "resolved-manifest.json").read_text())
|
|
self.assertEqual(resolved["applications"]["Agent-System-Backend"]["commit"], self.backend_sha)
|
|
self.assertEqual(resolved["go"]["rdm"]["backend_source"]["commit"], selected_sha)
|
|
self.assertEqual(resolved["go"]["rdm"]["backend_source"]["version"], selected_version)
|
|
admin_version = self.run_command([str(self.out / "agent-system-admin"), "version"]).stdout
|
|
self.assertEqual(admin_version, resolved["go"]["rdm"]["admin_version"])
|
|
self.assertIn(selected_sha[:12], admin_version)
|
|
self.assertFalse((self.base / "module-cache/example.com" / ("backend@" + selected_version) / "web/dist/index.html").exists())
|
|
|
|
def test_historical_rdm_product_still_builds_from_legacy_pins(self):
|
|
rdm = self.base / "legacy-rdm"
|
|
rdm.mkdir()
|
|
self.run_command(["git", "init", "-q", "-b", "main"], rdm)
|
|
self.run_command(["git", "config", "user.email", "fixture@example.invalid"], rdm)
|
|
self.run_command(["git", "config", "user.name", "Fixture"], rdm)
|
|
self.write(rdm, "go.mod", "module legacy-rdm\n\ngo 1.23\nrequire example.com/backend v0.0.0\nreplace example.com/backend => ./pins/agent-system-backend\n")
|
|
self.write(rdm, "main.go", 'package main\nimport ("fmt"; "example.com/backend/web")\nfunc main() { fmt.Print(web.Page()) }\n')
|
|
(rdm / "scripts").mkdir()
|
|
shutil.copy2(LEGACY_PRODUCT, rdm / "scripts/build-product.sh")
|
|
self.commit(rdm)
|
|
pins = rdm / "pins"
|
|
pins.mkdir()
|
|
for name, source in (("agent-system-backend", self.backend), ("agent-system-frontend", self.frontend), ("agent-system-ui", self.frontend)):
|
|
self.run_command(["git", "clone", "-q", str(source), str(pins / name)])
|
|
for name in ("agent-system-frontend", "agent-system-ui"):
|
|
package = json.loads((pins / name / "package.json").read_text())
|
|
del package["scripts"]["postinstall"]
|
|
self.write(pins / name, "package.json", json.dumps(package))
|
|
self.run_command(["npm", "install", "--package-lock-only", "--ignore-scripts", "--no-audit", "--no-fund"], pins / name)
|
|
result = self.run_command(["bash", "scripts/build-product.sh"], rdm, check=False)
|
|
self.assertEqual(result.returncode, 0, result.stdout)
|
|
self.assertEqual(self.run_command([str(rdm / "bin/agent-system-rdm-product")]).stdout, "immutable-page")
|
|
self.assertEqual(json.loads((rdm / "bin/PRODUCT_MANIFEST.json").read_text())["kind"], "rdm-product/1")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|