321 lines
16 KiB
Python
321 lines
16 KiB
Python
import hashlib
|
|
import contextlib
|
|
import io
|
|
import importlib.util
|
|
import json
|
|
import lzma
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
|
|
class BuilderTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.project = Path(self.tmp.name)
|
|
spec = importlib.util.spec_from_file_location('hydra_build', Path(__file__).parents[1] / 'hydra/build.py')
|
|
self.assertIsNotNone(spec, 'builder module is required')
|
|
self.assertTrue(Path(spec.origin).is_file(), 'builder implementation is missing')
|
|
self.module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.module)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def builder(self):
|
|
return self.module.Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
|
|
|
|
def test_bad_cached_hash_is_rejected(self):
|
|
artifact = self.project / 'archive'
|
|
artifact.write_bytes(b'tampered')
|
|
with self.assertRaisesRegex(self.module.BuildError, 'SHA256'):
|
|
self.module.verify_hash(artifact, hashlib.sha256(b'authentic').hexdigest())
|
|
|
|
def test_wrong_primary_signature_is_rejected(self):
|
|
status = '[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 DEADBEEF\n'
|
|
with self.assertRaisesRegex(self.module.BuildError, 'fingerprint'):
|
|
self.module.require_signature(status, '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910')
|
|
|
|
def test_signature_accepts_only_successful_matching_primary(self):
|
|
fingerprint = '13EBBDBEDE7A12775DFDB1BABB572E0E2D182910'
|
|
self.module.require_signature('[GNUPG:] VALIDSIG ABC 2026-10-13 123 0 4 0 1 10 00 ' + fingerprint, fingerprint)
|
|
with self.assertRaises(self.module.BuildError):
|
|
self.module.require_signature('[GNUPG:] BADSIG ABC test', fingerprint)
|
|
|
|
def test_refuses_root_and_outside_project(self):
|
|
with patch.object(self.module.os, 'geteuid', return_value=0):
|
|
with self.assertRaisesRegex(self.module.BuildError, 'root'):
|
|
self.builder()
|
|
with patch.object(self.module.os, 'geteuid', return_value=1000):
|
|
with self.assertRaisesRegex(self.module.BuildError, 'project'):
|
|
self.module.Builder(self.project, self.project.parent / 'unsafe', self.project / 'cache')
|
|
|
|
def test_podman_store_is_scoped_to_work(self):
|
|
argv = self.builder().podman_argv('container', 'exists', 'builder')
|
|
self.assertEqual(argv[0], 'podman')
|
|
self.assertEqual(argv[argv.index('--root') + 1], str(self.project / 'build/podman'))
|
|
self.assertIn('--runroot', argv)
|
|
self.assertEqual(argv[-3:], ['container', 'exists', 'builder'])
|
|
|
|
def test_resume_requires_actual_inventory(self):
|
|
builder = self.builder()
|
|
builder.work.mkdir()
|
|
(builder.work / 'manifest.json').write_text(json.dumps({'packages': {'complete': True}}))
|
|
self.assertFalse(builder.complete('packages'))
|
|
|
|
def test_symlink_escape_is_rejected(self):
|
|
(self.project / 'escape').symlink_to(self.project.parent, target_is_directory=True)
|
|
with self.assertRaisesRegex(self.module.BuildError, 'project'):
|
|
self.module.Builder(self.project, self.project / 'escape/unsafe', self.project / 'cache')
|
|
|
|
def test_changed_configuration_invalidates_fetch_checkpoint(self):
|
|
image = self.project / 'image'
|
|
image.mkdir()
|
|
config = image / 'sources.json'
|
|
config.write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
|
|
builder = self.builder()
|
|
with builder.locked():
|
|
(builder.cache / 'stage.tar').write_bytes(b'good')
|
|
(builder.cache / 'repo.tar').write_bytes(b'repo')
|
|
builder.record('fetch')
|
|
self.assertTrue(builder.complete('fetch'))
|
|
config.write_text('{"changed":true}')
|
|
self.assertFalse(builder.complete('fetch'))
|
|
|
|
def test_missing_fetch_archive_invalidates_checkpoint(self):
|
|
image = self.project / 'image'
|
|
image.mkdir()
|
|
(image / 'sources.json').write_text(json.dumps({'stage3': {'url': 'https://example.test/stage.tar', 'sha256': hashlib.sha256(b'good').hexdigest()}, 'portage': {'url': 'https://example.test/repo.tar', 'sha256': hashlib.sha256(b'repo').hexdigest()}}))
|
|
builder = self.builder()
|
|
with builder.locked():
|
|
builder.record('fetch')
|
|
self.assertFalse(builder.complete('fetch'))
|
|
|
|
def test_second_builder_cannot_share_active_work(self):
|
|
first, second = self.builder(), self.builder()
|
|
with first.locked():
|
|
with self.assertRaisesRegex(self.module.BuildError, 'running'):
|
|
with second.locked():
|
|
self.fail('overlapping build was allowed')
|
|
|
|
def test_container_uses_host_proxy_network_and_readonly_inputs(self):
|
|
builder = self.builder()
|
|
args = builder.container_run_args()
|
|
self.assertEqual(args[args.index('--network') + 1], 'host')
|
|
self.assertIn(str(self.project) + ':/hydra-source:ro', args)
|
|
self.assertIn(str(self.project / 'cache') + ':/hydra-cache:ro', args)
|
|
self.assertNotIn('--privileged', args)
|
|
|
|
def test_distinct_workspaces_have_distinct_podman_runtime_storage(self):
|
|
other = self.module.Builder(self.project, self.project / 'other-build', self.project / 'cache')
|
|
self.assertNotEqual(self.builder().runroot, other.runroot)
|
|
|
|
def existing_builder(self, source=None, cache=None, layer=None, snapshot='old-snapshot'):
|
|
"""A real executable models only the external Podman boundary."""
|
|
source = source or self.project
|
|
cache = cache or self.project / 'cache'
|
|
image = self.project / 'image'
|
|
image.mkdir(exist_ok=True)
|
|
inputs = {'stage3': {'url': 'https://example.test/stage.tar.xz',
|
|
'sha256': hashlib.sha256(lzma.compress(b'original layer')).hexdigest()},
|
|
'portage': {'filename': 'repo.tar.gz', 'commit': 'old-snapshot'}}
|
|
(image / 'sources.json').write_text(json.dumps(inputs))
|
|
builder = self.builder()
|
|
(builder.work / 'podman').mkdir(parents=True)
|
|
fixture = self.project / 'podman-fixture.json'
|
|
fixture.write_text(json.dumps({'Image': 'original-image-id',
|
|
'Mounts': [{'Source': str(source), 'Destination': '/hydra-source', 'RW': False},
|
|
{'Source': str(cache), 'Destination': '/hydra-cache', 'RW': False}],
|
|
'RootFS': {'Layers': [layer or 'sha256:' + hashlib.sha256(b'original layer').hexdigest()]},
|
|
'snapshot': snapshot}))
|
|
executable = self.project / 'fixture-podman'
|
|
executable.write_text('''#!/usr/bin/env python3
|
|
import json, pathlib, sys
|
|
root = pathlib.Path(__file__).parent
|
|
data = json.loads((root / 'podman-fixture.json').read_text())
|
|
args = sys.argv[7:]
|
|
if 'exists' in args:
|
|
sys.exit(0)
|
|
if 'inspect' in args:
|
|
print(json.dumps([data]))
|
|
elif args[0] == 'top':
|
|
print('HPID ARGS')
|
|
if data.get('active_job'):
|
|
print('123 /usr/bin/python3 /usr/lib/python-exec/python3/emerge --update package')
|
|
elif args[0] == 'cp':
|
|
snapshot = data.get('snapshot_pending') if args[1].endswith('.hydra-snapshot-input') else data['snapshot']
|
|
if snapshot is None:
|
|
sys.exit(1)
|
|
pathlib.Path(args[-1]).write_text(snapshot + '\\n')
|
|
elif args[0] == 'exec' and data.get('stopped'):
|
|
sys.exit(125)
|
|
elif args[0] == 'exec' and 'packages.lock' in args[-1]:
|
|
sys.exit(75 if data.get('job_lock') else 0)
|
|
elif args[0] == 'exec' and args[-1].startswith('cat '):
|
|
snapshot = data['snapshot'] or (data.get('snapshot_pending') if '.hydra-snapshot-input' in args[-1] else None)
|
|
if snapshot is None:
|
|
sys.exit(1)
|
|
print(snapshot)
|
|
else:
|
|
if args[0] == 'start':
|
|
data['stopped'] = False
|
|
(root / 'podman-fixture.json').write_text(json.dumps(data))
|
|
with (root / 'mutations.jsonl').open('a') as stream:
|
|
stream.write(json.dumps(args) + '\\n')
|
|
''')
|
|
executable.chmod(0o755)
|
|
builder.podman = str(executable)
|
|
with builder.locked():
|
|
(builder.cache / 'stage.tar.xz').write_bytes(lzma.compress(b'original layer'))
|
|
return builder, inputs
|
|
|
|
def assert_prepare_refused(self, builder, inputs, reason):
|
|
with patch.object(builder, 'fetch', return_value=inputs):
|
|
with self.assertRaisesRegex(self.module.BuildError, reason):
|
|
builder.prepare()
|
|
self.assertFalse((self.project / 'mutations.jsonl').exists(), 'refusal must precede container/config mutations')
|
|
self.assertNotIn('prepare', builder.manifest(), 'refusal must not record false provenance')
|
|
|
|
def test_other_worktree_mount_is_refused_before_preparation(self):
|
|
builder, inputs = self.existing_builder(source=self.project / 'other-worktree')
|
|
self.assert_prepare_refused(builder, inputs, 'source')
|
|
|
|
def test_other_cache_mount_is_refused_before_preparation(self):
|
|
builder, inputs = self.existing_builder(cache=self.project / 'other-cache')
|
|
self.assert_prepare_refused(builder, inputs, 'cache')
|
|
|
|
def test_changed_stage3_pin_refuses_old_container(self):
|
|
builder, inputs = self.existing_builder()
|
|
archive = lzma.compress(b'new stage layer')
|
|
(builder.cache / 'stage.tar.xz').write_bytes(archive)
|
|
inputs['stage3']['sha256'] = hashlib.sha256(archive).hexdigest()
|
|
self.assert_prepare_refused(builder, inputs, 'stage3')
|
|
|
|
def test_changed_snapshot_pin_refuses_existing_tree(self):
|
|
builder, inputs = self.existing_builder()
|
|
inputs['portage']['commit'] = 'new-snapshot'
|
|
self.assert_prepare_refused(builder, inputs, 'snapshot')
|
|
|
|
def test_matching_existing_builder_retains_compatible_provenance(self):
|
|
builder, inputs = self.existing_builder()
|
|
with patch.object(builder, 'fetch', return_value=inputs):
|
|
builder.prepare()
|
|
self.assertEqual(builder.manifest()['prepare'].get('stage3_sha256'), inputs['stage3']['sha256'])
|
|
self.assertEqual(builder.manifest()['prepare']['snapshot'], 'old-snapshot')
|
|
|
|
def test_status_invalidates_missing_package_inventory(self):
|
|
builder = self.builder()
|
|
with builder.locked():
|
|
builder.record('packages')
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
builder.status()
|
|
self.assertFalse(json.loads(output.getvalue())['packages']['complete'])
|
|
|
|
def test_status_cli_runs_during_active_build_without_mutations(self):
|
|
builder = self.builder()
|
|
self.module.subprocess.run(['git', 'init', '--quiet', str(self.project)], check=True)
|
|
output = io.StringIO()
|
|
with builder.locked():
|
|
builder.record('packages')
|
|
manifest_before = (builder.work / 'manifest.json').read_bytes()
|
|
with patch.object(self.module, '__file__', str(self.project / 'hydra/build.py')):
|
|
with contextlib.redirect_stdout(output):
|
|
result = self.module.main(['status'])
|
|
self.assertEqual(result, 0, 'status must not acquire the exclusive build lock')
|
|
data = json.loads(output.getvalue())
|
|
self.assertTrue(data['running'])
|
|
self.assertFalse(data['packages']['complete'])
|
|
self.assertEqual((builder.work / 'manifest.json').read_bytes(), manifest_before)
|
|
|
|
def test_detached_emerge_refuses_preparation_before_fetch_or_writes(self):
|
|
builder, inputs = self.existing_builder()
|
|
fixture = self.project / 'podman-fixture.json'
|
|
data = json.loads(fixture.read_text())
|
|
data['active_job'] = True
|
|
fixture.write_text(json.dumps(data))
|
|
with patch.object(builder, 'fetch', return_value=inputs) as fetch:
|
|
with self.assertRaisesRegex(self.module.BuildError, 'package job'):
|
|
builder.prepare()
|
|
fetch.assert_not_called()
|
|
self.assertFalse((self.project / 'mutations.jsonl').exists())
|
|
|
|
def test_contained_job_lock_refuses_preparation_before_fetch(self):
|
|
builder, inputs = self.existing_builder()
|
|
fixture = self.project / 'podman-fixture.json'
|
|
data = json.loads(fixture.read_text())
|
|
data['job_lock'] = True
|
|
fixture.write_text(json.dumps(data))
|
|
with patch.object(builder, 'fetch', return_value=inputs) as fetch:
|
|
with self.assertRaisesRegex(self.module.BuildError, 'package job'):
|
|
builder.prepare()
|
|
fetch.assert_not_called()
|
|
|
|
def test_status_detects_detached_emerge_without_outer_lock(self):
|
|
builder, inputs = self.existing_builder()
|
|
fixture = self.project / 'podman-fixture.json'
|
|
data = json.loads(fixture.read_text())
|
|
data['active_job'] = True
|
|
fixture.write_text(json.dumps(data))
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
builder.status()
|
|
status = json.loads(output.getvalue())
|
|
self.assertTrue(status['running'])
|
|
self.assertTrue(status['package_job_running'])
|
|
self.assertFalse(status['mutation_lock_held'])
|
|
|
|
def test_package_job_records_failure_and_log_without_host_collection(self):
|
|
self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
|
|
script = self.module.package_job_script(['/bin/sh', '-c', 'echo diagnostic; exit 7'],
|
|
self.project / 'job', self.project / 'pkgdb', 'inputs-one')
|
|
result = self.module.subprocess.run(['python3', '-c', script])
|
|
self.assertEqual(result.returncode, 7)
|
|
status = json.loads((self.project / 'job/packages.status.json').read_text())
|
|
self.assertFalse(status['running'])
|
|
self.assertEqual(status['exit_code'], 7)
|
|
self.assertIn('diagnostic', (self.project / 'job/packages.log').read_text())
|
|
self.assertFalse((self.project / 'job/packages.json').exists())
|
|
|
|
def test_package_job_persists_inventory_after_success_without_host_collection(self):
|
|
self.assertTrue(hasattr(self.module, 'package_job_script'), 'durable job runner is required')
|
|
package = self.project / 'pkgdb/category/example-1.0'
|
|
package.mkdir(parents=True)
|
|
(package / 'USE').write_text('wayland opengl')
|
|
(package / 'BUILD_TIME').write_text('123')
|
|
script = self.module.package_job_script(['/bin/sh', '-c', 'exit 0'], self.project / 'job',
|
|
self.project / 'pkgdb', 'inputs-one')
|
|
result = self.module.subprocess.run(['python3', '-c', script])
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(json.loads((self.project / 'job/packages.json').read_text()),
|
|
[{'package': 'category/example-1.0', 'use': ['wayland', 'opengl'], 'build_time': '123'}])
|
|
self.assertEqual(json.loads((self.project / 'job/packages.status.json').read_text())['exit_code'], 0)
|
|
|
|
def test_matching_stopped_builder_checks_provenance_before_start(self):
|
|
builder, inputs = self.existing_builder()
|
|
fixture = self.project / 'podman-fixture.json'
|
|
data = json.loads(fixture.read_text())
|
|
data['stopped'] = True
|
|
fixture.write_text(json.dumps(data))
|
|
with patch.object(builder, 'fetch', return_value=inputs):
|
|
try:
|
|
builder.prepare()
|
|
except self.module.BuildError as error:
|
|
self.fail(f'Compatible stopped builder must resume: {error}')
|
|
self.assertTrue(builder.manifest()['prepare']['complete'])
|
|
|
|
def test_unfinished_snapshot_is_not_reported_as_prepared(self):
|
|
builder, inputs = self.existing_builder(snapshot=None)
|
|
fixture = self.project / 'podman-fixture.json'
|
|
data = json.loads(fixture.read_text())
|
|
data['snapshot_pending'] = 'old-snapshot'
|
|
fixture.write_text(json.dumps(data))
|
|
builder.record('prepare')
|
|
self.assertFalse(builder.complete('prepare'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|