weichen
f2b7a9af68
Why: The lifecycle pin verification step executes candidate Python and shell code even when its YAML run value is inline. Evidence: SELF regression observes three credential-order failures on the prior candidate, then passes after moving pin verification behind all privileged acquisition and recognizing inline run entries. Impact: Retain the pin verification gate while ending credential use before every candidate entry point. Refs: Dependency-workspace independent security and standards validation; R8 and CI phase boundary.
376 lines
23 KiB
Python
376 lines
23 KiB
Python
"""Replay trusted workflow phases with real Git, Go, npm and disposable HTTP repos."""
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
import subprocess
|
|
import shutil
|
|
import tempfile
|
|
import textwrap
|
|
import unittest
|
|
|
|
from tests import test_ci_credentials as fixture
|
|
|
|
ROOT_TOKEN, SUB_TOKEN, SSH, WORKFLOW = fixture.ROOT_TOKEN, fixture.SUB_TOKEN, fixture.SSH, fixture.WORKFLOW
|
|
|
|
MODULE = "git-ssh.u1680832.nyat.app/weichen/Logging-System.git"
|
|
|
|
|
|
def phase(workflow, name):
|
|
return fixture.workflow_script(workflow, name)
|
|
|
|
|
|
class ComponentWorkflowFixtureTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.base = Path(self.temporary.name)
|
|
self.snapshots = self.base / "snapshots"
|
|
shutil.copytree(fixture.COMPONENT_FIXTURES, self.snapshots)
|
|
|
|
def test_all_required_component_snapshots_have_fixed_provenance(self):
|
|
for repository in fixture.COMPONENT_REPOSITORIES:
|
|
with self.subTest(repository=repository):
|
|
path = fixture.component_workflow(repository)
|
|
self.assertEqual(path, fixture.COMPONENT_FIXTURES / repository / "ci.yml")
|
|
self.assertTrue(phase(path, "Checkout"))
|
|
|
|
def test_modified_workflow_snapshot_is_rejected_before_execution(self):
|
|
path = self.snapshots / "Agent-Native/ci.yml"
|
|
path.write_bytes(path.read_bytes() + b"\n# unexpected workflow replacement\n")
|
|
with self.assertRaisesRegex(ValueError, "recorded source digest"):
|
|
fixture.component_workflow("Agent-Native", snapshots=self.snapshots)
|
|
|
|
def test_missing_component_snapshot_is_not_skipped(self):
|
|
(self.snapshots / "Agent-Native/ci.yml").unlink()
|
|
with self.assertRaises(FileNotFoundError):
|
|
fixture.component_workflow("Agent-Native", snapshots=self.snapshots)
|
|
|
|
def test_omitted_repository_provenance_is_rejected(self):
|
|
path = self.snapshots / "provenance.json"
|
|
provenance = json.loads(path.read_text())
|
|
del provenance["components"]["Agent-Native"]
|
|
path.write_text(json.dumps(provenance))
|
|
with self.assertRaisesRegex(ValueError, "every reviewed repository"):
|
|
fixture.component_workflow("Agent-Native", snapshots=self.snapshots)
|
|
|
|
def test_optional_source_validation_uses_immutable_commit_and_rejects_stale_head(self):
|
|
source_root = self.base / "sources"
|
|
source = source_root / "Agent-Native"
|
|
workflow = source / ".gitea/workflows/ci.yml"
|
|
workflow.parent.mkdir(parents=True)
|
|
captured = self.snapshots / "Agent-Native/ci.yml"
|
|
workflow.write_bytes(captured.read_bytes())
|
|
env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")}
|
|
env.update(GIT_CONFIG_GLOBAL="/dev/null", GIT_CONFIG_SYSTEM="/dev/null")
|
|
|
|
def git(*args):
|
|
return subprocess.check_output(["git", "-C", str(source), *args], env=env, text=True).strip()
|
|
|
|
git("init", "-q")
|
|
git("add", ".")
|
|
git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-q", "-m", "Source")
|
|
metadata = self.snapshots / "provenance.json"
|
|
provenance = json.loads(metadata.read_text())
|
|
provenance["components"]["Agent-Native"]["source_sha"] = git("rev-parse", "HEAD")
|
|
metadata.write_text(json.dumps(provenance))
|
|
# A mutable file must not substitute for the committed Git object.
|
|
workflow.write_text("uncommitted workflow replacement\n")
|
|
self.assertEqual(fixture.component_workflow("Agent-Native", snapshots=self.snapshots,
|
|
source_root=source_root), captured)
|
|
git("add", ".")
|
|
git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid", "commit", "-q", "-m", "Changed")
|
|
with self.assertRaisesRegex(ValueError, "source HEAD differs"):
|
|
fixture.component_workflow("Agent-Native", snapshots=self.snapshots, source_root=source_root)
|
|
provenance["components"]["Agent-Native"]["source_sha"] = git("rev-parse", "HEAD")
|
|
metadata.write_text(json.dumps(provenance))
|
|
with self.assertRaisesRegex(ValueError, "immutable Git source"):
|
|
fixture.component_workflow("Agent-Native", snapshots=self.snapshots, source_root=source_root)
|
|
|
|
|
|
class DependencyCITests(unittest.TestCase):
|
|
def test_privileged_acquisition_finishes_before_candidate_execution_in_each_job(self):
|
|
content = WORKFLOW.read_text()
|
|
jobs = re.split(r"^ ([A-Za-z0-9_-]+):\n", content, flags=re.M)
|
|
for index in range(1, len(jobs), 2):
|
|
job, body = jobs[index:index+2]
|
|
executed = False
|
|
steps = re.split(r"^ - name: ([^\n]+)\n", body, flags=re.M)
|
|
for step in range(1, len(steps), 2):
|
|
name, script = steps[step:step+2]
|
|
with self.subTest(job=job, step=name):
|
|
if re.search(r"\$\{\{\s*secrets\.", script):
|
|
self.assertFalse(executed, "privileged acquisition follows candidate execution")
|
|
# run may be an inline YAML scalar or a multiline block;
|
|
# both shell and Python project entry points execute code.
|
|
if re.search(r"(?:^|\n)\s*(?:run:\s*)?(?:make\s|go\s+(?:build|test|vet)\s|npm\s+(?:run|test)\s|bash\s+scripts/|python3?\s+(?:-m\s+unittest|scripts/))", script):
|
|
executed = True
|
|
|
|
setUp = fixture.CheckoutTests.setUp
|
|
git = fixture.CheckoutTests.git
|
|
publish = fixture.CheckoutTests.publish
|
|
run_checkout = fixture.CheckoutTests.run_checkout
|
|
|
|
def run_phase(self, name, directory=None, workflow=WORKFLOW, extra=None, after=""):
|
|
env = self.env | {"SUBMODULE_TOKEN": SUB_TOKEN, "GITHUB_SERVER_URL": self.url,
|
|
"GITHUB_REPOSITORY": "weichen/Agent-System-Dev",
|
|
"GOCACHE": str(self.base / "gocache"),
|
|
"GOMODCACHE": str(self.base / "gomodcache"),
|
|
"npm_config_cache": str(self.base / "npm-cache")}
|
|
return subprocess.run(["bash", "-c", phase(workflow, name) + "\n" + after],
|
|
cwd=directory or self.checkout, env=env | (extra or {}),
|
|
capture_output=True, text=True, timeout=60)
|
|
|
|
def commit_fixture(self, name, files):
|
|
sha = self.publish(name)
|
|
source = self.base / name
|
|
for path, content in files.items():
|
|
dest = source / path
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
dest.write_text(content)
|
|
self.git(source, "add", ".")
|
|
subprocess.run(["git", "-C", str(source), "-c", "user.name=Fixture",
|
|
"-c", "user.email=fixture@example.invalid", "commit", "-q", "-m", "Package"],
|
|
env=self.env | {"GIT_AUTHOR_DATE": "2026-01-02T03:04:05Z",
|
|
"GIT_COMMITTER_DATE": "2026-01-02T03:04:05Z"}, check=True)
|
|
sha = self.git(source, "rev-parse", "HEAD")
|
|
self.git(source, "push", "-q", str(self.published / "weichen" / (name + ".git")), "main")
|
|
return sha
|
|
|
|
def fixture_checksums(self, module, sha):
|
|
"""Use Go's checksum algorithm on a local fixture, outside the cache under test."""
|
|
directory = self.base / "checksum-fixture"
|
|
directory.mkdir()
|
|
version = "v0.0.0-20260102030405-" + sha[:12]
|
|
(directory / "go.mod").write_text(f"module checksumfixture\n\ngo 1.25\nrequire {module} {version}\n")
|
|
repo = self.published / "weichen/Logging-System.git"
|
|
env = self.env | {"GOWORK": "off", "GOTOOLCHAIN": "local", "GOENV": "off", "GOFLAGS": "",
|
|
"GOPROXY": "off", "GOSUMDB": "off", "GOMODCACHE": str(self.base / "checksum-cache"),
|
|
"GOPRIVATE": "git-ssh.u1680832.nyat.app/weichen", "GONOPROXY": "git-ssh.u1680832.nyat.app/weichen",
|
|
"GIT_CONFIG_GLOBAL": "/dev/null", "GIT_CONFIG_COUNT": "3",
|
|
"GIT_CONFIG_KEY_0": "protocol.file.allow", "GIT_CONFIG_VALUE_0": "always",
|
|
"GIT_CONFIG_KEY_1": "url." + repo.as_uri() + ".insteadOf",
|
|
"GIT_CONFIG_VALUE_1": "https://git-ssh.u1680832.nyat.app/weichen/Logging-System",
|
|
"GIT_CONFIG_KEY_2": "core.hooksPath", "GIT_CONFIG_VALUE_2": "/dev/null"}
|
|
result = subprocess.run(["go", "mod", "download", "-json"], cwd=directory, env=env,
|
|
capture_output=True, text=True, check=True)
|
|
info = json.loads(result.stdout)
|
|
return f"{module} {version} {info['Sum']}\n{module} {version}/go.mod {info['GoModSum']}\n"
|
|
|
|
def test_registered_application_lock_materializes_fixed_source(self):
|
|
backend = self.publish("Agent-System-Backend")
|
|
root = self.commit_fixture("Agent-System-Dev", {
|
|
"components.json": json.dumps({"schema": 1, "components": [{
|
|
"id": "Agent-System-Backend", "repo": SSH + "Agent-System-Backend.git",
|
|
"path": ".workspace/Agent-System-Backend", "role": "application", "modules": []}, {
|
|
"id": "Agent-System-UI", "repo": SSH + "Agent-System-UI.git",
|
|
"path": ".workspace/Agent-System-UI", "role": "npm-library", "modules": []}]}),
|
|
"components.lock.json": json.dumps({"schema": 1, "components": {
|
|
"Agent-System-Backend": {"commit": backend}}})})
|
|
result = self.run_checkout(root)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue((self.checkout / ".workspace/Agent-System-Backend/.git").exists(),
|
|
"versioned application source was not materialized")
|
|
self.assertEqual(self.git(self.checkout / ".workspace/Agent-System-Backend", "rev-parse", "HEAD"), backend)
|
|
self.assertEqual(self.git(self.checkout / ".workspace/Agent-System-Backend", "config", "remote.origin.url"),
|
|
SSH + "Agent-System-Backend.git")
|
|
|
|
def test_unregistered_source_lock_is_rejected_before_component_fetch(self):
|
|
root = self.commit_fixture("Agent-System-Dev", {
|
|
"components.json": json.dumps({"schema": 1, "components": [{
|
|
"id": "Agent-System-Backend", "repo": self.url + "/untrusted/receiver",
|
|
"path": ".workspace/Agent-System-Backend", "role": "application", "modules": []}]}),
|
|
"components.lock.json": json.dumps({"schema": 1, "components": {
|
|
"Agent-System-Backend": {"commit": "a" * 40}}})})
|
|
result = self.run_checkout(root)
|
|
self.assertNotEqual(result.returncode, 0, "untrusted manifest was accepted")
|
|
self.assertFalse(any("Backend" in path or "untrusted" in path for path, _ in self.requests))
|
|
|
|
def test_go_prefetch_uses_only_registered_private_repo_and_local_toolchain(self):
|
|
sha = self.commit_fixture("Logging-System", {
|
|
"go.mod": f"module {MODULE}\n\ngo 1.25\n", "logging.go": "package logging\nconst Value = 42\n"})
|
|
(self.checkout / "go.mod").write_text(
|
|
f"module fixture\n\ngo 1.25\n\nrequire {MODULE} v0.0.0-20260102030405-{sha[:12]}\n")
|
|
(self.checkout / "go.sum").write_text(self.fixture_checksums(MODULE, sha))
|
|
(self.checkout / "probe.go").write_text(f'package fixture\nimport logging "{MODULE}"\nvar Value = logging.Value\n')
|
|
result = self.run_phase("Prefetch Go modules without executing candidate code",
|
|
extra={"GOTOOLCHAIN": "auto", "GIT_TRACE": "1"})
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertTrue(self.requests)
|
|
for path, auth in self.requests:
|
|
self.assertRegex(path, r"^/weichen/Logging-System\.git/")
|
|
self.assertEqual(auth, "token " + SUB_TOKEN)
|
|
self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr)
|
|
tested = subprocess.run(["go", "test", "./..."], cwd=self.checkout,
|
|
env=self.env | {"GOTOOLCHAIN": "local", "GOWORK": "off", "GOPROXY": "off",
|
|
"GOSUMDB": "off", "GOMODCACHE": str(self.base / "gomodcache"),
|
|
"GOCACHE": str(self.base / "gocache")}, capture_output=True, text=True)
|
|
self.assertEqual(tested.returncode, 0, tested.stdout + tested.stderr)
|
|
for config in self.base.rglob("config"):
|
|
if config.is_file():
|
|
self.assertNotIn(SUB_TOKEN, config.read_text(errors="replace"))
|
|
|
|
def test_unregistered_go_module_rejected_before_authenticated_request(self):
|
|
(self.checkout / "go.mod").write_text(
|
|
"module fixture\n\ngo 1.25\nrequire git-ssh.u1680832.nyat.app/weichen/Unlisted.git v1.0.0\n")
|
|
result = self.run_phase("Prefetch Go modules without executing candidate code")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.requests)
|
|
|
|
def test_npm_git_package_installs_committed_output_without_prepare(self):
|
|
marker = self.base / "prepare-ran"
|
|
sha = self.commit_fixture("Agent-System-Core", {
|
|
"package.json": json.dumps({"name": "@agent-system/contracts", "version": "1.0.0",
|
|
"files": ["package"], "main": "package/index.js"}),
|
|
"package/index.js": "module.exports = {value: 42}\n"})
|
|
spec = "git+" + SSH + "Agent-System-Core.git#" + sha
|
|
(self.checkout / "package.json").write_text(json.dumps({"name": "fixture", "version": "1.0.0",
|
|
"dependencies": {"@agent-system/contracts": spec},
|
|
"scripts": {"postinstall": f"touch '{marker}'"}}))
|
|
(self.checkout / "package-lock.json").write_text(json.dumps({"name": "fixture", "version": "1.0.0",
|
|
"lockfileVersion": 3, "requires": True, "packages": {"": {"name": "fixture", "version": "1.0.0",
|
|
"dependencies": {"@agent-system/contracts": spec}}, "node_modules/@agent-system/contracts": {
|
|
"version": "1.0.0", "resolved": spec}}}))
|
|
result = self.run_phase("Prefetch npm packages without lifecycle scripts")
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertFalse(marker.exists(), "candidate lifecycle ran while credential config was present")
|
|
consumer = subprocess.run(["node", "-e", "if(require('@agent-system/contracts').value !== 42)process.exit(1)"],
|
|
cwd=self.checkout, env=self.env, capture_output=True, text=True)
|
|
self.assertEqual(consumer.returncode, 0, consumer.stderr)
|
|
self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr)
|
|
for path, auth in self.requests:
|
|
self.assertRegex(path, r"^/weichen/Agent-System-Core\.git/")
|
|
self.assertEqual(auth, "token " + SUB_TOKEN)
|
|
|
|
def test_npm_provider_prepare_is_rejected_without_execution(self):
|
|
marker = self.base / "prepare-ran"
|
|
sha = self.commit_fixture("Agent-System-Core", {
|
|
"package.json": json.dumps({"name": "@agent-system/contracts", "version": "1.0.0",
|
|
"files": ["package"], "scripts": {"prepare": f"touch '{marker}'"}}),
|
|
"package/index.js": "module.exports = {value: 42}\n"})
|
|
spec = "git+" + SSH + "Agent-System-Core.git#" + sha
|
|
(self.checkout / "package.json").write_text(json.dumps({"name": "fixture", "dependencies": {"contracts": spec}}))
|
|
(self.checkout / "package-lock.json").write_text(json.dumps({"lockfileVersion": 3,
|
|
"packages": {"node_modules/contracts": {"resolved": spec}}}))
|
|
result = self.run_phase("Prefetch npm packages without lifecycle scripts")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("without acquisition lifecycle", result.stderr)
|
|
self.assertFalse(marker.exists())
|
|
|
|
def test_external_npm_git_source_rejected_before_authenticated_request(self):
|
|
spec = self.url + "/untrusted/receiver.git#" + "a" * 40
|
|
(self.checkout / "package.json").write_text(json.dumps({"name": "fixture", "dependencies": {"evil": "git+" + spec}}))
|
|
(self.checkout / "package-lock.json").write_text(json.dumps({"lockfileVersion": 3,
|
|
"packages": {"node_modules/evil": {"resolved": "git+" + spec}}}))
|
|
result = self.run_phase("Prefetch npm packages without lifecycle scripts")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.requests)
|
|
|
|
def test_component_checkout_removes_tokens_before_candidate_executes(self):
|
|
for repo in ("Logging-System", "Error-Code-System", "Agent-System-Core", "Agent-Native", "Agent-Intent",
|
|
"Agent-Command", "Agent-Squad", "Agent-System-Backend", "Agent-System-UI",
|
|
"Agent-System-Frontend", "Agent-System-RDM"):
|
|
with self.subTest(repo=repo):
|
|
sha = self.publish(repo)
|
|
self.checkout = self.base / (repo + "-checkout")
|
|
self.checkout.mkdir()
|
|
result = self.run_phase("Checkout", workflow=fixture.component_workflow(repo),
|
|
extra={"TOKEN": ROOT_TOKEN, "GITHUB_SHA": sha, "GITHUB_REPOSITORY": "weichen/" + repo},
|
|
after="python3 probe.py")
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
|
|
def test_go_requirement_for_new_toolchain_fails_without_downloading_one(self):
|
|
(self.checkout / "go.mod").write_text("module fixture\n\ngo 99.0\n")
|
|
result = self.run_phase("Prefetch Go modules without executing candidate code", extra={"GOTOOLCHAIN": "auto"})
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.requests)
|
|
self.assertIn("GOTOOLCHAIN=local", result.stderr)
|
|
|
|
def test_npm_nonregistry_tarball_rejected_before_request(self):
|
|
spec = self.url + "/untrusted/payload.tgz"
|
|
(self.checkout / "package.json").write_text(json.dumps({"name": "fixture", "dependencies": {"evil": spec}}))
|
|
(self.checkout / "package-lock.json").write_text(json.dumps({"lockfileVersion": 3,
|
|
"packages": {"node_modules/evil": {"resolved": spec}}}))
|
|
result = self.run_phase("Prefetch npm packages without lifecycle scripts")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.requests, "candidate tarball source was contacted")
|
|
|
|
def test_component_ci_executes_nested_go_module_tests(self):
|
|
(self.checkout / "go.mod").write_text("module fixture\n\ngo 1.25\n")
|
|
(self.checkout / "main.go").write_text("package fixture\nconst Value = 42\n")
|
|
(self.checkout / ".ci-mirrors.env").write_text("export GOTOOLCHAIN=local GOWORK=off GOENV=off\n")
|
|
for repo, suffix in (("Logging-System", "examples/trace"), ("Agent-Squad", "adapters/native")):
|
|
with self.subTest(repo=repo):
|
|
directory = self.checkout / suffix
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(directory / "go.mod").write_text("module nestedfixture\n\ngo 1.25\n")
|
|
(directory / "regression_test.go").write_text(
|
|
'package nestedfixture\nimport "testing"\nfunc TestNested(t *testing.T) {t.Fatal("intentional nested regression")}\n')
|
|
lines = fixture.component_workflow(repo).read_text().splitlines()
|
|
command = next(i for i, line in enumerate(lines) if "go vet" in line)
|
|
start = max(i for i in range(command) if lines[i].strip().endswith("run: |")) + 1
|
|
end = next((i for i in range(start, len(lines))
|
|
if lines[i].strip() and not lines[i].startswith(" ")), len(lines))
|
|
script = textwrap.dedent("\n".join(lines[start:end]))
|
|
# Component CI's race gate uses the native runner C compiler;
|
|
# formal product builds independently set CGO_ENABLED=0.
|
|
result = subprocess.run(["bash", "-c", script], cwd=self.checkout, env=self.env | {"CGO_ENABLED": "1"},
|
|
capture_output=True, text=True, timeout=60)
|
|
self.assertNotEqual(result.returncode, 0, "nested regression was skipped by component CI")
|
|
self.assertIn("intentional nested regression", result.stdout + result.stderr)
|
|
|
|
def test_go_prefetch_covers_logging_trace_module(self):
|
|
sha = self.commit_fixture("Logging-System", {
|
|
"go.mod": f"module {MODULE}\n\ngo 1.25\n", "logging.go": "package logging\nconst Value = 42\n"})
|
|
(self.checkout / "go.mod").write_text("module fixture\n\ngo 1.25\n")
|
|
example = self.checkout / "examples/trace"
|
|
example.mkdir(parents=True)
|
|
(example / "go.mod").write_text(
|
|
f"module tracefixture\n\ngo 1.27\n\nrequire {MODULE} v0.0.0-20260102030405-{sha[:12]}\n")
|
|
(example / "go.sum").write_text(self.fixture_checksums(MODULE, sha))
|
|
before = (example / "go.sum").read_bytes()
|
|
result = self.run_phase("Prefetch Go modules without executing candidate code",
|
|
workflow=fixture.component_workflow("Logging-System"))
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertTrue(self.requests, "trace module's private requirement was not prefetched")
|
|
self.assertEqual((example / "go.sum").read_bytes(), before, "Go 1.27 prefetch pruned source checksums")
|
|
|
|
def test_go_prefetch_rejects_mutation_of_committed_checksums(self):
|
|
sha = self.commit_fixture("Logging-System", {
|
|
"go.mod": f"module {MODULE}\n\ngo 1.25\n", "logging.go": "package logging\nconst Value = 42\n"})
|
|
(self.checkout / "go.mod").write_text(
|
|
f"module fixture\n\ngo 1.25\n\nrequire {MODULE} v0.0.0-20260102030405-{sha[:12]}\n")
|
|
(self.checkout / "go.sum").write_text("")
|
|
result = self.run_phase("Prefetch Go modules without executing candidate code")
|
|
self.assertNotEqual(result.returncode, 0, "private acquisition silently rewrote committed checksums")
|
|
self.assertTrue("modified a committed Go dependency file" in result.stderr or "missing go.sum" in result.stderr,
|
|
result.stderr)
|
|
|
|
def test_rdm_composition_lock_fetches_fixed_dev_without_executing_code(self):
|
|
dev = self.commit_fixture("Agent-System-Dev", {"should-not-run.py": "raise RuntimeError('candidate executed')\n"})
|
|
(self.checkout / "composition.dev.json").write_text(json.dumps({
|
|
"schema": 1, "repo": SSH + "Agent-System-Dev.git", "commit": dev}))
|
|
result = self.run_phase("Prefetch fixed Dev product composition",
|
|
workflow=fixture.component_workflow("Agent-System-RDM"))
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.git(self.checkout / ".ci-composition-dev", "rev-parse", "HEAD"), dev)
|
|
for path, auth in self.requests:
|
|
self.assertRegex(path, r"^/weichen/Agent-System-Dev\.git/")
|
|
self.assertEqual(auth, "token " + SUB_TOKEN)
|
|
self.assertNotIn(SUB_TOKEN, result.stdout + result.stderr)
|
|
|
|
def test_rdm_composition_lock_rejects_external_missing_and_malformed_commits(self):
|
|
for lock in ({"schema": 1, "repo": self.url + "/untrusted/receiver", "commit": "a" * 40},
|
|
{"schema": 1, "repo": SSH + "Agent-System-Dev.git"},
|
|
{"schema": 1, "repo": SSH + "Agent-System-Dev.git", "commit": "main"}):
|
|
with self.subTest(lock=lock):
|
|
(self.checkout / "composition.dev.json").write_text(json.dumps(lock))
|
|
result = self.run_phase("Prefetch fixed Dev product composition",
|
|
workflow=fixture.component_workflow("Agent-System-RDM"))
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(self.requests)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|