60 lines
2.7 KiB
Python
60 lines
2.7 KiB
Python
|
import hashlib
|
||
|
|
import importlib.util
|
||
|
|
from pathlib import Path
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
import zipfile
|
||
|
|
|
||
|
|
spec = importlib.util.spec_from_file_location("media_provision", Path(__file__).with_name("provision.py"))
|
||
|
|
provision = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(provision)
|
||
|
|
|
||
|
|
|
||
|
|
class ProvisionTests(unittest.TestCase):
|
||
|
|
def package(self, directory, entries):
|
||
|
|
archive = directory / "sample.zip"
|
||
|
|
with zipfile.ZipFile(archive, "w") as package:
|
||
|
|
for name, data in entries:
|
||
|
|
package.writestr(name, data)
|
||
|
|
return archive
|
||
|
|
|
||
|
|
def test_selects_only_tools_and_preserves_distinct_notices(self):
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
directory = Path(temporary)
|
||
|
|
archive = self.package(directory, [("build/bin/ffmpeg.exe", b"tool1"),
|
||
|
|
("build/bin/ffprobe.exe", b"tool2"), ("build/README.txt", b"readme"),
|
||
|
|
("build/LICENSE", b"first license"), ("build/dependency/LICENSE", b"second license"),
|
||
|
|
("../../unselected.exe", b"unselected")])
|
||
|
|
target = directory / "out"
|
||
|
|
provision.extract(archive, target, True)
|
||
|
|
self.assertEqual((target / "bin/ffmpeg.exe").read_bytes(), b"tool1")
|
||
|
|
self.assertEqual(len(list((target / "licenses").rglob("LICENSE"))), 2)
|
||
|
|
self.assertFalse((directory / "unselected.exe").exists())
|
||
|
|
|
||
|
|
def test_missing_and_duplicate_executables_are_explicit_failures(self):
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
directory = Path(temporary)
|
||
|
|
for entries in [[("bin/ffmpeg.exe", b"tool")],
|
||
|
|
[("a/ffmpeg.exe", b"tool"), ("b/ffmpeg.exe", b"second"), ("bin/ffprobe.exe", b"probe")]]:
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
provision.extract(self.package(directory, entries), directory / "out", True)
|
||
|
|
|
||
|
|
def test_same_size_corruption_is_not_a_verified_cache(self):
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
path = Path(temporary) / "archive"
|
||
|
|
path.write_bytes(b"good")
|
||
|
|
artifact = {"bytes": 4, "sha256": hashlib.sha256(b"good").hexdigest()}
|
||
|
|
self.assertTrue(provision.verified(path, artifact))
|
||
|
|
path.write_bytes(b"evil")
|
||
|
|
self.assertFalse(provision.verified(path, artifact))
|
||
|
|
|
||
|
|
def test_notice_traversal_cannot_escape_output(self):
|
||
|
|
with tempfile.TemporaryDirectory() as temporary:
|
||
|
|
directory = Path(temporary)
|
||
|
|
archive = self.package(directory, [("../../LICENSE", b"escape")])
|
||
|
|
with self.assertRaises(ValueError):
|
||
|
|
provision.extract(archive, directory / "out", True)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|