#!/usr/bin/env python3
"""Packaging invariants; stdlib synthetic files, no Cargo, builds or GUI."""
import importlib.util
import json
from pathlib import Path
import struct
import tarfile
import tempfile
import unittest
from unittest.mock import patch

spec = importlib.util.spec_from_file_location('packaging', Path(__file__).with_name('package-artifacts.py'))
packaging = importlib.util.module_from_spec(spec)
spec.loader.exec_module(packaging)


class PackageContracts(unittest.TestCase):
    def setUp(self):
        self.temporary = tempfile.TemporaryDirectory(prefix='arcadia-packaging-test-')
        self.root = Path(self.temporary.name)

    def tearDown(self):
        self.temporary.cleanup()

    def file(self, name, data=b'test'):
        path = self.root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_bytes(data)
        return path

    def elf(self, needed=b'libc.so.6', interpreter=b'/lib64/ld-linux-x86-64.so.2', runpath=False):
        # One load maps the file, followed by interpreter and dynamic segments.
        data = bytearray(1024)
        data[:6] = b'\x7fELF\x02\x01'
        struct.pack_into('<HH', data, 16, 3, 62)
        struct.pack_into('<Q', data, 32, 64)
        struct.pack_into('<HH', data, 54, 56, 3)
        data[300:300 + len(interpreter) + 1] = interpreter + b'\0'
        strings = b'\0' + needed + b'\0'
        data[600:600 + len(strings)] = strings
        struct.pack_into('<IIQQQQQQ', data, 64, 1, 4, 0, 0x400000, 0, len(data), len(data), 4096)
        struct.pack_into('<IIQQQQQQ', data, 120, 3, 4, 300, 0x40012c, 0, len(interpreter) + 1, len(interpreter) + 1, 1)
        entries = [(5, 0x400000 + 600), (10, len(strings)), (1, 1)]
        if runpath:
            entries.append((29, 1))
        entries.append((0, 0))
        struct.pack_into('<IIQQQQQQ', data, 176, 2, 4, 400, 0x400190, 0, len(entries) * 16, len(entries) * 16, 8)
        for i, entry in enumerate(entries):
            struct.pack_into('<QQ', data, 400 + i * 16, *entry)
        return self.file('input.elf', data)

    def test_tar_is_deterministic_with_executable_modes_and_unicode_paths(self):
        self.file('stage/arcadia-studio', b'not executable in this test')
        self.file('stage/arcadia-server', b'another inert fixture')
        self.file('stage/examples/temperature delta.txt', 'Δ\n'.encode())
        first = packaging.make_archive(self.root / 'stage', self.root / 'a.tar.gz', 'fixture')
        second = packaging.make_archive(self.root / 'stage', self.root / 'b.tar.gz', 'fixture')
        self.assertEqual(first['sha256'], second['sha256'])
        with tarfile.open(self.root / 'a.tar.gz') as archive:
            self.assertEqual(archive.getmember('fixture/arcadia-studio').mode, 0o755)
            self.assertEqual(archive.getmember('fixture/examples/temperature delta.txt').mode, 0o644)
            self.assertEqual(len(archive.getmembers()), 3)

    def test_archive_refuses_overwrite_and_case_collisions(self):
        self.file('stage/a', b'first')
        archive = self.file('archive.tar.gz', b'preserve')
        with self.assertRaises(FileExistsError):
            packaging.make_archive(self.root / 'stage', archive, 'fixture')
        self.assertEqual(archive.read_bytes(), b'preserve')
        with patch.object(packaging, 'stage_files', return_value={}):
            with self.assertRaises(ValueError):
                packaging.make_archive(self.root / 'stage', archive, '../escape')

    def test_archive_refuses_a_file_added_during_readback(self):
        self.file('stage/a', b'first')
        original = packaging.tarfile.open
        def open_and_add(*args, **kwargs):
            result = original(*args, **kwargs)
            if len(args) > 1 and args[1] == 'r:gz':
                self.file('stage/late.txt', b'late')
            return result
        with patch.object(packaging.tarfile, 'open', side_effect=open_and_add):
            with self.assertRaisesRegex(ValueError, 'Stage changed'):
                packaging.make_archive(self.root / 'stage', self.root / 'archive.tar.gz', 'fixture')

    def test_input_rejects_a_reparse_ancestor(self):
        source = self.file('ordinary/file.txt')
        original = Path.lstat
        def reparse(path, **kwargs):
            if path == source.parent:
                result = original(path, **kwargs)
                return type('ReparseStat', (), {'st_mode': result.st_mode, 'st_file_attributes': 1024})()
            return original(path, **kwargs)
        with patch.object(Path, 'lstat', reparse), patch.object(packaging.stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 1024, create=True):
            with self.assertRaisesRegex(ValueError, 'reparse ancestor'):
                packaging.ordinary(source)

    def test_local_document_links_are_complete_and_cannot_escape(self):
        self.file('docs/README.md', b'[guide](guide%20one.md#section)\n[web](https://example.com)\n')
        self.file('docs/guide one.md', b'# section\n')
        self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 1)
        self.file('docs/README.md', b'[missing](missing.md)')
        with self.assertRaises(ValueError):
            packaging.check_documents(self.root / 'docs')
        self.file('docs/README.md', b'[escape](../outside.md)')
        with self.assertRaises(ValueError):
            packaging.check_documents(self.root / 'docs')

    def test_rst_directives_and_fenced_examples_are_not_links(self):
        self.file('docs/a.rst', b'.. code-block:: rust\n\n    sample\n')
        self.file('docs/a.md', b'```text\n[example](absent.md)\n```\n`[sample](absent.md)`\n')
        self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 0)

    def test_html_attribute_case_spacing_and_unquoted_links_are_checked(self):
        self.file('docs/a.html', b'<A HREF = "missing.html">Missing</A>')
        with self.assertRaises(ValueError):
            packaging.check_documents(self.root / 'docs')
        self.file('docs/missing.html', b'ok')
        self.file('docs/a.html', b'<a href=missing.html>Present</a>')
        self.assertEqual(packaging.check_documents(self.root / 'docs')['checked_local_document_links'], 1)

    def test_elf_accepts_reviewed_glibc_and_refuses_unknown_runtime(self):
        self.assertEqual(packaging.inspect_elf(self.elf())['needed'], ['libc.so.6'])
        with self.assertRaisesRegex(ValueError, 'Unreviewed'):
            packaging.inspect_elf(self.elf(needed=b'private-runtime.so'))
        with self.assertRaisesRegex(ValueError, 'RPATH'):
            packaging.inspect_elf(self.elf(runpath=True))

    def test_elf_rejects_wrong_architecture_truncation_and_loader(self):
        with self.assertRaisesRegex(ValueError, 'glibc dynamic loader'):
            packaging.inspect_elf(self.elf(interpreter=b'/lib/ld-musl-x86_64.so.1'))
        elf = self.elf()
        data = bytearray(elf.read_bytes())
        struct.pack_into('<H', data, 18, 183)
        elf.write_bytes(data)
        with self.assertRaisesRegex(ValueError, 'x86-64'):
            packaging.inspect_elf(elf)
        elf.write_bytes(b'\x7fELF')
        with self.assertRaisesRegex(ValueError, 'ELF64'):
            packaging.inspect_elf(elf)

    def test_container_refuses_wrong_target_roots_and_stale_lock_before_tools(self):
        lock = self.file('repo/Cargo.lock', b'lock')
        inventory = {'target': 'x86_64-pc-windows-msvc', 'root_packages': ['oa-server'],
                     'cargo_lock_sha256': packaging.digest(lock)}
        self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
        with patch.object(packaging.subprocess, 'check_output', side_effect=AssertionError('must not run')):
            with self.assertRaisesRegex(ValueError, 'server-only'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', 'x86_64-unknown-linux-gnu')
            inventory['target'] = 'x86_64-unknown-linux-gnu'
            inventory['cargo_lock_sha256'] = '0' * 64
            self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
            with self.assertRaisesRegex(ValueError, 'Cargo.lock mismatch'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])

    def test_container_verifies_actual_notices_compiler_libraries_and_sources(self):
        lock = self.file('repo/Cargo.lock', b'lock')
        source = self.file('repo/crates/oa-server/src/main.rs', b'source')
        self.file('repo/Cargo.toml', b'workspace')
        self.file('repo/LICENSE', b'license')
        self.file('repo/docs/openapi.json', b'{}')
        self.file('repo/build/windows_manifest.rs', b'build script')
        library = self.file('rust/lib/libstd.rlib', b'rust runtime')
        notice = self.file('notices/licenses/example.txt', b'upstream notice')
        version = ['release: 1.90.0', 'commit-hash: example']
        inventory = {
            'target': 'x86_64-unknown-linux-gnu', 'root_packages': ['oa-server'],
            'cargo_lock_sha256': packaging.digest(lock),
            'workspace_build_inputs_sha256': {name: packaging.digest(self.root / 'repo' / name) for name in
                                               ['Cargo.toml', 'Cargo.lock', 'LICENSE', 'docs/openapi.json', 'build/windows_manifest.rs']},
            'rust': {'version': version, 'target_libraries_sha256': {library.name: packaging.digest(library)},
                     'notices': [{'path': 'licenses/example.txt', 'sha256': packaging.digest(notice)}]},
            'packages': [{'name': 'oa-server', 'source': None,
                          'local_source_sha256': {'src/main.rs': packaging.digest(source)}}]}
        self.file('notices/dependency-notices.json', json.dumps(inventory).encode())
        def manifest():
            lines = [f'{packaging.digest(path)}  {path.relative_to(self.root / "notices").as_posix()}'
                     for path in sorted((self.root / 'notices').rglob('*'))
                     if path.is_file() and path.name != 'SHA256SUMS.txt']
            self.file('notices/SHA256SUMS.txt', ('\n'.join(lines) + '\n').encode())
        manifest()
        def compiler(args, **_):
            return '\n'.join(version) if args == ['rustc', '-Vv'] else str(library.parent)
        with patch.object(packaging.subprocess, 'check_output', side_effect=compiler):
            result = packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            self.assertEqual(result['container_notice_preflight'], 'passed')
            self.file('repo/crates/oa-server/src/added.rs', b'new module')
            with self.assertRaisesRegex(ValueError, 'workspace differs'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            (self.root / 'repo/crates/oa-server/src/added.rs').unlink()
            self.file('repo/docs/openapi.json', b'changed API')
            with self.assertRaisesRegex(ValueError, 'shared build input'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            self.file('repo/docs/openapi.json', b'{}')
            source.write_bytes(b'new source')
            with self.assertRaisesRegex(ValueError, 'workspace differs'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            source.write_bytes(b'source')
            notice.write_bytes(b'altered notice')
            with self.assertRaisesRegex(ValueError, 'file mismatch'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            manifest()
            with self.assertRaisesRegex(ValueError, 'Inventory notice reference'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])
            library.write_bytes(b'different compiler runtime')
            with self.assertRaisesRegex(ValueError, 'target-library'):
                packaging.verify_notices(self.root / 'notices', self.root / 'repo', inventory['target'])


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