245 lines
12 KiB
Python
245 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Control-plane regressions with fake tools; never an engine/native test pass."""
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location("validation_tests", ROOT / "scripts/test-cloud-fixture-validation.py")
|
|
fixtures = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(fixtures)
|
|
|
|
FAKE_PODMAN = r'''#!/usr/bin/env python3
|
|
import json,os,pathlib,sys,subprocess
|
|
root=pathlib.Path(os.environ['FAKE_ROOT']); args=sys.argv[1:]
|
|
with (root/'calls').open('a') as f:f.write(json.dumps(args)+'\n')
|
|
if args[0]=='create':
|
|
compile=any(a.startswith('archy-cloud-compile-') for a in args)
|
|
cid=('c' if compile else 'd')*64
|
|
(root/cid).write_text(json.dumps(args))
|
|
if compile:
|
|
for a in args:
|
|
if a.endswith(':/artifacts/test-binary:rw'):
|
|
pathlib.Path(a[:-len(':/artifacts/test-binary:rw')]).write_bytes(b'synthetic artifact, never executed')
|
|
print(cid)
|
|
elif args[0]=='start':
|
|
if args[-1]=='d'*64:
|
|
create=json.loads((root/args[-1]).read_text())
|
|
canary=next(a.split('=',1)[1] for a in create if a.startswith('ARCHY_TEST_HOST_CANARY='))
|
|
assert pathlib.Path(canary).is_file(), 'Host-only canary must exist during execution'
|
|
assert not any(a.startswith(canary+':') for a in create), 'Canary must never be mounted'
|
|
if args[-1]=='c'*64 and os.environ.get('FAIL_COMPILE'):sys.exit(1)
|
|
if args[-1]=='d'*64 and os.environ.get('FAIL_NAMESPACE'):
|
|
create=json.loads((root/args[-1]).read_text()); pos=create.index('-ec')
|
|
# Real shell guard, with a deliberately wrong synthetic readlink result.
|
|
# It MUST exit before attempting the artifact; no backend runs on host.
|
|
script=create[pos+1]
|
|
assert 'Fixture namespace changed' in script
|
|
result=subprocess.run(['/bin/sh','-ec',script,'cloud-netns','net:[200]'],env=os.environ)
|
|
sys.exit(result.returncode)
|
|
elif args[:2]==['image','exists']:
|
|
if os.environ.get('FAIL_RUNTIME'): sys.exit(1)
|
|
elif args[0]=='inspect': print((root/'inspect.json').read_text())
|
|
elif args[0]=='rm': pass
|
|
else: sys.exit(99)
|
|
'''
|
|
|
|
|
|
class Runner(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.path = Path(self.temp.name)
|
|
(self.path / 'bin').mkdir()
|
|
(self.path / 'repo/core').mkdir(parents=True)
|
|
(self.path / 'cache/registry').mkdir(parents=True)
|
|
for name, content in {
|
|
'podman': FAKE_PODMAN,
|
|
'cargo': '#!/bin/sh\necho called > "$FAKE_ROOT/host-cargo"\nexit 99\n',
|
|
'sudo': '#!/bin/sh\nprintf "net:[%s]\\n" "${FAKE_EXPECTED:-200}"\n',
|
|
'readlink': '#!/bin/sh\necho "net:[100]"\n',
|
|
}.items():
|
|
script = self.path / 'bin' / name
|
|
script.write_text(content); script.chmod(0o755)
|
|
(self.path / 'inspect.json').write_text(json.dumps(fixtures.fixture()))
|
|
self.env = dict(os.environ, PATH=f'{self.path}/bin:' + os.environ['PATH'],
|
|
FAKE_ROOT=str(self.path), ARCHY_TEST_ISOLATOR='podman',
|
|
ARCHY_TEST_REPO=str(self.path / 'repo'), ARCHY_TEST_IMAGE='b' * 64,
|
|
ARCHY_TEST_CARGO_HOME=str(self.path / 'cache'),
|
|
ARCHY_TEST_CLOUD_FIXTURE_CONTAINER=fixtures.NAME,
|
|
ARCHY_TEST_CLOUD_FIXTURE_IMAGE=fixtures.IMAGE)
|
|
|
|
def run_wrapper(self):
|
|
return subprocess.run(['bash', str(ROOT / 'scripts/test-backend-isolated.sh'), '--ignored'],
|
|
env=self.env, capture_output=True, text=True, timeout=20)
|
|
|
|
def calls(self):
|
|
file = self.path / 'calls'
|
|
return [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []
|
|
|
|
def test_systemd_cloud_lane_refuses_before_host_cargo(self):
|
|
self.env['ARCHY_TEST_ISOLATOR'] = 'systemd'
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 2, result.stderr)
|
|
self.assertFalse((self.path / 'host-cargo').exists())
|
|
self.assertFalse(self.calls())
|
|
|
|
def test_live_worktree_and_mutable_builder_are_refused(self):
|
|
git = self.path / 'repo/.git'; git.write_text('gitdir: /synthetic')
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
git.unlink(); self.env['ARCHY_TEST_IMAGE'] = 'compiler:latest'
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(self.calls())
|
|
|
|
def test_mocked_success_preserves_compile_and_execution_boundaries(self):
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
creates = [c for c in self.calls() if c[0] == 'create']
|
|
self.assertEqual(len(creates), 2)
|
|
compile, execute = creates
|
|
for args in creates:
|
|
for option in ['--pull=never', '--http-proxy=false', '--read-only', '--cap-drop=all',
|
|
'--security-opt=no-new-privileges', '--log-driver=none', '--cpus=2']:
|
|
self.assertIn(option, args)
|
|
writable = [a for a in args if a.endswith(':rw')]
|
|
self.assertTrue(all(a.endswith(':/artifacts/test-binary:rw') for a in writable))
|
|
self.assertIn(str(self.path / 'repo') + ':/workspace:ro', args)
|
|
self.assertFalse(any('podman.sock' in a or 'docker.sock' in a for a in args))
|
|
self.assertIn('--network=none', compile)
|
|
self.assertIn('--memory=8g', compile)
|
|
self.assertIn('--memory-swap=8g', compile)
|
|
self.assertIn('--pids-limit=512', compile)
|
|
self.assertIn('--timeout=1800', compile)
|
|
self.assertIn('CARGO_NET_OFFLINE=true', compile)
|
|
self.assertIn('CARGO_INCREMENTAL=0', compile)
|
|
self.assertIn('profile.test.package.archipelago.codegen-units=16', compile[compile.index('-ec') + 1])
|
|
self.assertIn(str(self.path / 'cache/registry') + ':/cargo-home/registry:ro', compile)
|
|
self.assertIn('--network=container:' + fixtures.NAME, execute)
|
|
self.assertIn('--memory=3g', execute)
|
|
self.assertIn('--pids-limit=256', execute)
|
|
self.assertIn('--timeout=120', execute)
|
|
for path, size in [('/tmp', '128m'), ('/root', '16m'), ('/run', '16m'),
|
|
('/var/lib/archipelago', '64m'), ('/var/lib/containers', '16m')]:
|
|
self.assertIn(f'{path}:rw,notmpcopyup,size={size}', execute)
|
|
self.assertIn('net:[200]', execute)
|
|
canaries = [a.split('=', 1)[1] for a in execute if a.startswith('ARCHY_TEST_HOST_CANARY=')]
|
|
self.assertEqual(len(canaries), 1)
|
|
self.assertTrue(canaries[0].startswith('/tmp/archy-cloud-test.'))
|
|
self.assertFalse(Path(canaries[0]).exists(), 'Cleanup must remove the host canary')
|
|
self.assertIn('readlink /proc/self/ns/net', execute[execute.index('-ec') + 1])
|
|
self.assertFalse((self.path / 'host-cargo').exists())
|
|
self.assertEqual(len([c for c in self.calls() if c[0] == 'rm']), 2)
|
|
|
|
def test_artifact_export_retains_binary_and_binding_without_overwrite(self):
|
|
output = self.path / 'export'; output.mkdir()
|
|
self.env['ARCHY_TEST_ARTIFACT_EXPORT'] = str(output)
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
receipt = json.loads((output / 'artifact.json').read_text())
|
|
self.assertEqual(receipt['compiler_image'], 'b' * 64)
|
|
self.assertEqual(len(receipt['source_sha256']), 64)
|
|
self.assertEqual((output / 'test-binary').read_bytes(), b'synthetic artifact, never executed')
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
|
|
def compile_artifact(self):
|
|
output = self.path / 'export'; output.mkdir()
|
|
self.env['ARCHY_TEST_ARTIFACT_EXPORT'] = str(output)
|
|
self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '1'
|
|
fixture = self.env.pop('ARCHY_TEST_CLOUD_FIXTURE_CONTAINER')
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(any(c[0] == 'inspect' for c in self.calls()))
|
|
self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)
|
|
self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '0'
|
|
self.env['ARCHY_TEST_CLOUD_FIXTURE_CONTAINER'] = fixture
|
|
del self.env['ARCHY_TEST_ARTIFACT_EXPORT']
|
|
self.env['ARCHY_TEST_REUSE_ARTIFACT'] = str(output)
|
|
(self.path / 'calls').unlink()
|
|
return output
|
|
|
|
def test_compile_before_fixture_then_exact_artifact_reuse(self):
|
|
self.compile_artifact()
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
creates = [c for c in self.calls() if c[0] == 'create']
|
|
self.assertEqual(len(creates), 1)
|
|
self.assertTrue(any('archy-cloud-execute-' in a for a in creates[0]))
|
|
self.assertFalse((self.path / 'host-cargo').exists())
|
|
|
|
def test_reuse_rejects_source_change_before_namespace_inspection(self):
|
|
self.compile_artifact()
|
|
(self.path / 'repo/core/changed').write_text('new source')
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))
|
|
|
|
def test_reuse_rejects_changed_runtime_and_binary(self):
|
|
output = self.compile_artifact()
|
|
self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
del self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE']
|
|
binary = output / 'test-binary'; binary.chmod(0o700); binary.write_bytes(b'changed')
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))
|
|
|
|
def test_reuse_rejects_symlinked_artifact(self):
|
|
output = self.compile_artifact()
|
|
binary = output / 'test-binary'; saved = output / 'saved'; binary.rename(saved); binary.symlink_to(saved)
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))
|
|
|
|
def test_compile_only_requires_export_before_creating_container(self):
|
|
self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '1'
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(any(c[0] == 'create' for c in self.calls()))
|
|
|
|
def test_runtime_image_is_pinned_and_preflighted_before_compile(self):
|
|
self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'browser:latest'
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(self.calls())
|
|
self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
|
|
self.env['FAIL_RUNTIME'] = '1'
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertFalse(any(c[0] == 'create' for c in self.calls()))
|
|
|
|
def test_browser_runtime_does_not_change_compiler_or_isolation(self):
|
|
self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
compile, execute = [c for c in self.calls() if c[0] == 'create']
|
|
self.assertEqual(compile[compile.index('-ec') - 1], 'b' * 64)
|
|
self.assertEqual(execute[execute.index('-ec') - 1], 'e' * 64)
|
|
self.assertIn('--network=none', compile)
|
|
self.assertIn('--network=container:' + fixtures.NAME, execute)
|
|
self.assertIn('--cap-drop=all', execute)
|
|
|
|
def test_compile_failure_never_inspects_or_enters_fixture(self):
|
|
self.env['FAIL_COMPILE'] = '1'
|
|
self.assertNotEqual(self.run_wrapper().returncode, 0)
|
|
self.assertFalse(any(c[0] == 'inspect' for c in self.calls()))
|
|
self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)
|
|
|
|
def test_host_namespace_refused_before_execution_container(self):
|
|
self.env['FAKE_EXPECTED'] = '100'
|
|
self.assertEqual(self.run_wrapper().returncode, 2)
|
|
self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)
|
|
|
|
def test_namespace_reuse_fails_real_shell_guard_before_artifact(self):
|
|
self.env['FAIL_NAMESPACE'] = '1'
|
|
result = self.run_wrapper()
|
|
self.assertEqual(result.returncode, 2, result.stderr)
|
|
self.assertIn('Fixture namespace changed', result.stdout)
|
|
|
|
def test_log_sink_is_bounded(self):
|
|
result = subprocess.run(['python3', str(ROOT / 'scripts/limit-test-output.py')],
|
|
input=b'x' * (8 * 1024 * 1024 + 1), capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 125)
|
|
self.assertLessEqual(len(result.stdout), 8 * 1024 * 1024)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|