119 lines
5.7 KiB
Python
119 lines
5.7 KiB
Python
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
|
|
|
|
|
|
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)
|
|
|
|
|
|
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", "media-runtime/bin/ffmpeg.exe", "media-runtime/bin/ffprobe.exe"]:
|
|
executable(resources / name)
|
|
for name in ["LLAMA-LICENSE.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")
|
|
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)
|
|
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)
|
|
with self.assertRaisesRegex(ValueError, "architecture differs: vulkan-1.dll"):
|
|
package(root, app, output)
|
|
self.assertEqual(output.read_bytes(), b"previous package")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|