import hashlib
import json
import os
from pathlib import Path
import py_compile
import runpy
import shutil
import struct
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest import mock
import venv


ROOT = Path(__file__).resolve().parents[1]
APP_DIR = ROOT / 'app'
sys.path.insert(0, str(APP_DIR))

import lifecycle_authority
import supervisor


def file_sha256(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def canonical(path):
    return os.path.normcase(os.path.realpath(os.path.abspath(os.fspath(path))))


def copy_bootstraps(app_dir):
    app_dir.mkdir()
    for name in ('child_bootstrap.py', 'runtime_bootstrap.py'):
        shutil.copy2(APP_DIR / name, app_dir / name)


def write_scanner_cache(app_dir, marker, invalidation_mode):
    scanner = app_dir / 'scanner.py'
    scanner.write_text('VALUE = "source"\n', encoding='ascii')
    payload = app_dir.parent / 'scanner_payload.py'
    payload.write_text(
        'import os\n'
        'with open(os.environ["PYC_PAYLOAD_MARKER"], "w", encoding="ascii") as handle:\n'
        '    handle.write("executed")\n',
        encoding='ascii',
    )
    cache = Path(importlib_cache_path(scanner))
    cache.parent.mkdir()
    py_compile.compile(
        str(payload),
        cfile=str(cache),
        dfile=str(scanner),
        doraise=True,
        invalidation_mode=invalidation_mode,
    )
    payload.unlink()
    if invalidation_mode == py_compile.PycInvalidationMode.TIMESTAMP:
        data = bytearray(cache.read_bytes())
        details = scanner.stat()
        data[8:16] = struct.pack('<II', int(details.st_mtime), details.st_size)
        cache.write_bytes(data)
    return cache


def importlib_cache_path(source):
    import importlib.util

    return importlib.util.cache_from_source(str(source))


def fixture_manifest(app_dir):
    files = {}
    for path in sorted(app_dir.rglob('*')):
        if not path.is_file():
            continue
        relative = path.relative_to(app_dir)
        in_cache = any(part.lower() == '__pycache__' for part in relative.parts[:-1])
        suffixes = ('.pyc',) if in_cache else ('.py', '.pyw', '.pyc', '.pyd')
        if path.suffix.lower() not in suffixes:
            continue
        name = relative.as_posix()
        files[name] = {'path': canonical(path), 'sha256': file_sha256(path)}
    return {
        'schema': lifecycle_authority.CODE_MANIFEST_SCHEMA,
        'root': canonical(app_dir),
        'files': files,
        'executables': {},
        'assets': {},
    }


class RuntimeBootstrapAuthorityTests(unittest.TestCase):
    def test_authenticated_discovery_producer_dispatches_console_runner(self):
        namespace = runpy.run_path(str(APP_DIR / 'child_bootstrap.py'))
        main = namespace['main']
        entrypoint = canonical(APP_DIR / 'console_runner.py')
        metadata = {'code_manifest': {'files': {'console': {'path': entrypoint}}}}
        captured = []
        authenticated = []
        dependencies = []

        def capture(path, run_name=None):
            captured.append((canonical(path), list(sys.argv), run_name))

        with mock.patch.object(sys, 'flags', SimpleNamespace(
            isolated=1, no_site=1, dont_write_bytecode=1,
        )), mock.patch.object(sys, 'path', list(sys.path)), mock.patch.object(
            sys, 'argv', [
                str(APP_DIR / 'child_bootstrap.py'), 'discovery-producer', '--',
                '--config', 'config.yaml', '--source', 'gitlab', '--once',
            ],
        ), mock.patch.dict(main.__globals__, {
            '_authenticate': lambda kind: authenticated.append(kind) or (canonical(APP_DIR), metadata),
            '_enable_dependency_paths': lambda kind: dependencies.append(kind),
        }), mock.patch.object(main.__globals__['runpy'], 'run_path', side_effect=capture):
            main()

        self.assertEqual(authenticated, ['discovery-producer'])
        self.assertEqual(dependencies, ['discovery-producer'])
        self.assertEqual(captured, [(
            entrypoint,
            [entrypoint, '--config', 'config.yaml', '--source', 'gitlab', '--once'],
            '__main__',
        )])

    def test_authenticated_docker_shadow_dispatches_exact_entrypoint(self):
        namespace = runpy.run_path(str(APP_DIR / 'child_bootstrap.py'))
        main = namespace['main']
        entrypoint = canonical(APP_DIR / 'docker_shadow.py')
        metadata = {'code_manifest': {'files': {'shadow': {'path': entrypoint}}}}
        captured = []
        authenticated = []

        def capture(path, run_name=None):
            captured.append((canonical(path), list(sys.argv), run_name))

        main_globals = main.__globals__
        with mock.patch.object(sys, 'flags', SimpleNamespace(
            isolated=1, no_site=1, dont_write_bytecode=1,
        )), mock.patch.object(sys, 'path', list(sys.path)), mock.patch.object(
            sys, 'argv', [
                str(APP_DIR / 'child_bootstrap.py'), 'docker-shadow', '--',
                '--config', 'config.yaml',
            ],
        ), mock.patch.dict(main_globals, {
            '_authenticate': lambda kind: authenticated.append(kind) or (canonical(APP_DIR), metadata),
            '_enable_dependency_paths': lambda kind: None,
        }), mock.patch.object(main_globals['runpy'], 'run_path', side_effect=capture):
            main()

        self.assertEqual(authenticated, ['docker-shadow'])
        self.assertEqual(captured, [(
            entrypoint, [entrypoint, '--config', 'config.yaml'], '__main__',
        )])

    def test_provider_bootstrap_consumes_one_required_separator_before_provider_parse(self):
        namespace = runpy.run_path(str(APP_DIR / 'child_bootstrap.py'))
        main = namespace['main']
        entrypoint = canonical(APP_DIR / 'keycheckers' / 'qwen' / 'qwenKeycheck.py')
        metadata = {'code_manifest': {'files': {'provider': {'path': entrypoint}}}}
        captured = []

        def capture(path, run_name=None):
            captured.append((canonical(path), list(sys.argv), run_name))

        main_globals = main.__globals__
        with mock.patch.object(sys, 'flags', SimpleNamespace(
            isolated=1, no_site=1, dont_write_bytecode=1,
        )), mock.patch.object(sys, 'path', list(sys.path)), mock.patch.object(
            sys, 'argv', [
                str(APP_DIR / 'child_bootstrap.py'), 'keycheck-provider',
                'keycheckers/qwen/qwenKeycheck.py', '--', '--input', 'managed.jsonl',
            ],
        ), mock.patch.dict(main_globals, {
            '_authenticate': lambda kind: (canonical(APP_DIR), metadata),
            '_enable_dependency_paths': lambda kind: None,
        }), mock.patch.object(main_globals['runpy'], 'run_path', side_effect=capture):
            main()

        self.assertEqual(captured, [(
            entrypoint, [entrypoint, '--input', 'managed.jsonl'], '__main__',
        )])

    def test_provider_bootstrap_rejects_missing_and_duplicate_separators(self):
        namespace = runpy.run_path(str(APP_DIR / 'child_bootstrap.py'))
        main = namespace['main']
        entrypoint = canonical(APP_DIR / 'keycheckers' / 'qwen' / 'qwenKeycheck.py')
        metadata = {'code_manifest': {'files': {'provider': {'path': entrypoint}}}}
        cases = (
            (['keycheckers/qwen/qwenKeycheck.py', '--input', 'managed.jsonl'], 'separator is required'),
            (['keycheckers/qwen/qwenKeycheck.py', '--', '--', '--input', 'managed.jsonl'], 'duplicate'),
        )
        for arguments, error in cases:
            with self.subTest(error=error), mock.patch.object(sys, 'flags', SimpleNamespace(
                isolated=1, no_site=1, dont_write_bytecode=1,
            )), mock.patch.object(sys, 'path', list(sys.path)), mock.patch.object(
                sys, 'argv', [str(APP_DIR / 'child_bootstrap.py'), 'keycheck-provider', *arguments],
            ), mock.patch.dict(main.__globals__, {
                '_authenticate': lambda kind: (canonical(APP_DIR), metadata),
                '_enable_dependency_paths': lambda kind: None,
            }), mock.patch.object(main.__globals__['runpy'], 'run_path') as provider:
                with self.assertRaisesRegex(RuntimeError, error):
                    main()
            provider.assert_not_called()

    def test_authenticated_entrypoint_exception_is_classified_as_runtime_failure(self):
        namespace = runpy.run_path(str(APP_DIR / 'child_bootstrap.py'))
        main = namespace['main']
        entrypoint = canonical(APP_DIR / 'console_runner.py')
        metadata = {'code_manifest': {'files': {'scanner': {'path': entrypoint}}}}
        main_globals = main.__globals__
        with mock.patch.object(sys, 'flags', SimpleNamespace(
            isolated=1, no_site=1, dont_write_bytecode=1,
        )), mock.patch.object(sys, 'path', list(sys.path)), mock.patch.object(
            sys, 'argv', [str(APP_DIR / 'child_bootstrap.py'), 'scanner', '--'],
        ), mock.patch.dict(main_globals, {
            '_authenticate': lambda kind: (canonical(APP_DIR), metadata),
            '_enable_dependency_paths': lambda kind: None,
        }), mock.patch.object(
            main_globals['runpy'], 'run_path', side_effect=RuntimeError('fixture database timeout'),
        ):
            with self.assertRaisesRegex(namespace['ChildRuntimeError'], 'fixture database timeout'):
                main()

    def test_runtime_rejects_timestamp_and_unchecked_scanner_cache_before_payload(self):
        modes = (
            py_compile.PycInvalidationMode.TIMESTAMP,
            py_compile.PycInvalidationMode.UNCHECKED_HASH,
        )
        for mode in modes:
            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as temp_dir:
                root = Path(temp_dir)
                app_dir = root / 'app'
                copy_bootstraps(app_dir)
                pyc_marker = root / 'pyc.marker'
                target_marker = root / 'target.marker'
                write_scanner_cache(app_dir, pyc_marker, mode)
                (app_dir / 'supervisor.py').write_text(
                    'import os\nimport scanner\n'
                    'with open(os.environ["TARGET_MARKER"], "w", encoding="ascii") as handle:\n'
                    '    handle.write("executed")\n',
                    encoding='ascii',
                )
                environment = {
                    **os.environ,
                    'PYC_PAYLOAD_MARKER': str(pyc_marker),
                    'TARGET_MARKER': str(target_marker),
                }

                completed = subprocess.run(
                    [
                        sys.executable, '-I', '-S', '-B',
                        str(app_dir / 'runtime_bootstrap.py'), 'supervisor', '--',
                    ],
                    cwd=root,
                    env=environment,
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    text=True,
                    timeout=20,
                    check=False,
                )

                self.assertNotEqual(completed.returncode, 0)
                self.assertIn('application __pycache__ bytecode is forbidden', completed.stdout)
                self.assertFalse(pyc_marker.exists())
                self.assertFalse(target_marker.exists())

    def test_authenticated_child_rejects_manifested_cache_before_payload(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            app_dir = root / 'app'
            copy_bootstraps(app_dir)
            pyc_marker = root / 'pyc.marker'
            target_marker = root / 'target.marker'
            write_scanner_cache(app_dir, pyc_marker, py_compile.PycInvalidationMode.UNCHECKED_HASH)
            (app_dir / 'console_runner.py').write_text(
                'import os\nimport scanner\n'
                'with open(os.environ["TARGET_MARKER"], "w", encoding="ascii") as handle:\n'
                '    handle.write("executed")\n',
                encoding='ascii',
            )
            supervisor_path = app_dir / 'supervisor.py'
            supervisor_path.write_text('# fixture\n', encoding='ascii')
            config_path = app_dir / 'config.yaml'
            config_path.write_text('{}\n', encoding='ascii')
            manifest = fixture_manifest(app_dir)
            manifest_digest = hashlib.sha256(json.dumps(
                manifest,
                ensure_ascii=True,
                sort_keys=True,
                separators=(',', ':'),
            ).encode('utf-8')).hexdigest()
            instance_file = root / 'instance.json'
            dsn = 'postgresql://truf:fixture@127.0.0.1:5432/truf'
            dsn_digest = hashlib.sha256(dsn.encode('utf-8')).hexdigest()
            metadata = {
                'schema': 2,
                'instance_file': canonical(instance_file),
                'instance_id': 'cache-test',
                'token': 't' * 48,
                'activation_state': 'ACTIVE',
                'config_path': canonical(config_path),
                'config_sha256': file_sha256(config_path),
                'supervisor_path': canonical(supervisor_path),
                'supervisor_sha256': file_sha256(supervisor_path),
                'code_manifest': manifest,
                'code_manifest_sha256': manifest_digest,
                'canonical_dsn_sha256': dsn_digest,
                'control': {'host': '127.0.0.1', 'port': 1},
            }
            instance_file.write_text(json.dumps(metadata), encoding='ascii')
            environment = {
                **os.environ,
                'TRUF_SUPERVISOR_INSTANCE_FILE': str(instance_file),
                'TRUF_SUPERVISOR_INSTANCE_ID': metadata['instance_id'],
                'TRUF_SUPERVISOR_TOKEN': metadata['token'],
                'TRUF_SUPERVISOR_CONFIG_SHA256': metadata['config_sha256'],
                'TRUF_SUPERVISOR_SHA256': metadata['supervisor_sha256'],
                'TRUF_SUPERVISOR_CODE_MANIFEST_SHA256': manifest_digest,
                'TRUF_SUPERVISOR_DSN_SHA256': dsn_digest,
                'TRUF_SUPERVISOR_CHILD_KIND': 'scanner',
                'TRUF_MANAGED_POSTGRES_DSN': dsn,
                'SCANNER_DB_URL': dsn,
                'DATABASE_URL': dsn,
                'PYC_PAYLOAD_MARKER': str(pyc_marker),
                'TARGET_MARKER': str(target_marker),
            }

            completed = subprocess.run(
                [sys.executable, '-I', '-S', '-B', str(app_dir / 'child_bootstrap.py'), 'scanner', '--'],
                cwd=root,
                env=environment,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                timeout=20,
                check=False,
            )

            self.assertNotEqual(completed.returncode, 0)
            self.assertIn('application __pycache__ bytecode is forbidden', completed.stdout)
            self.assertFalse(pyc_marker.exists())
            self.assertFalse(target_marker.exists())

    def test_venv_site_hooks_do_not_run_and_bytecode_free_launch_creates_no_cache(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            app_dir = root / 'app'
            copy_bootstraps(app_dir)
            (app_dir / 'helper_module.py').write_text('VALUE = "ready"\n', encoding='ascii')
            (app_dir / 'supervisor.py').write_text(
                'import os\nimport helper_module\nimport psycopg\nimport yaml\n'
                'with open(os.environ["TARGET_MARKER"], "w", encoding="ascii") as handle:\n'
                '    handle.write(helper_module.VALUE + ":" + psycopg.VALUE + ":" + yaml.VALUE)\n',
                encoding='ascii',
            )
            venv_dir = root / 'venv'
            venv.EnvBuilder(with_pip=False).create(venv_dir)
            if os.name == 'nt':
                venv_python = venv_dir / 'Scripts' / 'python.exe'
                site_packages = venv_dir / 'Lib' / 'site-packages'
            else:
                venv_python = venv_dir / 'bin' / 'python'
                version = f'python{sys.version_info.major}.{sys.version_info.minor}'
                site_packages = venv_dir / 'lib' / version / 'site-packages'
            site_packages.mkdir(parents=True, exist_ok=True)
            (site_packages / 'psycopg.py').write_text('VALUE = "psycopg"\n', encoding='ascii')
            (site_packages / 'yaml.py').write_text('VALUE = "yaml"\n', encoding='ascii')
            site_marker = root / 'site.marker'
            pth_marker = root / 'pth.marker'
            target_marker = root / 'target.marker'
            (site_packages / 'sitecustomize.py').write_text(
                'import os\nopen(os.environ["SITE_MARKER"], "w", encoding="ascii").write("ran")\n',
                encoding='ascii',
            )
            (site_packages / 'malicious.pth').write_text(
                'import os; open(os.environ["PTH_MARKER"], "w", encoding="ascii").write("ran")\n',
                encoding='ascii',
            )
            environment = {
                **os.environ,
                'SITE_MARKER': str(site_marker),
                'PTH_MARKER': str(pth_marker),
                'TARGET_MARKER': str(target_marker),
            }

            completed = subprocess.run(
                [
                    str(venv_python), '-I', '-S', '-B',
                    str(app_dir / 'runtime_bootstrap.py'), 'supervisor', '--',
                    '--runtime-bootstrap-entrypoint', str(app_dir / 'supervisor.py'),
                ],
                cwd=root,
                env=environment,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                timeout=30,
                check=False,
            )

            self.assertEqual(completed.returncode, 0, completed.stdout)
            self.assertEqual(target_marker.read_text(encoding='ascii'), 'ready:psycopg:yaml')
            self.assertFalse(site_marker.exists())
            self.assertFalse(pth_marker.exists())
            self.assertEqual(list(app_dir.rglob('*.pyc')), [])
            self.assertEqual([path for path in app_dir.rglob('__pycache__') if path.is_dir()], [])

    def test_command_builders_use_exact_isolation_flag_order(self):
        child = supervisor.child_bootstrap_command('scanner', ['--config', 'config.yaml'])
        self.assertEqual(child[:4], [sys.executable, '-I', '-S', '-B'])
        self.assertEqual(Path(child[4]).name, 'child_bootstrap.py')

        producer = supervisor.child_bootstrap_command(
            'discovery-producer', ['--config', 'config.yaml', '--source', 'gitlab', '--once'],
        )
        self.assertEqual(producer[:4], [sys.executable, '-I', '-S', '-B'])
        self.assertEqual(Path(producer[4]).name, 'child_bootstrap.py')
        self.assertEqual(producer[5:7], ['discovery-producer', '--'])

        args = SimpleNamespace(
            sources=None,
            once=False,
            autostart=False,
            status_interval=None,
            dashboard=False,
            no_dashboard=True,
            with_postgres=True,
        )
        background = supervisor.background_child_command(
            args,
            'config.yaml',
            'nonce',
            'instance.json',
        )
        self.assertEqual(background[:4], [sys.executable, '-I', '-S', '-B'])
        self.assertEqual(Path(background[4]).name, 'runtime_bootstrap.py')
        self.assertEqual(background[5:7], ['supervisor', '--'])
        self.assertIn(os.path.abspath(supervisor.__file__), background)
        dash_nonce = supervisor.background_child_command(
            args,
            'config.yaml',
            '-leading-dash',
            'instance.json',
        )
        self.assertIn('--launch-nonce=-leading-dash', dash_nonce)

    def test_direct_mutating_supervisor_fails_before_config_env_locks_or_children(self):
        args = SimpleNamespace(
            cmd=None,
            dry_run=False,
            background_status=False,
            background_child=False,
            with_postgres=True,
        )
        with mock.patch.object(supervisor, 'parse_args', return_value=args), \
                mock.patch.object(supervisor, 'sha256_file') as config_read, \
                mock.patch.object(supervisor, 'load_supervisor_runtime') as load_runtime, \
                mock.patch.object(supervisor, 'load_postgres_env') as load_env, \
                mock.patch.object(supervisor, 'ClusterAuthorityLock') as cluster_lock, \
                mock.patch.object(supervisor, 'SupervisorInstanceLock') as instance_lock, \
                mock.patch.object(supervisor, 'ManagedSource') as child, \
                mock.patch.dict(os.environ, {supervisor.RUNTIME_BOOTSTRAP_ENV: ''}, clear=False):
            with self.assertRaisesRegex(SystemExit, 'canonical runtime bootstrap'):
                supervisor.main()

        for blocked in (config_read, load_runtime, load_env, cluster_lock, instance_lock, child):
            blocked.assert_not_called()

    def test_preimport_gate_rejects_forged_marker_without_isolated_flags(self):
        environment = {**os.environ, supervisor.RUNTIME_BOOTSTRAP_ENV: supervisor.RUNTIME_BOOTSTRAP_VALUE}
        completed = subprocess.run(
            [
                sys.executable,
                str(APP_DIR / 'supervisor.py'),
                '--config', str(APP_DIR / 'config.yaml'),
                '--with-postgres',
                '--non-interactive',
            ],
            cwd=ROOT,
            env=environment,
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            timeout=20,
            check=False,
        )
        self.assertNotEqual(completed.returncode, 0)
        self.assertIn('python -I -S -B via runtime_bootstrap.py', completed.stdout)

    def test_generic_manifest_hashes_cache_bytecode_and_detects_removal(self):
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            app_dir = root / 'app'
            app_dir.mkdir()
            for name in lifecycle_authority.CODE_AUTHORITY_FILES:
                path = app_dir.joinpath(*name.split('/'))
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text(f'# fixture for {name}\n', encoding='ascii')
            external = root / 'runtime' / 'check-openrouter-keys.ps1'
            external.parent.mkdir()
            external.write_text('# fixture\n', encoding='ascii')
            for name in ('start_runtime.ps1', 'stop_runtime.ps1'):
                (root / name).write_text(f'# fixture for {name}\n', encoding='ascii')
            executable = root / 'trufflehog.exe'
            executable.write_bytes(b'fixture')
            cache = app_dir / 'package' / '__pycache__' / 'module.cpython-test.pyc'
            cache.parent.mkdir(parents=True)
            cache.write_bytes(b'cache fixture')

            manifest = lifecycle_authority.build_code_manifest(
                app_dir=app_dir,
                trufflehog_path=executable,
            )
            cache_name = cache.relative_to(app_dir).as_posix()
            self.assertEqual(manifest['schema'], 5)
            self.assertEqual(manifest['files'][cache_name]['sha256'], file_sha256(cache))

            cache.unlink()
            with self.assertRaisesRegex(
                lifecycle_authority.LifecycleAuthorityError,
                'file set drifted: package/__pycache__/module.cpython-test.pyc',
            ):
                lifecycle_authority.verify_code_manifest(manifest)


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