"""人脸模型随安装包分发：取模型的脚本、Docker 镜像的同一组文件，以及完整包到运行时的整条路。 真模型不进测试、不出网：四个文件换成小字节，哈希按它们重算。 """ import hashlib import json from pathlib import Path import re import tempfile import unittest from unittest.mock import patch import zipfile from peach import face_detect, face_match, face_models, update_packages from scripts import fetch_face_models from scripts.package_windows_release import package as package_release ROOT = Path(__file__).resolve().parents[1] def sha(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() class FixtureCase(unittest.TestCase): """四个文件换成小字节；`FILES` 与两个运行时模块的哈希一起换，脚本与运行时仍是同一组判据。""" def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name).resolve() self.payloads = {name: f"fixture {name}".encode() for name, _, _ in fetch_face_models.FILES} self.urls = {url: name for name, url, _ in fetch_face_models.FILES} files = tuple((name, url, sha(self.payloads[name])) for name, url, _ in fetch_face_models.FILES) for patcher in ( patch.object(fetch_face_models, "FILES", files), patch.object(face_detect, "MODEL_SHA256", sha(self.payloads[face_detect.MODEL_PATH.name])), patch.object(face_match, "MODEL_SHA256", sha(self.payloads[face_match.MODEL_PATH.name])), patch.object(face_detect, "_verified", {})): patcher.start() self.addCleanup(patcher.stop) self.fetched = [] def fetch(self, url: str) -> bytes: self.fetched.append(url) return self.payloads[self.urls[url]] class FetchScriptTests(FixtureCase): def test_the_bundle_holds_every_file_the_tool_package_requires(self): destination = self.root / "tools" / "face-models" fetch_face_models.install(destination, cache=self.root / "cache", fetch=self.fetch) _, required = update_packages.TOOL_PATHS["face-models"] self.assertEqual({path.name for path in destination.iterdir()}, required) manifest = json.loads((destination / "manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["commit"], face_detect.ZOO_COMMIT) self.assertEqual({model["file"]: (model["license"], model["license_file"]) for model in manifest["models"]}, {face_detect.MODEL_PATH.name: ("MIT", "LICENSE-yunet.txt"), face_match.MODEL_PATH.name: ("Apache-2.0", "LICENSE-sface.txt")}) def test_a_verified_cache_is_reused_without_network(self): cache = self.root / "cache" fetch_face_models.install(self.root / "first", cache=cache, fetch=self.fetch) self.assertEqual(len(self.fetched), 4) fetch_face_models.install(self.root / "second", cache=cache, fetch=self.fetch) self.assertEqual(len(self.fetched), 4, "缓存里哈希对得上的文件不再联网") self.assertEqual((self.root / "first" / "manifest.json").read_bytes(), (self.root / "second" / "manifest.json").read_bytes(), "清单内容固定，工具包哈希可复现") def test_a_download_whose_digest_differs_is_refused_and_not_cached(self): cache = self.root / "cache" with self.assertRaisesRegex(ValueError, "SHA-256 不符"): fetch_face_models.install(self.root / "out", cache=cache, fetch=lambda url: b"tampered") self.assertEqual(list(cache.iterdir()), []) def test_the_pinned_files_point_at_one_opencv_zoo_commit(self): for name, url, digest in fetch_face_models.FILES: with self.subTest(name=name): self.assertEqual(url.split("/opencv_zoo/", 1)[1].split("/", 1)[0], face_detect.ZOO_COMMIT) self.assertTrue(re.fullmatch(r"[0-9a-f]{64}", digest)) class DockerImageTests(unittest.TestCase): def test_the_image_adds_the_same_files_with_the_same_digests(self): dockerfile = (ROOT / "Dockerfile").read_text(encoding="utf-8") added = {Path(target).name: (url, digest) for digest, url, target in re.findall( r"ADD --chmod=a\+rX --checksum=sha256:([0-9a-f]{64}) \\\n\s+(\S+) \\\n\s+(/app/tools/face-models/\S+)", dockerfile)} self.assertEqual(added, {name: (url, digest) for name, url, digest in fetch_face_models.FILES}) self.assertEqual(re.findall(r"PEACH_FACE_MODELS=(\S+)", dockerfile), ["/app/tools/face-models"]) class FullPackageTests(FixtureCase): """完整包里带两个模型、哈希对得上；解开后运行时直接用它，不联网、不往数据根写。""" def test_the_full_package_ships_both_models_and_the_runtime_uses_them_offline(self): source = self.root / "dist" / "Peach" (source / "_internal").mkdir(parents=True) (source / "Peach.exe").write_bytes(b"application") (source / "_internal/standalone.txt").touch() for tool, (prefix, required) in update_packages.TOOL_PATHS.items(): if tool == "face-models": continue for name in required: path = source / prefix / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(name.encode()) fetch_face_models.install(source / "tools" / "face-models", cache=self.root / "cache", fetch=self.fetch) output = self.root / "release" manifest = package_release(source, output, "1.0.0") prefix = "Peach/tools/face-models/" with zipfile.ZipFile(output / "Peach-1.0.0-windows-x64.zip") as full: shipped = {name.removeprefix(prefix): sha(full.read(name)) for name in full.namelist() if name.startswith(prefix)} full.extractall(self.root / "installed") for module in (face_detect, face_match): self.assertEqual(shipped[module.MODEL_PATH.name], module.MODEL_SHA256) self.assertEqual({"tools/face-models/" + name for name in shipped}, set(manifest["tools"]["face-models"]["files"])) with zipfile.ZipFile(output / "Peach-1.0.0-windows-x64-update.zip") as thin: self.assertFalse([name for name in thin.namelist() if name.startswith(prefix)], "应用更新包不带模型") data = self.root / "data" bundle = self.root / "installed" / "Peach" / "tools" / "face-models" with patch.dict(face_detect.os.environ, {face_detect.BUNDLE_ENV: str(bundle)}), \ patch.object(face_detect, "MODEL_PATH", data / "yunet" / face_detect.MODEL_PATH.name), \ patch.object(face_match, "MODEL_PATH", data / "sface" / face_match.MODEL_PATH.name), \ patch.object(face_detect, "read_url", side_effect=AssertionError("程序目录有模型时不该联网")): self.assertEqual(face_detect.ensure_model(), bundle / face_detect.MODEL_PATH.name) self.assertEqual(face_match.ensure_model(), bundle / face_match.MODEL_PATH.name) self.assertEqual(face_models.download_all(), {"ok": True, "outcome": "人脸模型已就绪"}) self.assertFalse(data.exists(), "随包那份不往数据根复制") if __name__ == "__main__": unittest.main()