"""Immutable-source compatibility evidence, not installed/release acceptance."""
import hashlib
import subprocess
import tempfile
from pathlib import Path
import types
import unittest

ROOT = Path(__file__).resolve().parents[2]
TAG = 'v2.0.3-alpha'
COMMIT = '1a4ea15df9c8800b6d288277d2839c7b93662ff7'


def historical(path, ref=TAG):
    return subprocess.check_output(['git', '-C', str(ROOT), 'show', ref + ':' + path], text=True)


class ImmutableCompatibilityTests(unittest.TestCase):
    def test_exact_immutable_dispatch_and_standalone_client_contract(self):
        actual = subprocess.check_output(['git', '-C', str(ROOT), 'rev-parse', TAG + '^{commit}'], text=True).strip()
        self.assertEqual(actual, COMMIT)
        update = historical('core/archipelago/src/update.rs')
        entry = update.split('pub async fn apply_update(', 1)[1].split('\n}\n', 1)[0]
        self.assertIn('d09_lifecycle::prepare_apply', entry)
        self.assertNotIn('apply_legacy_update', entry)
        self.assertNotIn('runtime_paths', entry)
        lifecycle = historical('core/archipelago/src/d09_lifecycle.rs')
        self.assertIn('const CLIENT: &str = "/usr/libexec/archipelago-d09ctl"', lifecycle)
        invoke = lifecycle.split('async fn invoke(', 1)[1].split('\n}\n', 1)[0]
        self.assertIn('if crate::d09_transport::installed()', invoke)
        self.assertIn('!crate::host_access::installed()', invoke)
        self.assertIn('Command::new(CLIENT)', invoke)
        self.assertIn('metadata.uid() == 0', invoke)

    def test_historical_runtime_hook_is_reachable_but_nonfatal_and_after_frontend(self):
        update = historical('core/archipelago/src/update.rs', 'v1.8.22-alpha')
        entry = update.split('pub async fn apply_update(', 1)[1].split('/// Rollback', 1)[0]
        self.assertIn('runtime_paths', entry)
        self.assertIn('"archipelago" => 2', entry)
        self.assertIn('n if n.contains("runtime") && n.ends_with(".tar.gz") => 1', entry)
        self.assertIn('ExecStartPre=+-/opt/archipelago/scripts/ota-crash-guard.sh', entry)
        self.assertIn('"archipelago-doctor.service"', entry)
        bootstrap = historical('core/archipelago/src/bootstrap.rs')
        self.assertIn('("scripts", "/opt/archipelago/scripts")', bootstrap)
        self.assertIn('sync_doctor_script().await', bootstrap)
        self.assertIn('"archipelago-doctor.service"', bootstrap)
        # A helper left only in /opt/scripts or the doctor hook is not a stable
        # external worker: the receiving immutable runtime promotes over it.

    def test_immutable_runtime_guard_changes_candidate_template_to_preserve_live_routes(self):
        namespace = {'__name__': 'immutable_guard'}
        exec(compile(historical('scripts/dashboard-public-guard.py'), TAG + ':dashboard-public-guard.py', 'exec'), namespace)
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            active, candidate = root / 'active.conf', root / 'candidate.conf'
            guarded = namespace['guarded']
            # Retain the first failed fixture input: the real guard correctly
            # refuses a partial HTTP-only dashboard instead of weakening it.
            http_only = 'server {\n listen 80 default_server;\n server_name _;\n location /operator-route { return 200; }\n}\n'
            with self.assertRaisesRegex(ValueError, 'both default HTTP and HTTPS'):
                guarded(http_only)
            https = 'server {\n listen 443 ssl default_server;\n server_name _;\n location / { return 200; }\n}\n'
            active.write_text(guarded(http_only + https))
            candidate.write_text(guarded(http_only.replace('/operator-route', '/') + https))
            original_live = active.read_bytes()
            original_candidate = hashlib.sha256(candidate.read_bytes()).hexdigest()
            commands = []

            def nginx_validation_double(args, **kwargs):
                commands.append(args)
                return types.SimpleNamespace(returncode=0)

            namespace['install_runtime'](active, candidate, command=nginx_validation_double, lock_path=root / 'lock')
            self.assertEqual(active.read_bytes(), original_live)
            self.assertEqual(candidate.read_bytes(), original_live)
            self.assertNotEqual(hashlib.sha256(candidate.read_bytes()).hexdigest(), original_candidate)
            self.assertIn(['nginx', '-t'], commands)
            self.assertTrue(list(root.glob('runtime-template.before-preservation-*')))

    def test_public_manifest_is_not_a_d09_admission_or_bootstrap_policy(self):
        provision = historical('scripts/provision-d09-lifecycle.py')
        for required in ['exclusion_evidence_digest', 'effective_config', 'prerequisite',
                         'target_binding_digest', 'restore_artifacts', 'admission_epoch_id']:
            self.assertIn(required, provision)
        self.assertIn("raise ValueError('existing authority; use external readmission')", provision)
        client = historical('scripts/d09-lifecycle-client.py')
        self.assertIn("PUBLIC = pathlib.Path('/etc/archipelago-d09/client.json')", client)
        self.assertIn("selected release differs from external admission", client)
        self.assertIn('binding = json.loads(read(PUBLIC))', client)
        self.assertNotIn('install_package(', client)

    def test_standalone_bundle_is_exact_source_only_and_never_overwrites(self):
        with tempfile.TemporaryDirectory() as directory:
            output = Path(directory) / 'owner.py'
            command = ['python3', '-B', str(ROOT / 'scripts/package-public-update-worker.py'), '--output', str(output)]
            subprocess.run(command, check=True, capture_output=True)
            preparation = (ROOT / 'scripts/public_update_prepare.py').read_text().rsplit("\nif __name__ == '__main__':", 1)[0]
            expected = (preparation + '\n' + (ROOT / 'scripts/public_update_runtime.py').read_text()
                        + '\n' + (ROOT / 'scripts/public_update_adopt.py').read_text()
                        + '\n' + (ROOT / 'scripts/public_update_worker.py').read_text())
            self.assertEqual(output.read_text(), expected)
            self.assertNotEqual(subprocess.run(command, capture_output=True).returncode, 0)
            self.assertEqual(output.read_text(), expected)


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