#!/usr/bin/env python3
"""Control-plane regressions with fake tools; never an engine/native test pass."""
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("validation_tests", ROOT / "scripts/test-cloud-fixture-validation.py")
fixtures = importlib.util.module_from_spec(spec)
spec.loader.exec_module(fixtures)

FAKE_PODMAN = r'''#!/usr/bin/env python3
import json,os,pathlib,sys,subprocess
root=pathlib.Path(os.environ['FAKE_ROOT']); args=sys.argv[1:]
with (root/'calls').open('a') as f:f.write(json.dumps(args)+'\n')
if args[0]=='create':
    compile=any(a.startswith('archy-cloud-compile-') for a in args)
    cid=('c' if compile else 'd')*64
    (root/cid).write_text(json.dumps(args))
    if compile:
        for a in args:
            if a.endswith(':/artifacts/test-binary:rw'):
                pathlib.Path(a[:-len(':/artifacts/test-binary:rw')]).write_bytes(b'synthetic artifact, never executed')
    print(cid)
elif args[0]=='start':
    if args[-1]=='d'*64:
        create=json.loads((root/args[-1]).read_text())
        canary=next(a.split('=',1)[1] for a in create if a.startswith('ARCHY_TEST_HOST_CANARY='))
        assert pathlib.Path(canary).is_file(), 'Host-only canary must exist during execution'
        assert not any(a.startswith(canary+':') for a in create), 'Canary must never be mounted'
    if args[-1]=='c'*64 and os.environ.get('FAIL_COMPILE'):sys.exit(1)
    if args[-1]=='d'*64 and os.environ.get('FAIL_NAMESPACE'):
        create=json.loads((root/args[-1]).read_text()); pos=create.index('-ec')
        # Real shell guard, with a deliberately wrong synthetic readlink result.
        # It MUST exit before attempting the artifact; no backend runs on host.
        script=create[pos+1]
        assert 'Fixture namespace changed' in script
        result=subprocess.run(['/bin/sh','-ec',script,'cloud-netns','net:[200]'],env=os.environ)
        sys.exit(result.returncode)
elif args[:2]==['image','exists']:
    if os.environ.get('FAIL_RUNTIME'): sys.exit(1)
elif args[0]=='inspect': print((root/'inspect.json').read_text())
elif args[0]=='rm': pass
else: sys.exit(99)
'''


class Runner(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.path = Path(self.temp.name)
        (self.path / 'bin').mkdir()
        (self.path / 'repo/core').mkdir(parents=True)
        (self.path / 'cache/registry').mkdir(parents=True)
        for name, content in {
            'podman': FAKE_PODMAN,
            'cargo': '#!/bin/sh\necho called > "$FAKE_ROOT/host-cargo"\nexit 99\n',
            'sudo': '#!/bin/sh\nprintf "net:[%s]\\n" "${FAKE_EXPECTED:-200}"\n',
            'readlink': '#!/bin/sh\necho "net:[100]"\n',
        }.items():
            script = self.path / 'bin' / name
            script.write_text(content); script.chmod(0o755)
        (self.path / 'inspect.json').write_text(json.dumps(fixtures.fixture()))
        self.env = dict(os.environ, PATH=f'{self.path}/bin:' + os.environ['PATH'],
                        FAKE_ROOT=str(self.path), ARCHY_TEST_ISOLATOR='podman',
                        ARCHY_TEST_REPO=str(self.path / 'repo'), ARCHY_TEST_IMAGE='b' * 64,
                        ARCHY_TEST_CARGO_HOME=str(self.path / 'cache'),
                        ARCHY_TEST_CLOUD_FIXTURE_CONTAINER=fixtures.NAME,
                        ARCHY_TEST_CLOUD_FIXTURE_IMAGE=fixtures.IMAGE)

    def run_wrapper(self):
        return subprocess.run(['bash', str(ROOT / 'scripts/test-backend-isolated.sh'), '--ignored'],
                              env=self.env, capture_output=True, text=True, timeout=20)

    def calls(self):
        file = self.path / 'calls'
        return [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []

    def test_systemd_cloud_lane_refuses_before_host_cargo(self):
        self.env['ARCHY_TEST_ISOLATOR'] = 'systemd'
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertFalse((self.path / 'host-cargo').exists())
        self.assertFalse(self.calls())

    def test_live_worktree_and_mutable_builder_are_refused(self):
        git = self.path / 'repo/.git'; git.write_text('gitdir: /synthetic')
        self.assertEqual(self.run_wrapper().returncode, 2)
        git.unlink(); self.env['ARCHY_TEST_IMAGE'] = 'compiler:latest'
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(self.calls())

    def test_mocked_success_preserves_compile_and_execution_boundaries(self):
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 0, result.stderr)
        creates = [c for c in self.calls() if c[0] == 'create']
        self.assertEqual(len(creates), 2)
        compile, execute = creates
        for args in creates:
            for option in ['--pull=never', '--http-proxy=false', '--read-only', '--cap-drop=all',
                           '--security-opt=no-new-privileges', '--log-driver=none', '--cpus=2']:
                self.assertIn(option, args)
            writable = [a for a in args if a.endswith(':rw')]
            self.assertTrue(all(a.endswith(':/artifacts/test-binary:rw') for a in writable))
            self.assertIn(str(self.path / 'repo') + ':/workspace:ro', args)
            self.assertFalse(any('podman.sock' in a or 'docker.sock' in a for a in args))
        self.assertIn('--network=none', compile)
        self.assertIn('--memory=8g', compile)
        self.assertIn('--memory-swap=8g', compile)
        self.assertIn('--pids-limit=512', compile)
        self.assertIn('--timeout=1800', compile)
        self.assertIn('CARGO_NET_OFFLINE=true', compile)
        self.assertIn('CARGO_INCREMENTAL=0', compile)
        self.assertIn('profile.test.package.archipelago.codegen-units=16', compile[compile.index('-ec') + 1])
        self.assertIn(str(self.path / 'cache/registry') + ':/cargo-home/registry:ro', compile)
        self.assertIn('--network=container:' + fixtures.NAME, execute)
        self.assertIn('--memory=3g', execute)
        self.assertIn('--pids-limit=256', execute)
        self.assertIn('--timeout=120', execute)
        for path, size in [('/tmp', '128m'), ('/root', '16m'), ('/run', '16m'),
                           ('/var/lib/archipelago', '64m'), ('/var/lib/containers', '16m')]:
            self.assertIn(f'{path}:rw,notmpcopyup,size={size}', execute)
        self.assertIn('net:[200]', execute)
        canaries = [a.split('=', 1)[1] for a in execute if a.startswith('ARCHY_TEST_HOST_CANARY=')]
        self.assertEqual(len(canaries), 1)
        self.assertTrue(canaries[0].startswith('/tmp/archy-cloud-test.'))
        self.assertFalse(Path(canaries[0]).exists(), 'Cleanup must remove the host canary')
        self.assertIn('readlink /proc/self/ns/net', execute[execute.index('-ec') + 1])
        self.assertFalse((self.path / 'host-cargo').exists())
        self.assertEqual(len([c for c in self.calls() if c[0] == 'rm']), 2)

    def test_artifact_export_retains_binary_and_binding_without_overwrite(self):
        output = self.path / 'export'; output.mkdir()
        self.env['ARCHY_TEST_ARTIFACT_EXPORT'] = str(output)
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 0, result.stderr)
        receipt = json.loads((output / 'artifact.json').read_text())
        self.assertEqual(receipt['compiler_image'], 'b' * 64)
        self.assertEqual(len(receipt['source_sha256']), 64)
        self.assertEqual((output / 'test-binary').read_bytes(), b'synthetic artifact, never executed')
        self.assertEqual(self.run_wrapper().returncode, 2)

    def compile_artifact(self):
        output = self.path / 'export'; output.mkdir()
        self.env['ARCHY_TEST_ARTIFACT_EXPORT'] = str(output)
        self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '1'
        fixture = self.env.pop('ARCHY_TEST_CLOUD_FIXTURE_CONTAINER')
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertFalse(any(c[0] == 'inspect' for c in self.calls()))
        self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)
        self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '0'
        self.env['ARCHY_TEST_CLOUD_FIXTURE_CONTAINER'] = fixture
        del self.env['ARCHY_TEST_ARTIFACT_EXPORT']
        self.env['ARCHY_TEST_REUSE_ARTIFACT'] = str(output)
        (self.path / 'calls').unlink()
        return output

    def test_compile_before_fixture_then_exact_artifact_reuse(self):
        self.compile_artifact()
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 0, result.stderr)
        creates = [c for c in self.calls() if c[0] == 'create']
        self.assertEqual(len(creates), 1)
        self.assertTrue(any('archy-cloud-execute-' in a for a in creates[0]))
        self.assertFalse((self.path / 'host-cargo').exists())

    def test_reuse_rejects_source_change_before_namespace_inspection(self):
        self.compile_artifact()
        (self.path / 'repo/core/changed').write_text('new source')
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))

    def test_reuse_rejects_changed_runtime_and_binary(self):
        output = self.compile_artifact()
        self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
        self.assertEqual(self.run_wrapper().returncode, 2)
        del self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE']
        binary = output / 'test-binary'; binary.chmod(0o700); binary.write_bytes(b'changed')
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))

    def test_reuse_rejects_symlinked_artifact(self):
        output = self.compile_artifact()
        binary = output / 'test-binary'; saved = output / 'saved'; binary.rename(saved); binary.symlink_to(saved)
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(any(c[0] in ('create', 'inspect') for c in self.calls()))

    def test_compile_only_requires_export_before_creating_container(self):
        self.env['ARCHY_TEST_CLOUD_COMPILE_ONLY'] = '1'
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(any(c[0] == 'create' for c in self.calls()))

    def test_runtime_image_is_pinned_and_preflighted_before_compile(self):
        self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'browser:latest'
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(self.calls())
        self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
        self.env['FAIL_RUNTIME'] = '1'
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertFalse(any(c[0] == 'create' for c in self.calls()))

    def test_browser_runtime_does_not_change_compiler_or_isolation(self):
        self.env['ARCHY_TEST_CLOUD_RUNTIME_IMAGE'] = 'e' * 64
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 0, result.stderr)
        compile, execute = [c for c in self.calls() if c[0] == 'create']
        self.assertEqual(compile[compile.index('-ec') - 1], 'b' * 64)
        self.assertEqual(execute[execute.index('-ec') - 1], 'e' * 64)
        self.assertIn('--network=none', compile)
        self.assertIn('--network=container:' + fixtures.NAME, execute)
        self.assertIn('--cap-drop=all', execute)

    def test_compile_failure_never_inspects_or_enters_fixture(self):
        self.env['FAIL_COMPILE'] = '1'
        self.assertNotEqual(self.run_wrapper().returncode, 0)
        self.assertFalse(any(c[0] == 'inspect' for c in self.calls()))
        self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)

    def test_host_namespace_refused_before_execution_container(self):
        self.env['FAKE_EXPECTED'] = '100'
        self.assertEqual(self.run_wrapper().returncode, 2)
        self.assertEqual(len([c for c in self.calls() if c[0] == 'create']), 1)

    def test_namespace_reuse_fails_real_shell_guard_before_artifact(self):
        self.env['FAIL_NAMESPACE'] = '1'
        result = self.run_wrapper()
        self.assertEqual(result.returncode, 2, result.stderr)
        self.assertIn('Fixture namespace changed', result.stdout)

    def test_log_sink_is_bounded(self):
        result = subprocess.run(['python3', str(ROOT / 'scripts/limit-test-output.py')],
                                input=b'x' * (8 * 1024 * 1024 + 1), capture_output=True, timeout=10)
        self.assertEqual(result.returncode, 125)
        self.assertLessEqual(len(result.stdout), 8 * 1024 * 1024)


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