weichen
0d9f44a7e2
Why: A dirty composer could change artifacts under a committed Dev identity, and two CI jobs reacquired credentials after executing candidate build code. Evidence: SELF regressions reproduce both issues before the fixes; all22 real composition cases and21 dependency/CI cases pass after. CI syntax checked; full exact committed replay follows this checkpoint. Impact: Verify executing composer/transport bytes before import and publication, record explicit tooling provenance, order all privileged acquisition first, and model CGO correctly in the nested component race fixture. Existing product and required CI gates remain. Refs: Dependency-workspace early code review; correctness/standards/security P1 findings and R7-R8/R14.
505 lines
33 KiB
Python
505 lines
33 KiB
Python
"""Real Git/npm/Go fixtures for immutable product composition."""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
|
|
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"
|
|
|
|
|
|
@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 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)} }
|
|
''')
|
|
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)
|
|
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()
|