Package verified speech runtime in Linux headless archives

tools/test_package_linux.py

@@ -11,6 +11,7 @@ from unittest import mock
 import package_linux as pack
 from embeddinggemma2 import runtime_provenance as provenance
 import stage_extraction_worker as extraction
+import stage_speech_worker as speech
 
 
 def elf(path, machine=62):
@@ -22,6 +23,79 @@ def elf(path, machine=62):
 
 
 class LinuxPackageTests(unittest.TestCase):
+    def speech_fixture(self, fixture):
+        root = fixture[0]
+        runtime = root / 'speech-runtime'
+        runtime.mkdir()
+        for name in speech.INPUTS:
+            path = speech.input_path(root, name)
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(speech.input_path(speech.ROOT, name).read_bytes())
+        files = []
+        for name in ('searchi-speech-worker', 'libwhisper.so.1', 'libggml-cpu-x64.so', 'libggml-vulkan.so'):
+            path = runtime / name
+            elf(path)
+            files.append({'path': name, 'bytes': path.stat().st_size, 'sha256': pack.digest(path)})
+        (runtime / 'runtime-manifest.json').write_text(json.dumps({'schema': 1,
+            'upstream_commit': speech.UPSTREAM, 'device': 'auto', 'distribution': True,
+            'source_hashes': {name: pack.digest(speech.input_path(root, name)) for name in speech.INPUTS},
+            'files': files}))
+        for name in speech.NOTICES:
+            notice = root / 'assets/licenses' / name
+            notice.parent.mkdir(parents=True, exist_ok=True)
+            notice.write_text('Approved test notice ' + name)
+            (runtime / name).write_bytes(notice.read_bytes())
+        return runtime
+
+    def test_speech_archive_has_isolated_verified_files_modes_and_instructions(self):
+        with tempfile.TemporaryDirectory() as directory:
+            fixture = self.fixture(directory)
+            runtime = self.speech_fixture(fixture)
+            with mock.patch.object(pack, 'dynamic', side_effect=self.metadata), \
+                    mock.patch.object(speech, 'verify_linux_closure') as closure:
+                pack.package(*fixture, speech_runtime=runtime)
+                closure.assert_called_once_with(runtime)
+                first = fixture[-1].read_bytes()
+                pack.package(*fixture, speech_runtime=runtime)
+                self.assertEqual(first, fixture[-1].read_bytes())
+            with tarfile.open(fixture[-1]) as archive:
+                manifest = json.load(archive.extractfile('PACKAGE.json'))
+                self.assertTrue(manifest['features']['local_speech'])
+                self.assertEqual(manifest['features']['speech_status'], 'experimental')
+                self.assertEqual(archive.getmember('speech-runtime/searchi-speech-worker').mode, 0o755)
+                for name in (*speech.NOTICES, 'runtime-manifest.json'):
+                    self.assertEqual(archive.getmember('speech-runtime/' + name).mode, 0o644)
+                for record in manifest['files']:
+                    data = archive.extractfile(record['path']).read()
+                    self.assertEqual(record['sha256'], hashlib.sha256(data).hexdigest())
+                instructions = archive.extractfile('START-HERE.txt').read().decode()
+                self.assertIn('setup --speech --model-dir ./speech-models', instructions)
+                self.assertIn('--speech-worker ./speech-runtime/searchi-speech-worker', instructions)
+                self.assertIn('--speech-model-dir ./speech-models', instructions)
+
+    def test_invalid_speech_runtime_preserves_previous_archive(self):
+        for mutation in ('cpu', 'tampered', 'notice', 'closure'):
+            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
+                fixture = self.fixture(directory)
+                runtime = self.speech_fixture(fixture)
+                fixture[-1].write_bytes(b'previous archive')
+                if mutation == 'cpu':
+                    path = runtime / 'runtime-manifest.json'
+                    manifest = json.loads(path.read_text())
+                    manifest.update(device='cpu', distribution=False)
+                    path.write_text(json.dumps(manifest))
+                elif mutation == 'tampered':
+                    (runtime / 'libwhisper.so.1').write_bytes(b'tampered')
+                elif mutation == 'notice':
+                    (runtime / speech.NOTICES[0]).unlink()
+                with mock.patch.object(pack, 'dynamic', side_effect=self.metadata), \
+                        mock.patch.object(speech, 'verify_linux_closure',
+                            side_effect=ValueError('Invalid ELF closure') if mutation == 'closure' else None):
+                    with self.assertRaises(ValueError):
+                        pack.package(*fixture, speech_runtime=runtime)
+                self.assertEqual(fixture[-1].read_bytes(), b'previous archive')
+                self.assertEqual(list(fixture[-1].parent.glob('*.partial')), [])
+
     def test_extraction_missing_stale_tampered_or_wrong_architecture_preserves_output(self):
         for mutation in ("missing", "inventory_missing", "stale", "tampered", "architecture"):
             with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory: