searchi/tools/test_package_windows.py

import hashlib
import json
from pathlib import Path
import struct
import tempfile
import unittest
from unittest import mock
import zipfile

from package_windows import package, CRT_LIBRARIES


def executable(path, machine=0x8664):
    path.parent.mkdir(parents=True, exist_ok=True)
    data = bytearray(70)
    data[:2] = b"MZ"
    struct.pack_into("<I", data, 60, 64)
    data[64:68] = b"PE\0\0"
    struct.pack_into("<H", data, 68, machine)
    path.write_bytes(data)


def inventory(resources):
    worker = resources / "searchi-gemma-worker.exe"
    records = []
    for name in ["ggml-cpu-x64.dll", "ggml-vulkan.dll", "vulkan-1.dll", "ggml.dll", "ggml-base.dll", "llama.dll", "mtmd.dll", *sorted(CRT_LIBRARIES)]:
        data = (resources / name).read_bytes()
        records.append({"path": name, "bytes": len(data), "sha256": hashlib.sha256(data).hexdigest()})
    (resources / "runtime-libraries.json").write_text(json.dumps({"worker_sha256": hashlib.sha256(worker.read_bytes()).hexdigest(), "libraries": records}))


class PackageTests(unittest.TestCase):
    def fixture(self, directory):
        root = Path(directory)
        resources = root / "apps/desktop/src-tauri/resources"
        app = root / "build/app.exe"
        executable(app)
        (root / "LICENSE").write_text("application notice")
        for name in ["searchi-gemma-worker.exe", "vulkan-1.dll", "ggml-cpu-x64.dll", "ggml-vulkan.dll", "ggml.dll", "ggml-base.dll", "llama.dll", "mtmd.dll", "media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe"]:
            executable(resources / name)
        for name in sorted(CRT_LIBRARIES):
            executable(resources / name)
        for name in ["LLAMA-LICENSE.txt", "MSVC-REDISTRIBUTION.txt", "VULKAN-LICENSE.txt", "licenses/Inter-OFL.txt", "licenses/Fluent-MIT.txt", "media-runtime/licenses/LICENSE.txt", "media-runtime/manifest.json"]:
            path = resources / name
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text("notice")
        (resources / "settings.json").write_text("private settings")
        inventory(resources)
        return root, resources, app, root / "archive.zip"

    def test_complete_archive_excludes_settings_and_verifies_recorded_bytes(self):
        with tempfile.TemporaryDirectory() as directory:
            root, _, app, output = self.fixture(directory)
            package(root, app, output)
            with zipfile.ZipFile(output) as archive:
                self.assertNotIn("settings.json", archive.namelist())
                self.assertIn("media-runtime/bin/ffprobe.exe", archive.namelist())
                for record in json.loads(archive.read("PACKAGE.json"))["files"]:
                    self.assertEqual(hashlib.sha256(archive.read(record["path"])).hexdigest(), record["sha256"])
                    self.assertEqual(len(archive.read(record["path"])), record["bytes"])
                self.assertEqual(len(archive.namelist()), len(set(archive.namelist())))

    def test_mixed_architecture_preserves_previous_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            executable(resources / "searchi-gemma-worker.exe", 0xAA64)
            inventory(resources)
            with self.assertRaisesRegex(ValueError, "architecture differs"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")

    def test_changed_source_records_exact_archived_bytes(self):
        with tempfile.TemporaryDirectory() as directory:
            root, _, app, output = self.fixture(directory)
            original_open = zipfile.ZipFile.open

            def mutate_source(archive, name, mode="r", *args, **kwargs):
                if name == "searchi.exe" and mode == "w":
                    # Change bytes after preflight and after the input stream
                    # opens, at the point ZIP copying starts.
                    app.write_bytes(app.read_bytes() + b"changed while packaging")
                return original_open(archive, name, mode, *args, **kwargs)

            with mock.patch.object(zipfile.ZipFile, "open", mutate_source):
                package(root, app, output)
            with zipfile.ZipFile(output) as archive:
                content = archive.read("searchi.exe")
                record = json.loads(archive.read("PACKAGE.json"))["files"][0]
                self.assertIn(b"changed while packaging", content)
                self.assertEqual(record["bytes"], len(content))
                self.assertEqual(record["sha256"], hashlib.sha256(content).hexdigest())

    def test_mid_write_failure_preserves_previous_archive_and_cleans_temporary(self):
        with tempfile.TemporaryDirectory() as directory:
            root, _, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            original_open = zipfile.ZipFile.open

            def fail_second_entry(archive, name, mode="r", *args, **kwargs):
                if name == "LICENSE.txt" and mode == "w":
                    raise OSError("simulated storage failure")
                return original_open(archive, name, mode, *args, **kwargs)

            with mock.patch.object(zipfile.ZipFile, "open", fail_second_entry):
                with self.assertRaisesRegex(OSError, "storage failure"):
                    package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")
            self.assertEqual(list(output.parent.glob("*.partial")), [])

    def test_missing_named_notice_preserves_previous_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            (resources / "licenses/Inter-OFL.txt").unlink()
            with self.assertRaisesRegex(ValueError, "Inter-OFL"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")

    def test_mixed_loader_architecture_preserves_previous_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            executable(resources / "vulkan-1.dll", 0xAA64)
            inventory(resources)
            with self.assertRaisesRegex(ValueError, "architecture differs: vulkan-1.dll"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")

    def test_inventory_includes_dispatch_libraries_but_not_unlisted_dll(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            executable(resources / "private-plugin.dll")
            package(root, app, output)
            with zipfile.ZipFile(output) as archive:
                self.assertIn("ggml-cpu-x64.dll", archive.namelist())
                self.assertIn("ggml-vulkan.dll", archive.namelist())
                self.assertIn("runtime-libraries.json", archive.namelist())
                self.assertIn("msvcp140.dll", archive.namelist())
                self.assertIn("MSVC-REDISTRIBUTION.txt", archive.namelist())
                self.assertNotIn("private-plugin.dll", archive.namelist())

    def test_missing_inventory_or_corrupt_library_preserves_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            (resources / "ggml-cpu-x64.dll").write_bytes(b"corrupt")
            with self.assertRaisesRegex(ValueError, "checksum mismatch"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")
            (resources / "runtime-libraries.json").unlink()
            with self.assertRaisesRegex(ValueError, "inventory"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")

    def test_dependency_change_during_copy_preserves_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            original_open = zipfile.ZipFile.open

            def mutate_library(archive, name, mode="r", *args, **kwargs):
                if name == "ggml-cpu-x64.dll" and mode == "w":
                    library = resources / name
                    library.write_bytes(library.read_bytes() + b"changed")
                return original_open(archive, name, mode, *args, **kwargs)

            with mock.patch.object(zipfile.ZipFile, "open", mutate_library):
                with self.assertRaisesRegex(ValueError, "changed while packaging"):
                    package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")
            self.assertEqual(list(output.parent.glob("*.partial")), [])

    def test_missing_crt_or_notice_preserves_previous_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            (resources / "msvcp140.dll").unlink()
            with self.assertRaisesRegex(ValueError, "msvcp140"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")
            executable(resources / "msvcp140.dll")
            (resources / "MSVC-REDISTRIBUTION.txt").unlink()
            with self.assertRaisesRegex(ValueError, "MSVC-REDISTRIBUTION"):
                package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")

    def test_inventory_change_during_copy_preserves_previous_archive(self):
        with tempfile.TemporaryDirectory() as directory:
            root, resources, app, output = self.fixture(directory)
            output.write_bytes(b"previous package")
            original_open = zipfile.ZipFile.open

            def mutate_inventory(archive, name, mode="r", *args, **kwargs):
                if name == "runtime-libraries.json" and mode == "w":
                    manifest = resources / name
                    manifest.write_bytes(manifest.read_bytes() + b" ")
                return original_open(archive, name, mode, *args, **kwargs)

            with mock.patch.object(zipfile.ZipFile, "open", mutate_inventory):
                with self.assertRaisesRegex(ValueError, "changed while packaging"):
                    package(root, app, output)
            self.assertEqual(output.read_bytes(), b"previous package")


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