463 lines
23 KiB
Python
463 lines
23 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 rootfs_fixture(self):
|
|
from hydra import image
|
|
import subprocess
|
|
(self.project / 'image').mkdir()
|
|
(self.project / 'image/packages').write_text('app-emulation/qemu\n')
|
|
(self.project / 'hydra').mkdir()
|
|
(self.project / 'hydra/runtime.py').write_text('# runtime revision two\n')
|
|
builder = self.builder()
|
|
artifacts = builder.work / 'artifacts'
|
|
artifacts.mkdir(parents=True)
|
|
inventory = artifacts / 'packages.json'
|
|
inventory.write_text('[{"package":"app-emulation/qemu-9.2.4","use":["spice"],"build_time":"123"}]')
|
|
raw = artifacts / 'builder-rootfs.tar'
|
|
raw.write_bytes(b'verified raw builder snapshot')
|
|
original_complete = builder.complete
|
|
builder.complete = lambda step: True if step == 'prepare' else original_complete(step)
|
|
builder.record('packages', inventory_sha256=image.digest(inventory))
|
|
builder.record('rootfs', source_inputs='prior-runtime-identity', snapshot_sha256=image.digest(raw),
|
|
artifacts={'packages.json': image.digest(inventory)})
|
|
commands = []
|
|
def run(argv, capture=False, check=True):
|
|
if argv[0] == 'git':
|
|
return subprocess.CompletedProcess(argv, 0, 'source-commit\n', '')
|
|
args = argv[7:]
|
|
commands.append(args)
|
|
output = ''
|
|
if args[:2] == ['container', 'exists']:
|
|
return subprocess.CompletedProcess(argv, 0, '', '')
|
|
if args[:2] == ['container', 'inspect']:
|
|
output = json.dumps([{'Image': 'trusted-assembler-image', 'Mounts': [
|
|
{'Destination': '/hydra-source', 'Source': str(builder.source_dir), 'RW': False},
|
|
{'Destination': '/hydra-output', 'Source': str(builder.work), 'RW': True}]}])
|
|
elif args[:2] == ['image', 'inspect']:
|
|
output = json.dumps([{'RootFS': {'Layers': ['sha256:' + image.digest(raw)]}}])
|
|
elif args[0] == 'export':
|
|
raw.write_bytes(b'fresh raw builder snapshot')
|
|
elif args[0] == 'exec':
|
|
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
|
|
(artifacts / name).write_bytes(b'reassembled ' + name.encode())
|
|
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({
|
|
'kernel': '6.18-test', 'packages': json.loads(inventory.read_text()),
|
|
'source_files': image.source_files(builder.source_dir)}))
|
|
return subprocess.CompletedProcess(argv, 0, output, '')
|
|
builder.run = run # Fake only the external Podman/git boundary.
|
|
return builder, raw, inventory, commands
|
|
|
|
def test_runtime_change_reuses_verified_raw_export_and_reassembles(self):
|
|
from hydra import image
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
original_bytes = raw.read_bytes()
|
|
builder.rootfs()
|
|
self.assertFalse(any(args[0] == 'export' for args in commands))
|
|
self.assertEqual(raw.read_bytes(), original_bytes)
|
|
self.assertEqual(raw.stat().st_mode & 0o777, 0o600)
|
|
self.assertTrue(any(args[0] == 'exec' for args in commands), 'raw reuse must still run assembly')
|
|
self.assertEqual((inventory.parent / 'squashfs.img').read_bytes(), b'reassembled squashfs.img')
|
|
self.assertEqual(builder.manifest()['rootfs']['snapshot_sha256'], image.digest(raw))
|
|
self.assertTrue(builder.complete('rootfs'))
|
|
|
|
def test_corrupt_raw_export_falls_back_to_fresh_export(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
raw.write_bytes(b'corrupted raw cache')
|
|
builder.rootfs()
|
|
self.assertTrue(any(args[0] == 'export' for args in commands))
|
|
self.assertEqual(raw.read_bytes(), b'fresh raw builder snapshot')
|
|
|
|
def test_changed_image_inputs_require_fresh_raw_export(self):
|
|
from hydra import image
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
(self.project / 'image/packages').write_text('app-emulation/qemu\nsys-fs/btrfs-progs\n')
|
|
builder.record('packages', inventory_sha256=image.digest(inventory))
|
|
builder.rootfs()
|
|
self.assertTrue(any(args[0] == 'export' for args in commands))
|
|
|
|
def test_changed_validated_package_inventory_requires_fresh_export(self):
|
|
from hydra import image
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
inventory.write_text('[{"package":"app-emulation/qemu-10.0.0","use":["spice"],"build_time":"124"}]')
|
|
builder.record('packages', inventory_sha256=image.digest(inventory))
|
|
builder.rootfs()
|
|
self.assertTrue(any(args[0] == 'export' for args in commands))
|
|
|
|
def test_incomplete_raw_export_checkpoint_requires_fresh_export(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
builder.record('rootfs', complete=False)
|
|
builder.rootfs()
|
|
self.assertTrue(any(args[0] == 'export' for args in commands))
|
|
|
|
def test_missing_raw_export_falls_back_to_fresh_export(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
raw.unlink()
|
|
builder.rootfs()
|
|
self.assertTrue(any(args[0] == 'export' for args in commands))
|
|
self.assertEqual(raw.read_bytes(), b'fresh raw builder snapshot')
|
|
|
|
def test_invalid_package_inventory_blocks_even_matching_raw_cache(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
original_bytes = raw.read_bytes()
|
|
inventory.write_text('[]')
|
|
with self.assertRaisesRegex(self.module.BuildError, 'completed packages'):
|
|
builder.rootfs()
|
|
self.assertEqual(raw.read_bytes(), original_bytes)
|
|
self.assertEqual(commands, [])
|
|
|
|
def test_raw_export_parent_escape_is_rejected_before_podman_writes(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
outside = self.project / 'outside-artifacts'
|
|
inventory.parent.rename(outside)
|
|
(builder.work / 'artifacts').symlink_to(outside, target_is_directory=True)
|
|
with self.assertRaisesRegex(self.module.BuildError, 'dedicated work'):
|
|
builder.rootfs()
|
|
self.assertEqual((outside / raw.name).read_bytes(), b'verified raw builder snapshot')
|
|
self.assertEqual(commands, [])
|
|
|
|
def test_hardlinked_raw_export_is_rejected_before_podman_writes(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
linked = self.project / 'outside-private'
|
|
os.link(raw, linked)
|
|
with self.assertRaisesRegex(self.module.BuildError, 'singly linked'):
|
|
builder.rootfs()
|
|
self.assertEqual(linked.read_bytes(), b'verified raw builder snapshot')
|
|
self.assertEqual(commands, [])
|
|
|
|
def test_raw_export_symlink_is_rejected_before_podman_writes(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
outside = self.project / 'outside-private'
|
|
outside.write_bytes(b'private sentinel')
|
|
raw.unlink()
|
|
raw.symlink_to(outside)
|
|
with self.assertRaisesRegex(self.module.BuildError, 'dedicated work'):
|
|
builder.rootfs()
|
|
self.assertEqual(outside.read_bytes(), b'private sentinel')
|
|
self.assertEqual(commands, [])
|
|
|
|
def test_raw_export_directory_is_rejected_before_podman_writes(self):
|
|
builder, raw, inventory, commands = self.rootfs_fixture()
|
|
raw.unlink()
|
|
raw.mkdir()
|
|
with self.assertRaisesRegex(self.module.BuildError, 'regular file'):
|
|
builder.rootfs()
|
|
self.assertEqual(commands, [])
|
|
|
|
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()
|