"""Disposable artifact tests: exercise byte validation, never privileged runtime."""
import ast
import importlib.util
import io
import json
import os
from pathlib import Path
import shutil
import struct
import subprocess
import sys
import tarfile
import tempfile
import unittest

SCRIPTS = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(SCRIPTS))
import release_payload as payload

VERSION = '2.0.1-alpha'
COMMIT = 'a' * 40


class PayloadTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.source = self.root / 'source'
        self.dist = self.root / 'dist'
        for name, raw in {
            'scripts/d09_handoff.py': b'handoff',
            'scripts/c01_network_owner.py': b'network',
            'scripts/host_future_capability.py': b'new typed helper',
            'image-recipe/configs/archipelago-c01-network.service': b'unit',
            'neode-ui/public/packages/archipelago-companion.apk': b'qualified APK bytes',
            'neode-ui/public/packages/archipelago-companion.json': b'{"versionCode":58,"versionName":"0.5.38-debug"}',
        }.items():
            path = self.source / name
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(raw)
        payload.stage(self.source, self.dist, VERSION, COMMIT)
        for name, raw in {
            'index.html': b'dashboard', 'aiui/index.html': b'AIUI',
            'aiui/BUILD-INFO': f'commit={COMMIT}\nbase_path=/aiui/\n'.encode(),
            **{'packages/' + p.name: p.read_bytes() for p in (self.source / 'neode-ui/public/packages').iterdir()},
        }.items():
            path = self.dist / name
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_bytes(raw)

    def archive(self, omit=None, duplicate=None, prefix='', link=None):
        path = self.root / 'frontend.tar.gz'
        with tarfile.open(path, 'w:gz') as tar:
            for source in sorted(self.dist.rglob('*')):
                name = source.relative_to(self.dist).as_posix()
                if name == omit:
                    continue
                info = tar.gettarinfo(str(source), prefix + name)
                info.mode = 0o755 if source.is_dir() else 0o644
                if name == link:
                    info.type = tarfile.SYMTYPE
                    info.linkname = '/etc/passwd'
                    info.size = 0
                tar.addfile(info, io.BytesIO(source.read_bytes()) if info.isreg() else None)
                if name == duplicate:
                    tar.addfile(info, io.BytesIO(source.read_bytes()))
        return path

    def check(self, **kwargs):
        payload.check_archive(self.source, self.archive(**kwargs), VERSION, COMMIT)

    def test_complete_flat_archive(self):
        self.check()

    def test_staging_filters_private_apps_without_changing_source_or_public_apps(self):
        for domain in ('apps', 'docker'):
            for name in ('kithmoot', 'V4V', 'node-demo-v4v', 'public-app'):
                original = self.source / domain / name / 'retained.txt'
                original.parent.mkdir(parents=True)
                original.write_text('source must remain')
                staged = self.dist / 'archipelago-runtime' / domain / name / 'retained.txt'
                staged.parent.mkdir(parents=True)
                staged.write_bytes(original.read_bytes())
        catalog = self.dist / 'catalog.json'
        catalog.write_text(json.dumps({'apps': [{'id': 'kithmoot'}, {'id': 'public-app'}],
                                       'storefront': {'promotions': [{'appId': 'v4v'}]}}))
        payload.stage(self.source, self.dist, VERSION, COMMIT)
        for domain in ('apps', 'docker'):
            for name in ('kithmoot', 'V4V', 'node-demo-v4v'):
                self.assertFalse((self.dist / 'archipelago-runtime' / domain / name).exists())
                self.assertEqual((self.source / domain / name / 'retained.txt').read_text(), 'source must remain')
            self.assertTrue((self.dist / 'archipelago-runtime' / domain / 'public-app/retained.txt').exists())
        self.assertEqual(json.loads(catalog.read_bytes()), {'apps': [{'id': 'public-app'}], 'storefront': {'promotions': []}})
        self.check()

    def test_new_helper_is_required_without_allowlist_update(self):
        (self.source / 'scripts/host_concurrent.py').write_text('new lane helper')
        with self.assertRaisesRegex(ValueError, 'missing payload'):
            self.check()

    def test_network_and_handoff_omissions_rejected(self):
        for name in ('c01_network_owner.py', 'd09_handoff.py', 'host_future_capability.py'):
            with self.subTest(name=name), self.assertRaisesRegex(ValueError, 'missing payload'):
                self.check(omit='archipelago-runtime/scripts/' + name)

    def test_stale_companion_metadata_and_apk_rejected(self):
        for name in ('archipelago-companion.apk', 'archipelago-companion.json'):
            path = self.dist / 'packages' / name
            original = path.read_bytes()
            path.write_bytes(b'old')
            with self.subTest(name=name), self.assertRaisesRegex(ValueError, 'source mismatch'):
                self.check()
            path.write_bytes(original)

    def test_duplicate_link_and_nested_layout_rejected(self):
        for kwargs in ({'duplicate': 'index.html'}, {'link': 'index.html'}, {'prefix': 'neode-ui/'}):
            with self.subTest(kwargs=kwargs), self.assertRaises(ValueError):
                self.check(**kwargs)

    def test_false_hash_declaration_rejected(self):
        path = self.dist / 'archipelago-runtime/RELEASE-PAYLOAD.json'
        value = json.loads(path.read_text())
        value['version'] = '2.0.0-alpha'
        path.write_text(json.dumps(value))
        with self.assertRaisesRegex(ValueError, 'declaration mismatch'):
            self.check()

    def test_aiui_previous_commit_rejected(self):
        (self.dist / 'aiui/BUILD-INFO').write_text('commit=' + 'b' * 40 + '\nbase_path=/aiui/\n')
        with self.assertRaisesRegex(ValueError, 'AIUI'):
            self.check()

    def test_wrong_native_architecture_and_script_rejected(self):
        path = self.root / 'backend'
        header = bytearray(64)
        header[:6] = b'\x7fELF\x02\x01'
        struct.pack_into('<HH', header, 16, 3, 62)
        path.write_bytes(header)
        payload.check_native(path)
        struct.pack_into('<H', header, 18, 183)
        for raw in (header, b'#!/bin/sh\nexit 0\n'):
            path.write_bytes(raw)
            with self.assertRaises(ValueError):
                payload.check_native(path)

    def test_source_symlink_refused(self):
        (self.source / 'scripts/host_link.py').symlink_to('/etc/passwd')
        with self.assertRaisesRegex(ValueError, 'symlink'):
            payload.stage(self.source, self.dist, VERSION, COMMIT)


class NativePayloadTests(PayloadTests):
    def setUp(self):
        super().setUp()
        shell = self.source/'aiui/packages/app/src/NativeShell.vue'
        shell.parent.mkdir(parents=True); shell.write_text('native fixture')
        (self.source/'scripts/native-helper.js').write_text('helper fixture')
        payload.stage(self.source, self.dist, VERSION, COMMIT)
        shutil.rmtree(self.dist/'aiui')
        (self.dist/'assets').mkdir()
        chunk = self.dist/'assets/NativeShell-fixture.js'; chunk.write_bytes(b'native chunk')
        self.info = dict(schema=1,mode='native',source_commit=COMMIT,source_dirty=False,
                         dynamic_entry=True,native_chunk='assets/NativeShell-fixture.js',sha256=payload.digest(chunk.read_bytes()))
        self.write_info()

    def write_info(self):
        (self.dist/'NATIVE-AIUI.json').write_text(json.dumps(self.info))

    def test_aiui_previous_commit_rejected(self):
        self.info['source_commit']='b'*40; self.write_info()
        with self.assertRaisesRegex(ValueError,'native AIUI'): self.check()

    def test_native_chunk_tamper_and_dirty_source_refused(self):
        payload.check_ui(self.source,self.dist,COMMIT)
        for key,value in [('source_dirty',True),('dynamic_entry',False),('native_chunk','../escape')]:
            original=self.info[key]; self.info[key]=value; self.write_info()
            with self.assertRaises(ValueError): self.check()
            self.info[key]=original
        self.write_info()
        (self.dist/'assets/NativeShell-fixture.js').write_bytes(b'tampered')
        with self.assertRaisesRegex(ValueError,'native AIUI'): self.check()

    def test_native_rejects_standalone_bootstrap(self):
        (self.dist/'aiui').mkdir(); (self.dist/'aiui/index.html').write_text('stale standalone')
        with self.assertRaisesRegex(ValueError,'standalone AIUI'): self.check()


class OwnerPackageTests(unittest.TestCase):
    def test_provisioned_modules_include_local_import_dependencies(self):
        # Inspect the production copy list without importing its root owner.
        tree = ast.parse((SCRIPTS / 'provision-d09-lifecycle.py').read_text())
        modules = next(ast.literal_eval(node.value) for node in tree.body
                       if isinstance(node, ast.Assign)
                       and any(isinstance(target, ast.Name) and target.id == 'MODULES'
                               for target in node.targets))
        self.assertIn('c01_network_client.py', modules)
        for name in modules:
            if not name.endswith('.py'):
                continue
            for node in ast.walk(ast.parse((SCRIPTS / name).read_text())):
                imports = ([alias.name for alias in node.names] if isinstance(node, ast.Import)
                           else [node.module] if isinstance(node, ast.ImportFrom) and node.module
                           else [])
                for imported in imports:
                    dependency = imported.split('.')[0] + '.py'
                    if (SCRIPTS / dependency).is_file():
                        self.assertIn(dependency, modules, f'{name} imports uninstalled {dependency}')

    def test_complete_repository_runtime_inventory(self):
        files = payload.runtime_sources(SCRIPTS.parent)
        for name in ('scripts/c01_endpoints.py', 'scripts/c01_network_client.py',
                     'image-recipe/configs/archipelago-c01-network.service'):
            self.assertIn('archipelago-runtime/' + name, files)

    def test_reproducible_complete_owner_archive_and_overwrite_refusal(self):
        spec = importlib.util.spec_from_file_location('owner_package', SCRIPTS / 'package-d09-c01.py')
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        with tempfile.TemporaryDirectory() as temp:
            a, b = Path(temp) / 'a.tar', Path(temp) / 'b.tar'
            module.package(a)
            module.package(b)
            self.assertEqual(a.read_bytes(), b.read_bytes())
            with tarfile.open(a) as tar:
                hashes = json.load(tar.extractfile('D09-C01-SOURCE-SHA256.json'))
                for name in ('c01_network_owner.py', 'c01_network_client.py', 'c01_endpoints.py', 'd09_handoff.py'):
                    self.assertIn('usr/libexec/' + name, hashes)
                for name, expected in hashes.items():
                    self.assertEqual(payload.digest(tar.extractfile(name).read()), expected)
            with self.assertRaises(FileExistsError):
                module.package(a)


class SourceTests(unittest.TestCase):
    def test_tag_and_dirty_source_refusals(self):
        with tempfile.TemporaryDirectory() as temp:
            root = Path(temp)
            def git(*args):
                return subprocess.check_output(['git', '-C', str(root), *args], stderr=subprocess.DEVNULL, text=True).strip()
            git('init')
            (root / 'core/archipelago').mkdir(parents=True)
            (root / 'neode-ui').mkdir()
            (root / 'core/archipelago/Cargo.toml').write_text(f'[package]\nversion="{VERSION}"\n')
            (root / 'neode-ui/package.json').write_text(json.dumps({'version': VERSION}))
            git('add', '.')
            git('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'commit', '-m', 'fixture')
            git('tag', 'v' + VERSION)
            with self.assertRaisesRegex(ValueError, 'annotated'):
                payload.check_source(root, VERSION)
            git('tag', '-d', 'v' + VERSION)
            git('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'tag', '-a', 'v' + VERSION, '-m', 'fixture')
            self.assertEqual(payload.check_source(root, VERSION), git('rev-parse', 'HEAD'))
            (root / 'new-helper.py').write_text('unreviewed')
            with self.assertRaisesRegex(ValueError, 'clean reviewed'):
                payload.check_source(root, VERSION)


class ManifestIntegrationTests(unittest.TestCase):
    setUp = PayloadTests.setUp
    archive = PayloadTests.archive

    def test_manifest_from_disposable_annotated_source_and_real_archive(self):
        # Synthetic APK tools are explicit fixture programs, not production
        # bypass flags. The actual staged APK is verified separately by the lane.
        scripts = self.source / 'scripts'
        for name in ('create-release-manifest.sh', 'release_payload.py'):
            shutil.copyfile(SCRIPTS / name, scripts / name)
        for name in ('ota-crash-guard.sh', 'd09-lifecycle-client.py', 'd09-lifecycle-owner.py',
                     'provision-d09-lifecycle.py', 'd09_wire.py', 'd09_store.py', 'd09_worker.py', 'd09-wire-schema.json'):
            (scripts / name).write_text('fixture')
        for name in ('os_update_plan.py', 'd09_wire.py', 'd09-wire-schema.json'):
            shutil.copyfile(SCRIPTS / name, scripts / name)
        configs = self.source / 'image-recipe/configs'
        for name in ('prepare-terminal-workspace.sh', 'archipelago-d09-lifecycle.service',
                     '50-d09-lifecycle.conf', '60-terminal-workspace.conf'):
            (configs / name).write_text('fixture')
        (self.source / 'core/archipelago').mkdir(parents=True)
        (self.source / 'core/archipelago/Cargo.toml').write_text(f'[package]\nversion="{VERSION}"\n')
        (self.source / 'neode-ui/package.json').write_text(json.dumps({'version': VERSION}))
        (self.source / 'CHANGELOG.md').write_text(f'## v{VERSION} (2026-11-01)\n\n- Synthetic artifact fixture\n')
        def git(*args):
            return subprocess.check_output(['git', '-C', str(self.source), *args], stderr=subprocess.DEVNULL, text=True).strip()
        git('init'); git('add', '.')
        identity = ('-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid')
        git(*identity, 'commit', '-m', 'synthetic source')
        git(*identity, 'tag', '-a', 'v' + VERSION, '-m', 'fixture')
        commit = git('rev-parse', 'HEAD')
        payload.stage(self.source, self.dist, VERSION, commit)
        (self.dist / 'aiui/BUILD-INFO').write_text(f'commit={commit}\nbase_path=/aiui/\n')
        archive = self.archive()
        archive = archive.rename(self.root / f'archipelago-frontend-{VERSION}.tar.gz')
        aapt, signer = self.root / 'aapt', self.root / 'apksigner'
        aapt.write_text("#!/bin/sh\nprintf \"package: name='com.archipelago.app.debug' versionCode='58' versionName='0.5.38-debug'\\n\"\n")
        signer.write_text("#!/bin/sh\nprintf '%s\\n' 'Verified using v1 scheme (fixture): true' 'Verified using v2 scheme (fixture): true' 'Verified using v3 scheme (fixture): true' 'Signer #1 certificate SHA-256 digest: d622e07e7f474246e8e2bcee828ca9bf6c6cdf9434c065c58fe89bb4eec2664d'\n")
        aapt.chmod(0o755); signer.chmod(0o755)
        output = self.root / 'manifest.json'
        from test_os_update import candidate
        plan = self.root / 'os-update-plan.json'
        plan.write_text(json.dumps(candidate()))
        command = ['bash', str(scripts / 'create-release-manifest.sh'), '--version', VERSION,
                   '--backend', '/bin/true', '--frontend', str(archive), '--output', str(output),
                   '--os-plan', str(plan)]
        env = dict(os.environ, AAPT=str(aapt), APKSIGNER=str(signer), PYTHONDONTWRITEBYTECODE='1')
        result = subprocess.run(command, env=env, capture_output=True, text=True)
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        manifest = json.loads(output.read_text())
        self.assertEqual(manifest['source_commit'], commit)
        self.assertEqual(len(manifest['components']), 3)
        self.assertEqual(manifest['components'][1]['sha256'], payload.digest(archive.read_bytes()))
        saved = output.read_bytes()
        result = subprocess.run(command, env=env, capture_output=True, text=True)
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(output.read_bytes(), saved)

        public_output = self.root / 'public-manifest.json'
        public_command = command[:-2]
        public_command[public_command.index('--output') + 1] = str(public_output)
        public_command.append('--application-only')
        public_result = subprocess.run(public_command, env=env, capture_output=True, text=True)
        self.assertEqual(public_result.returncode, 0, public_result.stdout + public_result.stderr)
        public = json.loads(public_output.read_text())
        self.assertEqual(public['application_update'], {'version': 2, 'mode': 'signed-application-runtime'})
        self.assertNotIn('os_update', public)
        self.assertEqual(len(public['components']), 2)
        payload.validate_release_components(public)
        self.assertEqual(public['components'], manifest['components'][:2])
        mixed = dict(public, os_update=manifest['os_update'])
        with self.assertRaises(ValueError):
            payload.validate_release_components(mixed)
        boolean_policy = json.loads(json.dumps(public))
        boolean_policy['application_update']['version'] = True
        with self.assertRaises(ValueError):
            payload.validate_release_components(boolean_policy)
        missing_policy = dict(public)
        missing_policy.pop('application_update')
        with self.assertRaises(ValueError):
            payload.validate_release_components(missing_policy)


if __name__ == '__main__':
    unittest.main()