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()