| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import os |
| import sys |
| from pathlib import Path |
| from types import ModuleType, SimpleNamespace |
|
|
| import numpy as np |
| import pytest |
|
|
| import capture_generated_tokens as capture_module |
| from capture_generated_tokens import DepthCodeCapture, build_capture_metadata |
| from encode_audio import NativeTokenizerUnavailableError, encode_audio |
| from native_token_compatibility import validate_native_tokens |
|
|
|
|
| def _paired_row(c0: int) -> np.ndarray: |
| row = np.array([c0, 1, 2, 3, 4, 5, 6, 7], dtype=np.int64) |
| return np.stack((row, row), axis=0) |
|
|
|
|
| def test_capture_preserves_return_and_skips_exactly_one_priming_call(): |
| rows = iter((_paired_row(10), _paired_row(11), _paired_row(12))) |
| sentinel = object() |
|
|
| def official_function(): |
| return next(rows), sentinel |
|
|
| capture = DepthCodeCapture(priming_calls=1) |
| wrapped = capture.wrap(official_function) |
| assert wrapped()[1] is sentinel |
| assert wrapped()[1] is sentinel |
| assert wrapped()[1] is sentinel |
|
|
| tokens = capture.emitted_tokens() |
| assert capture.captured_calls == 3 |
| assert tokens.shape == (2, 8) |
| assert tokens[:, 0].tolist() == [11, 12] |
|
|
|
|
| def test_patch_context_restores_official_function(monkeypatch): |
| rows = iter((_paired_row(10), _paired_row(11))) |
|
|
| def official_function(): |
| return next(rows), np.zeros(1) |
|
|
| fake_encoders = SimpleNamespace(_generate_depth_codes=official_function) |
| monkeypatch.setattr( |
| capture_module.importlib, |
| "import_module", |
| lambda _: fake_encoders, |
| ) |
|
|
| with capture_module.capture_official_depth_codes() as capture: |
| assert fake_encoders._generate_depth_codes is not official_function |
| fake_encoders._generate_depth_codes() |
| fake_encoders._generate_depth_codes() |
|
|
| assert fake_encoders._generate_depth_codes is official_function |
| assert capture.emitted_tokens().shape == (1, 8) |
|
|
|
|
| def test_run_capture_builds_pipeline_before_frame_rate_validation(monkeypatch, tmp_path): |
| events = [] |
| rows = iter((_paired_row(10), _paired_row(11))) |
|
|
| def official_depth_codes(): |
| return next(rows), np.zeros(1) |
|
|
| fake_encoders = SimpleNamespace(_generate_depth_codes=official_depth_codes) |
|
|
| class FakeGenerator: |
| def __init__(self, device): |
| events.append(("generator", device)) |
|
|
| def manual_seed(self, seed): |
| events.append(("seed", seed)) |
| return self |
|
|
| class FakePipe: |
| sampling_rate = 44_100 |
|
|
| def load_components(self, *, dtype): |
| events.append(("load_components", dtype)) |
|
|
| @property |
| def frame_rate(self): |
| events.append(("frame_rate", capture_module.FRAME_RATE_HZ)) |
| return capture_module.FRAME_RATE_HZ |
|
|
| def to(self, device): |
| events.append(("to", device)) |
|
|
| def __call__(self, **kwargs): |
| events.append(("generate", kwargs["audio_duration"])) |
| fake_encoders._generate_depth_codes() |
| fake_encoders._generate_depth_codes() |
| return np.zeros((1, 2, 160), dtype=np.float32) |
|
|
| class FakeModularPipeline: |
| @classmethod |
| def from_pretrained(cls, model): |
| events.append(("from_pretrained", model)) |
| return FakePipe() |
|
|
| fake_torch = ModuleType("torch") |
| fake_torch.bfloat16 = object() |
| fake_torch.Generator = FakeGenerator |
|
|
| fake_diffusers = ModuleType("diffusers") |
| fake_diffusers.__file__ = str(tmp_path / "diffusers" / "__init__.py") |
| fake_diffusers.ModularPipeline = FakeModularPipeline |
|
|
| fake_soundfile = ModuleType("soundfile") |
|
|
| def fake_write(path, audio, rate, *, format): |
| Path(path).write_bytes(b"RIFF-fake-wave") |
| events.append(("write_wav", Path(path), audio.shape, rate, format)) |
|
|
| fake_soundfile.write = fake_write |
|
|
| monkeypatch.setitem(sys.modules, "torch", fake_torch) |
| monkeypatch.setitem(sys.modules, "diffusers", fake_diffusers) |
| monkeypatch.setitem(sys.modules, "soundfile", fake_soundfile) |
| monkeypatch.setattr( |
| capture_module, |
| "_verified_diffusers_revision", |
| lambda _module, expected: { |
| "commit": expected, |
| "source_kind": "git_checkout", |
| "source_location": "/source", |
| "tracked_clean": True, |
| "relevant_source_path": "/source/encoders.py", |
| "relevant_source_sha256": "a" * 64, |
| }, |
| ) |
| monkeypatch.setattr( |
| capture_module, |
| "_resolve_model_snapshot", |
| lambda *_args, **_kwargs: ( |
| tmp_path / "snapshots" / "resolved-model-commit", |
| "resolved-model-commit", |
| ), |
| ) |
| monkeypatch.setattr( |
| capture_module.importlib, |
| "import_module", |
| lambda _: fake_encoders, |
| ) |
| real_replace = os.replace |
|
|
| def tracked_replace(old, new): |
| events.append(("replace", Path(new).name)) |
| real_replace(old, new) |
|
|
| monkeypatch.setattr(capture_module.os, "replace", tracked_replace) |
|
|
| args = SimpleNamespace( |
| model="model-id", |
| model_revision="requested-model-revision", |
| diffusers_revision="diffusers-revision", |
| local_files_only=True, |
| dtype="bfloat16", |
| device="cuda", |
| seed=7, |
| prompt="prompt", |
| lyrics="[instrumental]", |
| audio_duration=1.0, |
| num_inference_steps=2, |
| output_tokens=tmp_path / "tokens.npy", |
| output_wav=tmp_path / "audio.wav", |
| output_metadata=tmp_path / "metadata.json", |
| ) |
| metadata = capture_module.run_capture(args) |
|
|
| event_names = [event[0] for event in events] |
| assert event_names.index("from_pretrained") < event_names.index("load_components") |
| assert event_names.index("load_components") < event_names.index("frame_rate") |
| assert event_names.index("frame_rate") < event_names.index("to") |
| assert event_names.index("to") < event_names.index("generate") |
| replaced = [event[1] for event in events if event[0] == "replace"] |
| assert replaced == ["tokens.npy", "audio.wav", "metadata.json"] |
| assert metadata["token_shape_frames_first"] == [1, 8] |
| assert metadata["resolved_model_commit"] == "resolved-model-commit" |
| assert metadata["device"] == "cuda" |
| assert metadata["dtype"] == "bfloat16" |
| assert np.load(args.output_tokens, allow_pickle=False).shape == (1, 8) |
| saved = json.loads(args.output_metadata.read_text()) |
| assert saved["wav_reencoding_performed"] is False |
| assert saved["artifact_sha256"]["wav"] == hashlib.sha256( |
| args.output_wav.read_bytes() |
| ).hexdigest() |
| assert saved["artifact_sha256"]["tokens_npy"] == hashlib.sha256( |
| args.output_tokens.read_bytes() |
| ).hexdigest() |
|
|
| def test_capture_rejects_unpaired_cfg_rows(): |
| capture = DepthCodeCapture() |
| unpaired = _paired_row(10) |
| unpaired[1, 3] += 1 |
| wrapped = capture.wrap(lambda: (unpaired, np.zeros(1))) |
|
|
| with pytest.raises(RuntimeError, match="not identical"): |
| wrapped() |
|
|
|
|
|
|
| def test_diffusers_verifier_rejects_dirty_tracked_checkout(monkeypatch, tmp_path): |
| repository = tmp_path / "diffusers-repository" |
| (repository / ".git").mkdir(parents=True) |
| package = repository / "src" / "diffusers" |
| package.mkdir(parents=True) |
| fake_diffusers = SimpleNamespace(__file__=str(package / "__init__.py")) |
| outputs = iter( |
| ( |
| SimpleNamespace(stdout="a" * 40 + "\n"), |
| SimpleNamespace(stdout=" M src/diffusers/__init__.py\n"), |
| ) |
| ) |
| monkeypatch.setattr( |
| capture_module.subprocess, |
| "run", |
| lambda *args, **kwargs: next(outputs), |
| ) |
|
|
| with pytest.raises(RuntimeError, match="dirty tracked files"): |
| capture_module._verified_diffusers_revision(fake_diffusers, "a" * 40) |
|
|
|
|
|
|
| def _metadata_arguments() -> dict: |
| return { |
| "model_id": "model-id", |
| "requested_model_revision": "requested-model-revision", |
| "resolved_model_commit": "resolved-model-commit", |
| "diffusers_identity": { |
| "commit": "diffusers-revision", |
| "source_kind": "git_checkout", |
| "source_location": "/source", |
| "tracked_clean": True, |
| "relevant_source_path": "/source/encoders.py", |
| "relevant_source_sha256": "a" * 64, |
| }, |
| "device": "cuda", |
| "dtype": "bfloat16", |
| "prompt": "prompt", |
| "lyrics": "[instrumental]", |
| "requested_audio_duration_seconds": 1.0, |
| "num_inference_steps": 30, |
| "seed": 7, |
| "sampling_rate": 32_000, |
| "wav_sha256": "b" * 64, |
| "tokens_npy_sha256": "c" * 64, |
| "capture_script_sha256": "d" * 64, |
| } |
|
|
|
|
| def test_metadata_proves_internal_generation_not_wav_reencoding(): |
| tokens = np.stack((_paired_row(11)[0], _paired_row(12)[0])) |
| audio = np.zeros((2, 3_200), dtype=np.float32) |
| metadata = build_capture_metadata( |
| tokens=tokens, |
| audio=audio, |
| captured_calls=3, |
| priming_rows_skipped=1, |
| **_metadata_arguments(), |
| ) |
|
|
| assert metadata["capture_kind"] == "official_internal_generation_tokens" |
| assert metadata["wav_reencoding_performed"] is False |
| assert metadata["model_id"] == "model-id" |
| assert metadata["resolved_model_commit"] == "resolved-model-commit" |
| assert metadata["diffusers_source_identity"]["tracked_clean"] is True |
| assert metadata["token_shape_frames_first"] == [2, 8] |
| assert metadata["audio_shape_channels_first"] == [2, 3_200] |
| assert metadata["audio_duration_seconds"] == pytest.approx(0.1) |
|
|
|
|
| def test_metadata_rejects_call_to_frame_misalignment(): |
| with pytest.raises(ValueError, match="alignment mismatch"): |
| build_capture_metadata( |
| tokens=np.stack((_paired_row(11)[0], _paired_row(12)[0])), |
| audio=np.zeros((2, 3_200), dtype=np.float32), |
| captured_calls=2, |
| priming_rows_skipped=1, |
| **_metadata_arguments(), |
| ) |
|
|
| ARTIFACT_ENV_NAMES = ( |
| "MINIMAX_DAV_PATH", |
| "MINIMAX_GENERATED_WAV_PATH", |
| "MINIMAX_INTERNAL_TOKENS_PATH", |
| "MINIMAX_GENERATION_CAPTURE_PATH", |
| ) |
|
|
|
|
| def _artifact_paths() -> tuple[Path, Path, Path, Path] | None: |
| values = [os.environ.get(name) for name in ARTIFACT_ENV_NAMES] |
| configured = [bool(value) for value in values] |
| if any(configured) and not all(configured): |
| missing = [ |
| name for name, is_configured in zip(ARTIFACT_ENV_NAMES, configured) |
| if not is_configured |
| ] |
| raise RuntimeError( |
| "generated-capture integration requires all four artifact " |
| f"environment variables; missing: {', '.join(missing)}" |
| ) |
| if not any(configured): |
| return None |
| return tuple(Path(value) for value in values) |
|
|
|
|
| def test_partial_artifact_environment_is_rejected(monkeypatch): |
| for name in ARTIFACT_ENV_NAMES: |
| monkeypatch.delenv(name, raising=False) |
| monkeypatch.setenv("MINIMAX_DAV_PATH", "/only/dav.pth") |
|
|
| with pytest.raises(RuntimeError, match="requires all four"): |
| _artifact_paths() |
|
|
| def test_real_generated_sample_has_valid_internal_tokens_but_wav_encoding_is_blocked(): |
| paths = _artifact_paths() |
| if paths is None: |
| pytest.skip("set all four generated-capture artifact environment variables") |
| dav_path, wav_path, tokens_path, metadata_path = paths |
|
|
| import soundfile as sf |
|
|
| tokens = np.load(tokens_path, allow_pickle=False) |
| validated = validate_native_tokens(tokens, layout="frames_first") |
| metadata = json.loads(metadata_path.read_text(encoding="utf-8")) |
| audio_info = sf.info(wav_path) |
|
|
| required = { |
| "capture_kind", |
| "wav_reencoding_performed", |
| "model_id", |
| "requested_model_revision", |
| "resolved_model_commit", |
| "diffusers_revision", |
| "diffusers_source_identity", |
| "capture_script_sha256", |
| "device", |
| "dtype", |
| "prompt", |
| "lyrics", |
| "requested_audio_duration_seconds", |
| "num_inference_steps", |
| "seed", |
| "artifact_sha256", |
| } |
| assert required <= metadata.keys() |
| assert metadata["capture_kind"] == "official_internal_generation_tokens" |
| assert metadata["wav_reencoding_performed"] is False |
| assert metadata["model_id"] == "MiniMaxAI/MiniMax-Music3" |
| assert metadata["requested_model_revision"] == ( |
| "fbdf52fbaaca799592917417eb05f1899f1255ec" |
| ) |
| assert metadata["resolved_model_commit"] == ( |
| "fbdf52fbaaca799592917417eb05f1899f1255ec" |
| ) |
| assert metadata["diffusers_revision"] == ( |
| "90b4e34e79a86ec5e7f2437634fe95ecd2108796" |
| ) |
| source_identity = metadata["diffusers_source_identity"] |
| assert source_identity["commit"] == metadata["diffusers_revision"] |
| assert source_identity["source_kind"] in {"git_checkout", "vcs_install"} |
| if source_identity["source_kind"] == "git_checkout": |
| assert source_identity["tracked_clean"] is True |
| relevant_source = Path(source_identity["relevant_source_path"]) |
| assert source_identity["relevant_source_sha256"] == hashlib.sha256( |
| relevant_source.read_bytes() |
| ).hexdigest() |
| assert metadata["capture_script_sha256"] == hashlib.sha256( |
| Path(capture_module.__file__).read_bytes() |
| ).hexdigest() |
| assert metadata["device"] == "cuda" |
| assert metadata["dtype"] == "bfloat16" |
| assert metadata["prompt"] == ( |
| "Instrumental French house, 126 BPM, E minor, filtered disco loop, " |
| "punchy kick and warm bass." |
| ) |
| assert metadata["lyrics"] == "[instrumental]" |
| assert metadata["requested_audio_duration_seconds"] == 1.0 |
| assert metadata["num_inference_steps"] == 30 |
| assert metadata["seed"] == 7 |
| assert metadata["artifact_sha256"]["wav"] == hashlib.sha256( |
| wav_path.read_bytes() |
| ).hexdigest() |
| assert metadata["artifact_sha256"]["tokens_npy"] == hashlib.sha256( |
| tokens_path.read_bytes() |
| ).hexdigest() |
|
|
| assert validated.min(axis=0).tolist() == [1012, 95, 25, 42, 83, 2, 3, 67] |
| assert validated.max(axis=0).tolist() == [ |
| 16163, 984, 1005, 950, 941, 967, 984, 957 |
| ] |
| assert validated.shape == (25, 8) |
| assert metadata["captured_calls_including_priming"] == 26 |
| assert metadata["priming_rows_skipped"] == 1 |
| assert metadata["token_shape_frames_first"] == [25, 8] |
| assert metadata["token_mins"] == validated.min(axis=0).tolist() |
| assert metadata["token_maxs"] == validated.max(axis=0).tolist() |
| assert audio_info.samplerate == 44_100 |
| assert audio_info.channels == 2 |
| assert metadata["sampling_rate"] == 44_100 |
| assert metadata["audio_shape_channels_first"] == [2, 44_032] |
| assert metadata["audio_duration_seconds"] == pytest.approx(0.9984580498866213) |
| assert audio_info.frames == 44_032 |
|
|
| |
| |
| with pytest.raises(NativeTokenizerUnavailableError): |
| encode_audio(wav_path, dav_path=dav_path) |
|
|
