tests/test_ffmpeg_build.py
Rama master · réplica de lectura del repositorio original.
"""FFmpeg 对应源码归档与发行清单的行为验证。"""
from __future__ import annotations
import importlib.util
import io
import json
import os
from pathlib import Path
import tarfile
import tempfile
import subprocess
import sys
import unittest
from unittest import mock
import urllib.error
import zipfile
from contextlib import nullcontext
import httpx
SCRIPT = Path(__file__).resolve().parents[1] / "scripts/fetch_ffmpeg_sources.py"
SPEC = importlib.util.spec_from_file_location("ffmpeg_source_build", SCRIPT)
assert SPEC and SPEC.loader
builder = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(builder)
class FFmpegSourceBuildTests(unittest.TestCase):
def setUp(self):
self.opened = mock.Mock()
client = mock.MagicMock()
client.__enter__.return_value = client
def stream(method, url, **kwargs):
result = self.opened(method, url, **kwargs)
return nullcontext(httpx.Response(200, content=result.read(), request=httpx.Request(method, url)))
client.stream.side_effect = stream
self.enterContext(mock.patch.object(builder, "http_client", return_value=client))
def test_source_cache_falls_back_to_pinned_upstream_on_transport_failure(self):
import hashlib
data = b"pinned source bytes"
source = {"name": "ffmpeg", "url": "https://upstream.example/source.tar.xz", "archive": "source.tar.xz",
"sha256": hashlib.sha256(data).hexdigest()}
with tempfile.TemporaryDirectory() as temporary, mock.patch.dict(os.environ, {"PEACH_SOURCE_CACHE_BASE": "https://cache.example/_sources"}), \
mock.patch.object(self, "opened", side_effect=[
urllib.error.HTTPError("https://cache.example", 404, "missing", {}, None), io.BytesIO(data)]) as opened:
archive = Path(temporary).resolve() / "source.tar.xz"
builder.obtain_source(source, archive)
self.assertEqual(archive.read_bytes(), data)
self.assertEqual([call.args[1] for call in opened.call_args_list],
[f"https://cache.example/_sources/{source['sha256']}/source.tar.xz", source["url"]])
def test_source_cache_hash_mismatch_is_fatal(self):
with tempfile.TemporaryDirectory() as temporary, mock.patch.dict(os.environ, {"PEACH_SOURCE_CACHE_BASE": "https://cache.example/_sources"}), \
mock.patch.object(self, "opened", return_value=io.BytesIO(b"corrupt")) as opened:
archive = Path(temporary).resolve() / "source.tar.xz"
with self.assertRaisesRegex(ValueError, "SHA256"):
builder.obtain_source({"name": "ffmpeg", "url": "https://upstream.example/source.tar.xz",
"archive": "source.tar.xz", "sha256": "a" * 64}, archive)
self.assertEqual(opened.call_count, 1)
self.assertFalse(archive.exists())
def test_download_retries_transport_failure_and_keeps_only_verified_archive(self):
data = b"fixed upstream source"
import hashlib
source = {"name": "ffmpeg", "url": "https://upstream.example/source.tar.xz",
"sha256": hashlib.sha256(data).hexdigest()}
with tempfile.TemporaryDirectory() as temporary:
archive = Path(temporary).resolve() / "source.tar.xz"
with mock.patch.object(self, "opened", side_effect=[TimeoutError(), io.BytesIO(data)]) as opened, \
mock.patch.object(builder.time, "sleep") as sleep:
builder.download_source(source, archive)
self.assertEqual(archive.read_bytes(), data)
self.assertEqual(opened.call_count, 2)
sleep.assert_called_once_with(2)
self.assertFalse(archive.with_suffix(".xz.part").exists())
def test_download_integrity_or_permanent_http_error_is_not_retried(self):
for failure in (None, urllib.error.HTTPError("https://upstream.example", 403, "Forbidden", {}, None)):
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as temporary:
archive = Path(temporary).resolve() / "source.tar.xz"
with mock.patch.object(self, "opened", side_effect=failure,
return_value=io.BytesIO(b"corrupt")) as opened, \
mock.patch.object(builder.time, "sleep") as sleep:
with self.assertRaises((ValueError, urllib.error.HTTPError)):
builder.download_source({"name": "ffmpeg", "url": "https://upstream.example/source.tar.xz",
"sha256": "a" * 64}, archive)
self.assertEqual(opened.call_count, 1)
sleep.assert_not_called()
self.assertFalse(archive.exists())
self.assertFalse(archive.with_suffix(".xz.part").exists())
def test_download_stops_after_three_transport_attempts(self):
with tempfile.TemporaryDirectory() as temporary:
archive = Path(temporary).resolve() / "source.tar.xz"
with mock.patch.object(self, "opened", side_effect=TimeoutError()) as opened, \
mock.patch.object(builder.time, "sleep") as sleep:
with self.assertRaises(TimeoutError):
builder.download_source({"name": "ffmpeg", "url": "https://upstream.example/source.tar.xz",
"sha256": "a" * 64}, archive)
self.assertEqual(opened.call_count, 3)
self.assertEqual(sleep.call_args_list, [mock.call(2), mock.call(4)])
self.assertFalse(archive.exists())
def test_source_lock_pins_all_linked_external_dependencies(self):
lock = builder.load_sources()
self.assertEqual(lock["version"], "9.0.2")
self.assertEqual(lock["license"], "GPL-3.0-or-later")
self.assertEqual({source["name"] for source in lock["sources"]},
{"ffmpeg", "x264", "nv-codec-headers", "dav1d", "amf-headers", "libvpl"})
def make_sources(self, output):
lock = builder.load_sources()
directory = output / "sources"
directory.mkdir()
for source in lock["sources"]:
archive = directory / source["archive"]
with tarfile.open(archive, "w:gz") as package:
data = b"upstream source and license\n"
entry = tarfile.TarInfo(f"{source['name']}-revision/README")
entry.size = len(data)
package.addfile(entry, io.BytesIO(data))
source["sha256"] = builder.digest(archive)
return lock
def test_prepare_verifies_archives_before_extracting(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary).resolve()
lock = self.make_sources(output)
builder.prepare(output, lock)
for source in lock["sources"]:
self.assertTrue((output / "work" / source["name"] / "README").is_file())
with self.assertRaises(FileExistsError):
builder.prepare(output, lock)
def test_prepare_rejects_source_archive_tampering(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary).resolve()
lock = self.make_sources(output)
(output / "sources" / lock["sources"][0]["archive"]).write_bytes(b"corrupt")
with self.assertRaisesRegex(ValueError, "SHA256"):
builder.prepare(output, lock)
self.assertFalse((output / "work/ffmpeg").exists())
def test_finish_includes_sources_scripts_licenses_and_matching_hashes(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary).resolve()
lock = self.make_sources(output)
builder.prepare(output, lock)
for path in ("ffmpeg/COPYING.GPLv3", "ffmpeg/COPYING.LGPLv2.1",
"ffmpeg/COPYING.LGPLv3", "x264/COPYING", "dav1d/COPYING",
"libvpl/LICENSE", "libvpl/third-party-programs.txt"):
(output / "work" / path).write_text("license", encoding="utf-8")
headers = output / "work/nv-codec-headers/include/ffnvcodec"
headers.mkdir(parents=True)
(headers / "test.h").write_text("/* MIT license */\ncode", encoding="utf-8")
amf = output / "work/amf-headers/AMF/core"
amf.mkdir(parents=True)
for name in ("Version.h", "Platform.h"):
(amf / name).write_text("// MIT license\n// Copyright AMD\n#pragma once\n", encoding="utf-8")
(output / "ffmpeg/bin").mkdir(parents=True)
for name in ("ffmpeg", "ffprobe"):
(output / f"ffmpeg/bin/{name}.exe").write_bytes(b"MZ fixture")
(output / "buildconf.mak").write_text("CONFIG_GPL=yes\n", encoding="utf-8")
manifest = builder.finish(output, lock)
source_zip = output / manifest["source_archive"]["name"]
self.assertEqual(builder.digest(source_zip), manifest["source_archive"]["sha256"])
for path, expected in manifest["files"].items():
self.assertEqual(builder.digest(output / "ffmpeg" / path), expected)
self.assertEqual((output / "ffmpeg/LICENSE-amf-headers.txt").read_text(encoding="utf-8"),
"// MIT license\n// Copyright AMD\n")
self.assertTrue({"LICENSE-libvpl-LICENSE.txt", "LICENSE-libvpl-third-party-programs.txt"}
<= set(manifest["files"]))
with zipfile.ZipFile(source_zip) as package:
names = set(package.namelist())
self.assertTrue({f"sources/{s['archive']}" for s in lock["sources"]} <= names)
self.assertTrue({"scripts/build_ffmpeg_windows.sh", "scripts/fetch_ffmpeg_sources.py",
"scripts/ffmpeg-sources.json", "README.md", "buildconf.txt"} <= names)
self.assertFalse(any("manifest.json" in name for name in names))
independent = output / "independent-source"
package.extractall(independent)
help_result = subprocess.run([sys.executable, "-S", str(independent / "scripts/fetch_ffmpeg_sources.py"), "--help"],
cwd=independent, env={**os.environ, "PYTHONPATH": ""},
capture_output=True, text=True, encoding="utf-8", check=True)
self.assertIn("prepare", help_result.stdout)
stored = json.loads((output / "ffmpeg/manifest.json").read_text(encoding="utf-8"))
self.assertEqual(stored, manifest)
if __name__ == "__main__":
unittest.main()