Dmitry Dementiev
38c99855e2
Зачем: - Проверить переносимость Intel-handoff на Ryzen и закрыть выбор CPU-профиля. Что: - Усилена проверка связи Ryzen-manifest с капсулой и переносимости рецептов. - Добавлены регрессионные тесты и публичный отчет с исходом new map. Проверка: - uv run --no-sync pytest -q (389 passed, 2 skipped). - uv run --no-sync ruff check src/local_transcriber/benchmark_experiment.py tests/test_benchmark_experiment.py.
1294 lines
45 KiB
Python
1294 lines
45 KiB
Python
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import zipfile
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from local_transcriber import benchmark_experiment as experiment
|
||
|
||
|
||
def _write(path: Path, content: bytes) -> dict[str, object]:
|
||
path.write_bytes(content)
|
||
return {
|
||
"path": str(path),
|
||
"sha256": hashlib.sha256(content).hexdigest(),
|
||
"size_bytes": len(content),
|
||
}
|
||
|
||
|
||
def _artifact(tmp_path: Path, artifact_id: str, content: bytes) -> dict[str, object]:
|
||
return {
|
||
"id": artifact_id,
|
||
"kind": "onnx-model",
|
||
**_write(tmp_path / f"{artifact_id}.onnx", content),
|
||
"source_url": f"https://example.invalid/{artifact_id}",
|
||
"revision": "v1",
|
||
"license": "Apache-2.0",
|
||
}
|
||
|
||
|
||
def _cell(name: str, position: int, role: str) -> dict[str, object]:
|
||
return {
|
||
"name": name,
|
||
"stage": "intel",
|
||
"phase": "window",
|
||
"recording_ids": ["data-test", "t2-bdma", "yantar"],
|
||
"window_shift_ratio": 0.1 if role == "baseline" else 0.2,
|
||
"segmentation_artifact_id": "pyannote-fp32",
|
||
"embedding_artifact_id": "wespeaker-fp32",
|
||
"counter": "threshold",
|
||
"clustering": {"mode": "threshold", "threshold": 0.89, "num_clusters": None},
|
||
"inference": {
|
||
"mode": "sequential",
|
||
"outer_workers": 1,
|
||
"session_count": 1,
|
||
"intra_op_threads": 4,
|
||
"inter_op_threads": 1,
|
||
"batch_size": 1,
|
||
},
|
||
"repetition": 1,
|
||
"schedule_position": position,
|
||
"pair_id": "window-ab",
|
||
"pair_role": role,
|
||
"enabled": True,
|
||
"skip_reason": None,
|
||
"build_sha256": "9" * 64,
|
||
}
|
||
|
||
|
||
def _manifest(tmp_path: Path) -> dict[str, object]:
|
||
artifacts = [
|
||
_artifact(tmp_path, "pyannote-fp32", b"segmentation"),
|
||
_artifact(tmp_path, "wespeaker-fp32", b"wespeaker"),
|
||
_artifact(tmp_path, "wespeaker-qdq", b"qdq"),
|
||
]
|
||
recordings = []
|
||
for index, (recording_id, speakers) in enumerate(
|
||
(("data-test", 3), ("t2-bdma", 2), ("yantar", 2))
|
||
):
|
||
media = _write(tmp_path / f"{recording_id}.wav", recording_id.encode())
|
||
words = _write(
|
||
tmp_path / f"{recording_id}.words.json",
|
||
json.dumps([{"start": 0.0, "end": 0.5, "text": "слово"}]).encode(),
|
||
)
|
||
reference = None
|
||
if index:
|
||
side = _write(tmp_path / f"{recording_id}.md", b"reference")
|
||
reference = {
|
||
"path": side["path"],
|
||
"sha256": side["sha256"],
|
||
"format": "hypescribe_markdown",
|
||
}
|
||
recordings.append(
|
||
{
|
||
"id": recording_id,
|
||
"role": "control",
|
||
"path": media["path"],
|
||
"sha256": media["sha256"],
|
||
"start": 0.0,
|
||
"duration": 300.0,
|
||
"expected_speakers": speakers,
|
||
"reference": reference,
|
||
"asr_words": {
|
||
"path": words["path"],
|
||
"sha256": words["sha256"],
|
||
"format": "local_transcriber_words_v1",
|
||
},
|
||
}
|
||
)
|
||
calibration = _write(tmp_path / "calibration.wav", b"separate calibration")
|
||
return {
|
||
"schema_version": 3,
|
||
"experiment": {
|
||
"id": "cpu-diarization-intel",
|
||
"stage": "intel",
|
||
"source_commit": "1" * 64,
|
||
"handoff_commit": None,
|
||
"sherpa_patch_sha256": "2" * 64,
|
||
"python_version": "3.13.0",
|
||
"dependencies": {
|
||
"sherpa-onnx": "1.13.6",
|
||
"sherpa-onnx-core": "1.13.6",
|
||
"onnxruntime": "1.28.0",
|
||
"numpy": "2.4.3",
|
||
"onnx": "1.20.0",
|
||
"kaldi-native-fbank": "1.22.3",
|
||
"psutil": "7.0.0",
|
||
},
|
||
},
|
||
"machine": {
|
||
"id": "intel-i7-6820hq",
|
||
"sku": "Intel Core i7-6820HQ",
|
||
"physical_cores": 4,
|
||
"logical_cores": os.cpu_count(),
|
||
"ram_bytes": 32 * 1024**3,
|
||
"os": "Linux test",
|
||
"cpu_flags": ["avx", "avx2"],
|
||
"power": {
|
||
"supply": "ac",
|
||
"profile": "balanced",
|
||
"governor": "powersave",
|
||
"turbo": True,
|
||
"power_limits": "platform-default",
|
||
"temperature_celsius": 55.0,
|
||
"throttling": False,
|
||
"swap_total_bytes": 8 * 1024**3,
|
||
"swap_used_bytes": 0,
|
||
},
|
||
},
|
||
"artifacts": artifacts,
|
||
"recordings": recordings,
|
||
"calibration": [
|
||
{
|
||
"id": "calibration-1",
|
||
"path": calibration["path"],
|
||
"sha256": calibration["sha256"],
|
||
"source_id": "calibration-source-1",
|
||
"source_sha256": calibration["sha256"],
|
||
"source_start": 10.0,
|
||
"source_duration": 30.0,
|
||
"derived": False,
|
||
}
|
||
],
|
||
"qdq": {
|
||
"enabled": True,
|
||
"source_ids": ["calibration-1"],
|
||
"preprocessing": {"sample_rate": 16000},
|
||
"reader": {"name": "speaker-features-v1"},
|
||
"quant_pre_process": {"skip_optimization": False},
|
||
"quantize_static": {
|
||
"format": "QDQ",
|
||
"activation": "QInt8",
|
||
"weight": "QInt8",
|
||
},
|
||
"excluded_nodes": [],
|
||
"tool_versions": {"onnxruntime": "1.28.0"},
|
||
"graph_artifact_id": "wespeaker-qdq",
|
||
"skip_reason": None,
|
||
},
|
||
"cells": [_cell("W0", 1, "baseline"), _cell("W2", 2, "candidate")],
|
||
"handoff": {
|
||
"outcome": "pending",
|
||
"source_commit": None,
|
||
"candidate_recipes": [],
|
||
"public_report_sha256": None,
|
||
"capsule_id": None,
|
||
"capsule_sha256": None,
|
||
},
|
||
"rss_sample_interval_ms": 10,
|
||
}
|
||
|
||
|
||
def _environment() -> dict[str, object]:
|
||
return {
|
||
"packages": {
|
||
"sherpa-onnx": "1.13.6",
|
||
"sherpa-onnx-core": "1.13.6",
|
||
"onnxruntime": "1.28.0",
|
||
"numpy": "2.4.3",
|
||
"onnx": "1.20.0",
|
||
"kaldi-native-fbank": "1.22.3",
|
||
"psutil": "7.0.0",
|
||
},
|
||
"python": "3.13.0",
|
||
"platform": "Linux test",
|
||
"logical_cores": os.cpu_count(),
|
||
"physical_cores": 4,
|
||
"cpu_sku": "Intel Core i7-6820HQ",
|
||
"cpu_flags": ["avx", "avx2"],
|
||
"ram_bytes": 32 * 1024**3,
|
||
"swap_total_bytes": 8 * 1024**3,
|
||
"swap_used_bytes": 0,
|
||
"build_sha256": "9" * 64,
|
||
}
|
||
|
||
|
||
def _finalized_handoff_manifest(tmp_path: Path) -> dict[str, object]:
|
||
manifest = _manifest(tmp_path)
|
||
baseline = manifest["cells"][0]
|
||
baseline.update(
|
||
name="W0",
|
||
phase="combination",
|
||
pair_id="combination-c2",
|
||
pair_role="baseline",
|
||
schedule_position=10,
|
||
)
|
||
candidate = copy.deepcopy(baseline)
|
||
candidate.update(
|
||
name="C2",
|
||
pair_role="candidate",
|
||
schedule_position=11,
|
||
)
|
||
candidate["inference"].update(
|
||
mode="shared-session",
|
||
outer_workers=2,
|
||
session_count=1,
|
||
intra_op_threads=4,
|
||
)
|
||
manifest["cells"] = [baseline, candidate]
|
||
cell_id = experiment.make_cell_id(manifest, candidate)
|
||
manifest["handoff"] = {
|
||
"outcome": "handoff",
|
||
"source_commit": manifest["experiment"]["source_commit"],
|
||
"candidate_recipes": [
|
||
{
|
||
"name": "C2",
|
||
"cell_id": cell_id,
|
||
"result_file": "results/w0.json",
|
||
"result_sha256": "3" * 64,
|
||
"mandatory_passed": True,
|
||
"memory_passed": True,
|
||
"diagnostic_approved": True,
|
||
}
|
||
],
|
||
"public_report_sha256": "4" * 64,
|
||
"capsule_id": "opaque-41",
|
||
"capsule_sha256": "5" * 64,
|
||
}
|
||
return manifest
|
||
|
||
|
||
def _ryzen_capsule_fixture(tmp_path: Path):
|
||
intel = _finalized_handoff_manifest(tmp_path)
|
||
payloads = {}
|
||
|
||
def make_relative(item, archive_path):
|
||
content = Path(item["path"]).read_bytes()
|
||
item["path"] = archive_path
|
||
payloads[archive_path] = content
|
||
local_path = tmp_path / archive_path
|
||
local_path.parent.mkdir(parents=True, exist_ok=True)
|
||
local_path.write_bytes(content)
|
||
|
||
for artifact in intel["artifacts"]:
|
||
make_relative(artifact, f"artifacts/{artifact['id']}.bin")
|
||
for recording in intel["recordings"]:
|
||
make_relative(recording, f"recordings/{recording['id']}.wav")
|
||
make_relative(
|
||
recording["asr_words"], f"sidecars/{recording['id']}.words.json"
|
||
)
|
||
if recording["reference"] is not None:
|
||
make_relative(
|
||
recording["reference"], f"references/{recording['id']}.md"
|
||
)
|
||
for source in intel["calibration"]:
|
||
make_relative(source, f"calibration/{source['id']}.wav")
|
||
|
||
result_bytes = b'{"aggregate":{"quality_passed":true}}'
|
||
result_path = "results/c2.json"
|
||
payloads[result_path] = result_bytes
|
||
candidate = intel["handoff"]["candidate_recipes"][0]
|
||
candidate["result_file"] = result_path
|
||
candidate["result_sha256"] = hashlib.sha256(result_bytes).hexdigest()
|
||
intel_bytes = json.dumps(intel, separators=(",", ":")).encode()
|
||
files = {
|
||
"manifest.json": hashlib.sha256(intel_bytes).hexdigest(),
|
||
**{name: hashlib.sha256(value).hexdigest() for name, value in payloads.items()},
|
||
}
|
||
capsule = {
|
||
"capsule_id": intel["handoff"]["capsule_id"],
|
||
"source_commit": intel["handoff"]["source_commit"],
|
||
"public_report_sha256": intel["handoff"]["public_report_sha256"],
|
||
"files": files,
|
||
}
|
||
archive_path = tmp_path / "capsule.zip"
|
||
with zipfile.ZipFile(archive_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(capsule))
|
||
archive.writestr("manifest.json", intel_bytes)
|
||
for name, content in payloads.items():
|
||
archive.writestr(name, content)
|
||
|
||
ryzen = copy.deepcopy(intel)
|
||
ryzen["experiment"].update(
|
||
id="cpu-diarization-ryzen",
|
||
stage="ryzen",
|
||
handoff_commit="6" * 40,
|
||
)
|
||
ryzen["machine"].update(
|
||
id="ryzen-8845h",
|
||
sku="AMD Ryzen 7 8845H",
|
||
physical_cores=8,
|
||
logical_cores=16,
|
||
)
|
||
for cell in ryzen["cells"]:
|
||
cell["stage"] = "ryzen"
|
||
cell["inference"]["intra_op_threads"] = 8
|
||
return archive_path, ryzen, files["manifest.json"]
|
||
|
||
|
||
def _guard(**changes) -> dict[str, object]:
|
||
result = {
|
||
"supply": "ac",
|
||
"profile": "balanced",
|
||
"governor": "powersave",
|
||
"turbo": True,
|
||
"power_limits": {"package": "platform-default"},
|
||
"temperature_celsius": 55.0,
|
||
"throttling": False,
|
||
"throttle_count": 10,
|
||
"swap_used_bytes": 0,
|
||
"swap_sin_bytes": 100,
|
||
"swap_sout_bytes": 200,
|
||
}
|
||
result.update(changes)
|
||
return result
|
||
|
||
|
||
def test_manifest_identity_is_path_independent_and_covers_schedule(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
experiment.validate_manifest(manifest)
|
||
first_id = experiment.make_cell_id(manifest, manifest["cells"][0])
|
||
moved = copy.deepcopy(manifest)
|
||
for artifact in moved["artifacts"]:
|
||
artifact["path"] = "/private/moved/" + Path(artifact["path"]).name
|
||
for recording in moved["recordings"]:
|
||
recording["path"] = "/private/moved/recording"
|
||
recording["asr_words"]["path"] = "/private/moved/words"
|
||
if recording["reference"]:
|
||
recording["reference"]["path"] = "/private/moved/reference"
|
||
moved["calibration"][0]["path"] = "/private/moved/calibration"
|
||
|
||
assert experiment.make_cell_id(moved, moved["cells"][0]) == first_id
|
||
|
||
moved["cells"][0]["schedule_position"] = 3
|
||
assert experiment.make_cell_id(moved, moved["cells"][0]) != first_id
|
||
assert experiment.make_experiment_id(moved) != experiment.make_experiment_id(
|
||
manifest
|
||
)
|
||
|
||
finalized = copy.deepcopy(manifest)
|
||
finalized["handoff"]["outcome"] = "stop-before-ryzen"
|
||
assert experiment.make_experiment_id(finalized) == experiment.make_experiment_id(
|
||
manifest
|
||
)
|
||
|
||
|
||
def test_cell_identity_covers_sidecars_and_qdq_recipe(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
cell = manifest["cells"][0]
|
||
cell["embedding_artifact_id"] = "wespeaker-qdq"
|
||
first_id = experiment.make_cell_id(manifest, cell)
|
||
|
||
changed_sidecar = copy.deepcopy(manifest)
|
||
changed_sidecar["recordings"][0]["asr_words"]["sha256"] = "7" * 64
|
||
assert experiment.make_cell_id(changed_sidecar, changed_sidecar["cells"][0]) != first_id
|
||
|
||
changed_recipe = copy.deepcopy(manifest)
|
||
changed_recipe["qdq"]["quantize_static"]["weight"] = "QInt4"
|
||
assert experiment.make_cell_id(changed_recipe, changed_recipe["cells"][0]) != first_id
|
||
|
||
|
||
def test_identity_and_runtime_validation_cover_exact_versions_and_build(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
first_id = experiment.make_cell_id(manifest, manifest["cells"][0])
|
||
|
||
changed_python = copy.deepcopy(manifest)
|
||
changed_python["experiment"]["python_version"] = "3.13.1"
|
||
assert (
|
||
experiment.make_cell_id(changed_python, changed_python["cells"][0])
|
||
!= first_id
|
||
)
|
||
|
||
changed_numpy = copy.deepcopy(manifest)
|
||
changed_numpy["experiment"]["dependencies"]["numpy"] = "2.4.4"
|
||
assert (
|
||
experiment.make_cell_id(changed_numpy, changed_numpy["cells"][0])
|
||
!= first_id
|
||
)
|
||
with pytest.raises(ValueError, match="numpy==2.4.4"):
|
||
experiment.validate_environment(changed_numpy, _environment())
|
||
|
||
changed_build = copy.deepcopy(_environment())
|
||
changed_build["build_sha256"] = "8" * 64
|
||
with pytest.raises(ValueError, match="runtime-сборки"):
|
||
experiment.validate_environment(manifest, changed_build)
|
||
|
||
|
||
def test_portable_handoff_check_skips_only_machine_identity(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
foreign_environment = copy.deepcopy(_environment())
|
||
foreign_environment.update(
|
||
platform="Linux Ryzen",
|
||
logical_cores=24,
|
||
physical_cores=12,
|
||
cpu_sku="AMD Ryzen 9 5900X 12-Core Processor",
|
||
cpu_flags=["avx", "avx2", "avx512f"],
|
||
ram_bytes=64 * 1024**3,
|
||
swap_total_bytes=0,
|
||
)
|
||
|
||
experiment.validate_environment(
|
||
manifest,
|
||
foreign_environment,
|
||
check_machine=False,
|
||
)
|
||
|
||
foreign_environment["packages"]["numpy"] = "2.4.4"
|
||
with pytest.raises(ValueError, match="numpy==2.4.3"):
|
||
experiment.validate_environment(
|
||
manifest,
|
||
foreign_environment,
|
||
check_machine=False,
|
||
)
|
||
|
||
|
||
def test_ryzen_manifest_preserves_intel_capsule_inputs_and_evidence(tmp_path):
|
||
intel = _finalized_handoff_manifest(tmp_path)
|
||
experiment.validate_manifest(intel, verify_files=False)
|
||
ryzen = copy.deepcopy(intel)
|
||
ryzen["experiment"].update(
|
||
id="cpu-diarization-ryzen",
|
||
stage="ryzen",
|
||
handoff_commit="6" * 40,
|
||
)
|
||
ryzen["machine"].update(
|
||
id="ryzen-8845h",
|
||
sku="AMD Ryzen 7 8845H",
|
||
physical_cores=8,
|
||
logical_cores=16,
|
||
)
|
||
for cell in ryzen["cells"]:
|
||
cell["stage"] = "ryzen"
|
||
cell["inference"]["intra_op_threads"] = 8
|
||
ryzen["cells"][0]["name"] = "R-W0-primary"
|
||
ryzen["cells"][1]["name"] = "R-C2-primary"
|
||
|
||
experiment.validate_manifest(ryzen, verify_files=False)
|
||
experiment._validate_ryzen_capsule_manifest(ryzen, intel)
|
||
|
||
changed_input = copy.deepcopy(ryzen)
|
||
changed_input["artifacts"][0]["sha256"] = "7" * 64
|
||
with pytest.raises(ValueError, match="закрепленные входы"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_input, intel)
|
||
|
||
changed_handoff = copy.deepcopy(ryzen)
|
||
changed_handoff["handoff"]["public_report_sha256"] = "8" * 64
|
||
with pytest.raises(ValueError, match="handoff"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_handoff, intel)
|
||
|
||
changed_recipe = copy.deepcopy(ryzen)
|
||
changed_recipe["cells"][1]["window_shift_ratio"] = 0.2
|
||
with pytest.raises(ValueError, match="рецептом handoff"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_recipe, intel)
|
||
|
||
changed_candidate = copy.deepcopy(intel)
|
||
changed_candidate["handoff"]["candidate_recipes"][0]["cell_id"] = "8" * 64
|
||
changed_candidate_ryzen = copy.deepcopy(ryzen)
|
||
changed_candidate_ryzen["handoff"]["candidate_recipes"][0]["cell_id"] = (
|
||
"8" * 64
|
||
)
|
||
with pytest.raises(ValueError, match="Intel combination"):
|
||
experiment._validate_ryzen_capsule_manifest(
|
||
changed_candidate_ryzen, changed_candidate
|
||
)
|
||
|
||
changed_baseline = copy.deepcopy(ryzen)
|
||
changed_baseline["cells"][0]["window_shift_ratio"] = 0.5
|
||
with pytest.raises(ValueError, match="baseline"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_baseline, intel)
|
||
|
||
changed_phase = copy.deepcopy(ryzen)
|
||
changed_phase["cells"][1]["phase"] = "mechanism"
|
||
changed_phase["cells"][1]["window_shift_ratio"] = 0.5
|
||
with pytest.raises(ValueError, match="механизмом handoff"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_phase, intel)
|
||
|
||
changed_patch = copy.deepcopy(ryzen)
|
||
changed_patch["experiment"]["sherpa_patch_sha256"] = "7" * 64
|
||
with pytest.raises(ValueError, match="runtime"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_patch, intel)
|
||
|
||
changed_dependencies = copy.deepcopy(ryzen)
|
||
changed_dependencies["experiment"]["dependencies"]["onnxruntime"] = "1.99.0"
|
||
with pytest.raises(ValueError, match="runtime"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_dependencies, intel)
|
||
|
||
changed_rss_interval = copy.deepcopy(ryzen)
|
||
changed_rss_interval["rss_sample_interval_ms"] = 5000
|
||
with pytest.raises(ValueError, match="RSS"):
|
||
experiment._validate_ryzen_capsule_manifest(changed_rss_interval, intel)
|
||
|
||
literal_workers_intel = copy.deepcopy(intel)
|
||
literal_workers_intel["cells"][1]["inference"]["outer_workers"] = 4
|
||
literal_workers_intel["handoff"]["candidate_recipes"][0]["cell_id"] = (
|
||
experiment.make_cell_id(
|
||
literal_workers_intel, literal_workers_intel["cells"][1]
|
||
)
|
||
)
|
||
literal_workers_ryzen = copy.deepcopy(ryzen)
|
||
literal_workers_ryzen["handoff"] = copy.deepcopy(
|
||
literal_workers_intel["handoff"]
|
||
)
|
||
literal_workers_ryzen["cells"][1]["inference"].update(
|
||
outer_workers=4,
|
||
intra_op_threads=4,
|
||
)
|
||
experiment._validate_ryzen_capsule_manifest(
|
||
literal_workers_ryzen, literal_workers_intel
|
||
)
|
||
scaled_workers_ryzen = copy.deepcopy(literal_workers_ryzen)
|
||
scaled_workers_ryzen["cells"][1]["inference"].update(
|
||
outer_workers=8,
|
||
intra_op_threads=8,
|
||
)
|
||
with pytest.raises(ValueError, match="рецептом handoff"):
|
||
experiment._validate_ryzen_capsule_manifest(
|
||
scaled_workers_ryzen, literal_workers_intel
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("cell_id", [None, "not-a-sha256"])
|
||
def test_ryzen_manifest_rejects_invalid_handoff_cell_id(tmp_path, cell_id):
|
||
manifest = _finalized_handoff_manifest(tmp_path)
|
||
manifest["experiment"].update(stage="ryzen", handoff_commit="6" * 40)
|
||
for cell in manifest["cells"]:
|
||
cell["stage"] = "ryzen"
|
||
manifest["handoff"]["candidate_recipes"][0]["cell_id"] = cell_id
|
||
|
||
with pytest.raises(ValueError, match="cell_id"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_manifest_accepts_unavailable_temperature_sensor(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["machine"]["power"]["temperature_celsius"] = None
|
||
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest["machine"]["power"]["temperature_celsius"] = -1.0
|
||
with pytest.raises(ValueError, match="temperature_celsius"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_manifest_rejects_stage_boundary_thread_budget_and_non_alternating_ab(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][0]["stage"] = "ryzen"
|
||
with pytest.raises(ValueError, match="граница"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][0]["inference"]["outer_workers"] = 5
|
||
with pytest.raises(ValueError, match="бюджет"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][1]["pair_role"] = "baseline"
|
||
with pytest.raises(ValueError, match="baseline/candidate"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
second_pair = []
|
||
for source, name, position in (
|
||
(manifest["cells"][0], "W0-repeat", 3),
|
||
(manifest["cells"][1], "W2-repeat", 4),
|
||
):
|
||
cell = copy.deepcopy(source)
|
||
cell.update(name=name, schedule_position=position, pair_id="window-ab-2")
|
||
second_pair.append(cell)
|
||
manifest["cells"].extend(second_pair)
|
||
with pytest.raises(ValueError, match="начальная роль"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_ab_pair_requires_exactly_two_adjacent_schedule_positions(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][1]["schedule_position"] = 3
|
||
with pytest.raises(ValueError, match="соседними"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
for source, name, position in (
|
||
(manifest["cells"][0], "W0-repeat", 3),
|
||
(manifest["cells"][1], "W2-repeat", 4),
|
||
):
|
||
cell = copy.deepcopy(source)
|
||
cell.update(name=name, schedule_position=position)
|
||
manifest["cells"].append(cell)
|
||
with pytest.raises(ValueError, match="ровно одна пара"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("mutate", "message"),
|
||
[
|
||
(lambda manifest: manifest["cells"][0]["inference"].update(outer_workers=1.5), "целое"),
|
||
(lambda manifest: manifest["cells"][0].update(window_shift_ratio=0.3), "сетку"),
|
||
(lambda manifest: manifest["cells"][0]["clustering"].update(threshold=None), "число"),
|
||
(
|
||
lambda manifest: manifest["cells"][0].update(
|
||
recording_ids=["data-test", "data-test"]
|
||
),
|
||
"запись",
|
||
),
|
||
],
|
||
)
|
||
def test_manifest_rejects_fractional_or_ambiguous_cell_fields(
|
||
tmp_path, mutate, message
|
||
):
|
||
manifest = _manifest(tmp_path)
|
||
mutate(manifest)
|
||
|
||
with pytest.raises((TypeError, ValueError), match=message):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_manifest_rejects_enabled_qdq_without_graph(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["qdq"]["graph_artifact_id"] = None
|
||
|
||
with pytest.raises(ValueError, match="нужен граф"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("changes", "message"),
|
||
[
|
||
({"supply": "battery"}, "от сети"),
|
||
({"profile": "performance"}, "balanced"),
|
||
({"throttling": True}, "нестабильно"),
|
||
],
|
||
)
|
||
def test_manifest_rejects_unstable_power_contract(tmp_path, changes, message):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["machine"]["power"].update(changes)
|
||
|
||
with pytest.raises(ValueError, match=message):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_resolve_manifest_paths_rejects_symlink_escape(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
capsule = tmp_path / "capsule"
|
||
capsule.mkdir()
|
||
outside = tmp_path / "outside"
|
||
outside.mkdir()
|
||
(capsule / "escape").symlink_to(outside, target_is_directory=True)
|
||
manifest["artifacts"][0]["path"] = "escape/model.onnx"
|
||
|
||
with pytest.raises(ValueError, match="выходит за корень"):
|
||
experiment.resolve_manifest_paths(manifest, capsule)
|
||
|
||
|
||
def test_raw_manifest_requires_capsule_relative_paths(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
|
||
with pytest.raises(ValueError, match="относительным"):
|
||
experiment.validate_manifest(
|
||
manifest,
|
||
verify_files=False,
|
||
require_relative_paths=True,
|
||
)
|
||
|
||
|
||
def test_private_locations_inside_repository_must_be_git_ignored():
|
||
repository_root = Path(experiment.__file__).parents[2]
|
||
|
||
with pytest.raises(ValueError, match="не игнорируется Git"):
|
||
experiment._require_private_location(
|
||
repository_root / "docs" / "private-manifest.json",
|
||
"manifest",
|
||
)
|
||
|
||
experiment._require_private_location(
|
||
repository_root / "tmp" / "private-manifest.json",
|
||
"manifest",
|
||
)
|
||
|
||
|
||
def test_qdq_rejects_control_origin_and_skips_missing_material_safely(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["calibration"][0]["source_sha256"] = manifest["recordings"][0]["sha256"]
|
||
with pytest.raises(ValueError, match="контрольная запись"):
|
||
experiment.validate_manifest(manifest)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
Path(manifest["calibration"][0]["path"]).unlink()
|
||
warnings = experiment.validate_manifest(manifest)
|
||
plan = experiment.safe_plan(manifest, warnings, _environment())
|
||
|
||
assert plan["qdq_available"] is False
|
||
serialized = json.dumps(plan, ensure_ascii=False)
|
||
assert str(tmp_path) not in serialized
|
||
assert "separate calibration" not in serialized
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"cluster_sizes, dimensions, seed, noise, expected",
|
||
[((3, 17), 4, 35, 0.1, 2), ((2, 3, 15), 3, 1, 0.3, 3)],
|
||
)
|
||
def test_nme_and_eigengap_count_block_fixtures(
|
||
cluster_sizes, dimensions, seed, noise, expected
|
||
):
|
||
embeddings = experiment.synthetic_block_embeddings(
|
||
cluster_sizes, dimensions=dimensions, seed=seed, noise=noise
|
||
)
|
||
|
||
assert experiment.estimate_eigengap(embeddings)["num_clusters"] == expected
|
||
assert experiment.estimate_nme(embeddings)["num_clusters"] == expected
|
||
|
||
|
||
def test_counter_selection_handles_mixed_model_outcomes():
|
||
results = []
|
||
for model_id, nme_counts, eigengap_counts, nme_time, eigengap_time in (
|
||
("campplus", [3, 2, 2], [3, 2, 2], 12.0, 11.5),
|
||
("titanet", [3, 2, 2], [3, 3, 2], 8.0, 3.0),
|
||
("broken", [3, 3, 2], [3, 2, 3], 1.0, 1.0),
|
||
):
|
||
for counter, counts, wall in (
|
||
("nme", nme_counts, nme_time),
|
||
("eigengap", eigengap_counts, eigengap_time),
|
||
):
|
||
for num_clusters in counts:
|
||
results.append(
|
||
{
|
||
"embedding_id": model_id,
|
||
"counter": counter,
|
||
"num_clusters": num_clusters,
|
||
"wall_seconds": wall / 3,
|
||
}
|
||
)
|
||
|
||
assert experiment.select_counter(results) == {
|
||
"broken": None,
|
||
"campplus": "eigengap",
|
||
"titanet": "nme",
|
||
}
|
||
|
||
|
||
def test_window_selection_and_recipe_assembly_cover_step_only_and_deduplication():
|
||
results = [
|
||
{
|
||
"window_shift_ratio": 0.1,
|
||
"wall_seconds": 100.0,
|
||
"quality_passed": True,
|
||
"work_counters_explained": True,
|
||
},
|
||
{
|
||
"window_shift_ratio": 0.15,
|
||
"wall_seconds": 88.0,
|
||
"quality_passed": True,
|
||
"work_counters_explained": True,
|
||
},
|
||
{
|
||
"window_shift_ratio": 0.2,
|
||
"wall_seconds": 85.0,
|
||
"quality_passed": True,
|
||
"work_counters_explained": True,
|
||
},
|
||
]
|
||
assert experiment.select_working_shift(results) == 0.15
|
||
assert experiment.assemble_candidate_recipes(0.15, []) == [
|
||
{
|
||
"id": "C0",
|
||
"window_shift_ratio": 0.15,
|
||
"embedding": "wespeaker-fp32",
|
||
"inference": "sequential",
|
||
}
|
||
]
|
||
|
||
branches = [
|
||
{
|
||
"quality_passed": True,
|
||
"memory_passed": True,
|
||
"diagnostic_approved": True,
|
||
"speedup": 0.2,
|
||
"recipe": {"embedding": "wespeaker-qdq", "inference": "shared-session"},
|
||
"isa_stack": True,
|
||
"stack_order": 1,
|
||
},
|
||
{
|
||
"quality_passed": True,
|
||
"memory_passed": True,
|
||
"diagnostic_approved": True,
|
||
"speedup": 0.15,
|
||
"recipe": {"embedding": "wespeaker-qdq", "inference": "shared-session"},
|
||
"isa_stack": False,
|
||
},
|
||
]
|
||
recipes = experiment.assemble_candidate_recipes(0.15, branches)
|
||
assert len(recipes) <= 3
|
||
assert len(
|
||
{
|
||
json.dumps({k: v for k, v in item.items() if k != "id"}, sort_keys=True)
|
||
for item in recipes
|
||
}
|
||
) == len(recipes)
|
||
|
||
unapproved = copy.deepcopy(branches)
|
||
unapproved[0]["diagnostic_approved"] = False
|
||
unapproved[0]["speedup"] = 0.9
|
||
assert all(
|
||
item.get("embedding") != "wespeaker-qdq"
|
||
for item in experiment.assemble_candidate_recipes(0.15, unapproved[:1])
|
||
)
|
||
|
||
|
||
def test_schedule_preserves_raw_repeats_and_resumes_only_success(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
output_path = tmp_path / "results.json"
|
||
work_dir = tmp_path / "work"
|
||
calls = []
|
||
|
||
def runner(request):
|
||
calls.append(request["cell"]["name"])
|
||
return {"raw": request["cell"]["repetition"]}
|
||
|
||
first = experiment.run_schedule(
|
||
manifest,
|
||
output_path,
|
||
work_dir,
|
||
runner,
|
||
environment=_environment(),
|
||
guard_reader=_guard,
|
||
)
|
||
resumed_environment = _environment()
|
||
resumed_environment["swap_used_bytes"] = 4096
|
||
second = experiment.run_schedule(
|
||
manifest,
|
||
output_path,
|
||
work_dir,
|
||
runner,
|
||
environment=resumed_environment,
|
||
guard_reader=_guard,
|
||
)
|
||
|
||
assert calls == ["W0", "W2"]
|
||
assert len(first["cells"]) == 2
|
||
assert len(second["cells"]) == 2
|
||
assert all(item["result"] == {"raw": 1} for item in second["cells"])
|
||
assert all(item["guard"]["attempts"][0]["valid"] for item in second["cells"])
|
||
|
||
|
||
def test_runtime_profile_falls_back_to_powerprofilesctl(monkeypatch):
|
||
monkeypatch.setattr(experiment, "_read_optional_text", lambda _path: None)
|
||
monkeypatch.setattr(
|
||
experiment.subprocess,
|
||
"run",
|
||
lambda *args, **kwargs: experiment.subprocess.CompletedProcess(
|
||
args[0], 0, stdout="balanced\n", stderr=""
|
||
),
|
||
)
|
||
|
||
assert experiment._runtime_profile() == "balanced"
|
||
|
||
|
||
def test_runtime_guard_retries_once_and_persists_transition_evidence(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][1]["enabled"] = False
|
||
snapshots = iter(
|
||
[
|
||
_guard(),
|
||
_guard(swap_sin_bytes=101),
|
||
_guard(temperature_celsius=57.0),
|
||
_guard(temperature_celsius=58.0),
|
||
]
|
||
)
|
||
calls = []
|
||
|
||
output = experiment.run_schedule(
|
||
manifest,
|
||
tmp_path / "results.json",
|
||
tmp_path / "work",
|
||
lambda request: calls.append(request["cell"]["name"]) or {"raw": 1},
|
||
environment=_environment(),
|
||
guard_reader=lambda: next(snapshots),
|
||
)
|
||
|
||
assert calls == ["W0", "W0"]
|
||
attempts = output["cells"][0]["guard"]["attempts"]
|
||
assert [item["valid"] for item in attempts] == [False, True]
|
||
assert attempts[0]["reasons"] == ["swap_sin_bytes-changed"]
|
||
assert attempts[0]["before"]["temperature_celsius"] == 55.0
|
||
assert attempts[1]["after"]["temperature_celsius"] == 58.0
|
||
|
||
|
||
def test_runtime_guard_rejects_second_invalid_transition(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][1]["enabled"] = False
|
||
snapshots = iter(
|
||
[
|
||
_guard(),
|
||
_guard(throttle_count=11),
|
||
_guard(),
|
||
_guard(swap_sout_bytes=201),
|
||
]
|
||
)
|
||
output_path = tmp_path / "results.json"
|
||
|
||
with pytest.raises(RuntimeError, match="W0"):
|
||
experiment.run_schedule(
|
||
manifest,
|
||
output_path,
|
||
tmp_path / "work",
|
||
lambda _request: {"raw": 1},
|
||
environment=_environment(),
|
||
guard_reader=lambda: next(snapshots),
|
||
)
|
||
|
||
persisted = json.loads(output_path.read_text(encoding="utf-8"))["cells"][0]
|
||
assert persisted["status"] == "failed"
|
||
assert persisted["error_type"] == "RuntimeGuardError"
|
||
assert len(persisted["guard"]["attempts"]) == 2
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("after", "reason"),
|
||
[
|
||
(_guard(governor="performance"), "governor-mismatch"),
|
||
(_guard(turbo=False), "turbo-mismatch"),
|
||
(_guard(throttling=True), "throttling"),
|
||
(
|
||
_guard(power_limits={"package": "changed"}),
|
||
"power-limits-changed",
|
||
),
|
||
(_guard(swap_sout_bytes=201), "swap_sout_bytes-changed"),
|
||
],
|
||
)
|
||
def test_runtime_guard_checks_stable_machine_conditions(tmp_path, after, reason):
|
||
reasons = experiment.validate_runtime_guard_transition(
|
||
_manifest(tmp_path), _guard(), after
|
||
)
|
||
|
||
assert reason in reasons
|
||
|
||
|
||
def test_failed_cell_is_persisted_without_private_exception_text(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
output_path = tmp_path / "results.json"
|
||
|
||
def runner(_request):
|
||
raise RuntimeError(f"private path: {tmp_path / 'secret.wav'}")
|
||
|
||
with pytest.raises(RuntimeError, match="W0"):
|
||
experiment.run_schedule(
|
||
manifest,
|
||
output_path,
|
||
tmp_path / "work",
|
||
runner,
|
||
environment=_environment(),
|
||
guard_reader=_guard,
|
||
)
|
||
|
||
serialized = output_path.read_text(encoding="utf-8")
|
||
assert str(tmp_path) not in serialized
|
||
assert "secret.wav" not in serialized
|
||
assert "RuntimeError" in serialized
|
||
|
||
|
||
def test_residual_cluster_metrics_keep_time_and_word_units_separate():
|
||
intervals = [
|
||
{"speaker": 0, "start": 0.0, "end": 8.0},
|
||
{"speaker": 1, "start": 8.0, "end": 10.0},
|
||
{"speaker": 2, "start": 10.0, "end": 11.0},
|
||
]
|
||
words = [
|
||
{"speaker": 0},
|
||
{"speaker": 1},
|
||
{"speaker": 2},
|
||
{"speaker": 2},
|
||
{"speaker": None},
|
||
]
|
||
|
||
metrics = experiment.residual_cluster_metrics(intervals, words, 2)
|
||
|
||
assert metrics["residual_speech_time_share"] == pytest.approx(1 / 11)
|
||
assert metrics["residual_assigned_word_share"] == pytest.approx(1 / 2)
|
||
|
||
|
||
def test_capsule_verification_checks_hashes_commits_and_safe_paths(tmp_path):
|
||
payload = b"private"
|
||
payload_hash = hashlib.sha256(payload).hexdigest()
|
||
manifest_path = tmp_path / "manifest.json"
|
||
manifest_value = {
|
||
"schema_version": 3,
|
||
"artifacts": [
|
||
{
|
||
"id": "input",
|
||
"path": "data/payload.bin",
|
||
"sha256": payload_hash,
|
||
}
|
||
],
|
||
"recordings": [],
|
||
"calibration": [],
|
||
}
|
||
manifest_bytes = json.dumps(manifest_value).encode()
|
||
manifest_path.write_bytes(manifest_bytes)
|
||
capsule = {
|
||
"capsule_id": "opaque-41",
|
||
"source_commit": "3" * 64,
|
||
"public_report_sha256": "4" * 64,
|
||
"files": {
|
||
"manifest.json": hashlib.sha256(manifest_bytes).hexdigest(),
|
||
"data/payload.bin": payload_hash,
|
||
},
|
||
}
|
||
archive_path = tmp_path / "capsule.zip"
|
||
with zipfile.ZipFile(archive_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(capsule))
|
||
archive.writestr("manifest.json", manifest_bytes)
|
||
archive.writestr("data/payload.bin", payload)
|
||
archive_hash = experiment.file_sha256(archive_path)
|
||
|
||
result = experiment.verify_capsule(
|
||
archive_path,
|
||
archive_hash,
|
||
expected_manifest_path=manifest_path,
|
||
expected_source_commit="3" * 64,
|
||
expected_report_sha256="4" * 64,
|
||
expected_capsule_id="opaque-41",
|
||
)
|
||
|
||
assert result == {
|
||
"capsule_id": "opaque-41",
|
||
"source_commit": "3" * 64,
|
||
"public_report_sha256": "4" * 64,
|
||
"file_count": 2,
|
||
"manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
|
||
}
|
||
|
||
with pytest.raises(ValueError, match="ровно один"):
|
||
experiment.verify_capsule(
|
||
archive_path,
|
||
archive_hash,
|
||
expected_manifest_path=None,
|
||
)
|
||
with pytest.raises(ValueError, match="ровно один"):
|
||
experiment.verify_capsule(
|
||
archive_path,
|
||
archive_hash,
|
||
expected_manifest_path=manifest_path,
|
||
expected_ryzen_manifest=manifest_value,
|
||
)
|
||
|
||
incomplete_capsule = copy.deepcopy(capsule)
|
||
incomplete_capsule["files"].pop("data/payload.bin")
|
||
incomplete_path = tmp_path / "incomplete.zip"
|
||
with zipfile.ZipFile(incomplete_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(incomplete_capsule))
|
||
archive.writestr("manifest.json", manifest_bytes)
|
||
with pytest.raises(ValueError, match="Входной файл manifest"):
|
||
experiment.verify_capsule(
|
||
incomplete_path,
|
||
experiment.file_sha256(incomplete_path),
|
||
expected_manifest_path=manifest_path,
|
||
)
|
||
|
||
unrelated_manifest = tmp_path / "unrelated-manifest.json"
|
||
unrelated_manifest.write_bytes(b'{"schema_version":3,"other":true}\n')
|
||
with pytest.raises(ValueError, match="не совпал с manifest.json капсулы"):
|
||
experiment.verify_capsule(
|
||
archive_path,
|
||
archive_hash,
|
||
expected_manifest_path=unrelated_manifest,
|
||
)
|
||
|
||
unsafe_path = tmp_path / "unsafe.zip"
|
||
with zipfile.ZipFile(unsafe_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(capsule))
|
||
archive.writestr("manifest.json", manifest_bytes)
|
||
archive.writestr("../payload.bin", payload)
|
||
with pytest.raises(ValueError, match="небезопасный путь"):
|
||
experiment.verify_capsule(
|
||
unsafe_path,
|
||
experiment.file_sha256(unsafe_path),
|
||
expected_manifest_path=manifest_path,
|
||
)
|
||
|
||
|
||
def test_run_cli_accepts_verified_ryzen_capsule_manifest(
|
||
tmp_path, capsys, monkeypatch
|
||
):
|
||
archive_path, ryzen, intel_manifest_sha256 = _ryzen_capsule_fixture(tmp_path)
|
||
manifest_path = tmp_path / "ryzen.json"
|
||
manifest_path.write_text(json.dumps(ryzen), encoding="utf-8")
|
||
args = SimpleNamespace(
|
||
manifest=manifest_path,
|
||
output=None,
|
||
work_dir=tmp_path / "work",
|
||
capsule=archive_path,
|
||
capsule_sha256=experiment.file_sha256(archive_path),
|
||
validate_only=True,
|
||
prepare_asr=None,
|
||
)
|
||
monkeypatch.setattr(experiment, "environment_snapshot", _environment)
|
||
|
||
experiment.run_cli(args, ryzen)
|
||
|
||
plan = json.loads(capsys.readouterr().out)
|
||
assert plan["stage"] == "ryzen"
|
||
assert plan["capsule"]["manifest_sha256"] == intel_manifest_sha256
|
||
assert plan["capsule"]["file_count"] == 14
|
||
|
||
|
||
def test_verify_capsule_rejects_unlinked_ryzen_manifest(tmp_path):
|
||
archive_path, ryzen, _manifest_sha256 = _ryzen_capsule_fixture(tmp_path)
|
||
ryzen["artifacts"][0]["sha256"] = "7" * 64
|
||
|
||
with pytest.raises(ValueError, match="закрепленные входы"):
|
||
experiment.verify_capsule(
|
||
archive_path,
|
||
experiment.file_sha256(archive_path),
|
||
expected_manifest_path=None,
|
||
expected_ryzen_manifest=ryzen,
|
||
)
|
||
|
||
|
||
def test_capsule_binds_candidate_result_evidence_to_file_index(tmp_path):
|
||
input_bytes = b"input"
|
||
input_hash = hashlib.sha256(input_bytes).hexdigest()
|
||
result_bytes = b'{"aggregate":{"quality_passed":true}}'
|
||
result_hash = hashlib.sha256(result_bytes).hexdigest()
|
||
manifest_value = {
|
||
"artifacts": [
|
||
{"id": "input", "path": "data/input.bin", "sha256": input_hash}
|
||
],
|
||
"recordings": [],
|
||
"calibration": [],
|
||
"handoff": {
|
||
"candidate_recipes": [
|
||
{
|
||
"result_file": "results/candidate.json",
|
||
"result_sha256": result_hash,
|
||
}
|
||
]
|
||
}
|
||
}
|
||
manifest_path = tmp_path / "manifest.json"
|
||
manifest_bytes = json.dumps(manifest_value, separators=(",", ":")).encode()
|
||
manifest_path.write_bytes(manifest_bytes)
|
||
capsule = {
|
||
"capsule_id": "opaque-41",
|
||
"source_commit": "3" * 64,
|
||
"public_report_sha256": "4" * 64,
|
||
"files": {
|
||
"manifest.json": hashlib.sha256(manifest_bytes).hexdigest(),
|
||
"data/input.bin": input_hash,
|
||
"results/candidate.json": result_hash,
|
||
},
|
||
}
|
||
archive_path = tmp_path / "capsule.zip"
|
||
with zipfile.ZipFile(archive_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(capsule))
|
||
archive.writestr("manifest.json", manifest_bytes)
|
||
archive.writestr("data/input.bin", input_bytes)
|
||
archive.writestr("results/candidate.json", result_bytes)
|
||
|
||
experiment.verify_capsule(
|
||
archive_path,
|
||
experiment.file_sha256(archive_path),
|
||
expected_manifest_path=manifest_path,
|
||
)
|
||
|
||
broken_manifest_value = copy.deepcopy(manifest_value)
|
||
broken_manifest_value["handoff"]["candidate_recipes"][0]["result_sha256"] = (
|
||
"5" * 64
|
||
)
|
||
broken_manifest_bytes = json.dumps(
|
||
broken_manifest_value, separators=(",", ":")
|
||
).encode()
|
||
broken_manifest_path = tmp_path / "broken-manifest.json"
|
||
broken_manifest_path.write_bytes(broken_manifest_bytes)
|
||
capsule["files"]["manifest.json"] = hashlib.sha256(
|
||
broken_manifest_bytes
|
||
).hexdigest()
|
||
broken_path = tmp_path / "broken-capsule.zip"
|
||
with zipfile.ZipFile(broken_path, "w") as archive:
|
||
archive.writestr("capsule.json", json.dumps(capsule))
|
||
archive.writestr("manifest.json", broken_manifest_bytes)
|
||
archive.writestr("data/input.bin", input_bytes)
|
||
archive.writestr("results/candidate.json", result_bytes)
|
||
with pytest.raises(ValueError, match="Результат кандидата"):
|
||
experiment.verify_capsule(
|
||
broken_path,
|
||
experiment.file_sha256(broken_path),
|
||
expected_manifest_path=broken_manifest_path,
|
||
)
|
||
|
||
|
||
def test_handoff_enforces_candidate_limit_and_stop_contract(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["handoff"]["outcome"] = "stop-before-ryzen"
|
||
manifest["handoff"]["candidate_recipes"] = ["C0"]
|
||
with pytest.raises(ValueError, match="ноль кандидатов"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
manifest = _manifest(tmp_path)
|
||
manifest["handoff"]["outcome"] = "handoff"
|
||
manifest["handoff"]["candidate_recipes"] = ["C0", "C1", "C2", "C3"]
|
||
with pytest.raises(ValueError, match="трех"):
|
||
experiment.validate_manifest(manifest, verify_files=False)
|
||
|
||
|
||
def test_finalize_handoff_binds_candidate_result_and_all_gates(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][0]["phase"] = "combination"
|
||
cell_id = experiment.make_cell_id(manifest, manifest["cells"][0])
|
||
payload = {"aggregate": {"quality_passed": True}, "raw": [1, 2, 3]}
|
||
output = {
|
||
"experiment_id": experiment.make_experiment_id(manifest),
|
||
"cells": [{"cell_id": cell_id, "status": "complete", "result": payload}],
|
||
}
|
||
|
||
finalized = experiment.finalize_handoff(
|
||
manifest,
|
||
output,
|
||
["W0"],
|
||
capsule_root=tmp_path / "capsule",
|
||
gate_evidence={
|
||
"W0": {"memory_passed": True, "diagnostic_approved": True}
|
||
},
|
||
public_report_sha256="4" * 64,
|
||
capsule_id="opaque-41",
|
||
)
|
||
|
||
candidate = finalized["handoff"]["candidate_recipes"][0]
|
||
assert candidate == {
|
||
"name": "W0",
|
||
"cell_id": cell_id,
|
||
"result_file": f"results/{cell_id}.json",
|
||
"result_sha256": experiment._digest_json(payload),
|
||
"mandatory_passed": True,
|
||
"memory_passed": True,
|
||
"diagnostic_approved": True,
|
||
}
|
||
result_file = tmp_path / "capsule" / candidate["result_file"]
|
||
assert experiment.file_sha256(result_file) == candidate["result_sha256"]
|
||
assert json.loads(result_file.read_text(encoding="utf-8")) == payload
|
||
experiment.validate_manifest(finalized, verify_files=False)
|
||
|
||
finalized["handoff"]["capsule_sha256"] = "5" * 64
|
||
experiment.validate_manifest(finalized, verify_files=False)
|
||
|
||
|
||
def test_finalize_handoff_rejects_missing_diagnostic_approval(tmp_path):
|
||
manifest = _manifest(tmp_path)
|
||
manifest["cells"][0]["phase"] = "combination"
|
||
cell_id = experiment.make_cell_id(manifest, manifest["cells"][0])
|
||
output = {
|
||
"experiment_id": experiment.make_experiment_id(manifest),
|
||
"cells": [
|
||
{
|
||
"cell_id": cell_id,
|
||
"status": "complete",
|
||
"result": {"aggregate": {"quality_passed": True}},
|
||
}
|
||
],
|
||
}
|
||
|
||
with pytest.raises(ValueError, match="обязательные gates"):
|
||
experiment.finalize_handoff(
|
||
manifest,
|
||
output,
|
||
["W0"],
|
||
capsule_root=tmp_path / "capsule",
|
||
gate_evidence={
|
||
"W0": {"memory_passed": True, "diagnostic_approved": False}
|
||
},
|
||
public_report_sha256="4" * 64,
|
||
capsule_id="opaque-41",
|
||
)
|