tests/test_batch.py

import csv
import hashlib
import json
import math
import struct
import wave

from autoace_audio.batch import ManifestEntry, parse_manifest_for_inference, process_batch
from autoace_audio.pipeline import prepare_audio


def write_wav(path, frequency=440, duration=0.15, sample_rate=16_000):
    with wave.open(str(path), "wb") as output:
        output.setnchannels(1)
        output.setsampwidth(2)
        output.setframerate(sample_rate)
        frames = bytearray()
        for index in range(int(duration * sample_rate)):
            sample = int(8_000 * math.sin(2 * math.pi * frequency * index / sample_rate))
            frames.extend(struct.pack("<h", sample))
        output.writeframes(frames)


def write_manifest(path, label_value):
    with path.open("w", encoding="utf-8", newline="") as stream:
        writer = csv.DictWriter(stream, fieldnames=("name", "result_json"))
        writer.writeheader()
        writer.writerow({"name": "call.wav", "result_json": label_value})


def test_batch_isolates_file_failures_and_continues(tmp_path):
    write_wav(tmp_path / "first.wav", 440)
    (tmp_path / "broken.mp3").write_bytes(b"not audio")
    write_wav(tmp_path / "third.wav", 660)
    entries = (
        ManifestEntry("first.wav"),
        ManifestEntry("broken.mp3"),
        ManifestEntry("third.wav"),
    )

    report = process_batch(entries, tmp_path, prepare_audio)

    assert [item.status for item in report.items] == ["ok", "error", "ok"]
    assert report.success_count == 2
    assert report.failure_count == 1
    assert report.items[1].failure is not None
    assert report.items[1].failure.code == "audio_decode_failed"
    assert str(tmp_path) not in report.items[1].failure.message


def test_missing_file_is_an_isolated_error(tmp_path):
    write_wav(tmp_path / "present.wav")
    entries = (ManifestEntry("missing.wav"), ManifestEntry("present.wav"))

    report = process_batch(entries, tmp_path, lambda path: path.name)

    assert [item.status for item in report.items] == ["error", "ok"]
    assert report.items[0].failure.code == "missing_file"


def test_inference_results_do_not_change_when_labels_are_removed_or_altered(tmp_path):
    audio = tmp_path / "call.wav"
    write_wav(audio)
    empty_manifest = tmp_path / "empty.csv"
    altered_manifest = tmp_path / "altered.csv"
    write_manifest(empty_manifest, "")
    write_manifest(
        altered_manifest,
        json.dumps(
            {
                "emotional_tone": "distressed",
                "emotional_intensity": "high",
                "background_noise_present": True,
                "background_noise_type": "television",
                "background_noise_severity": "high",
                "audio_quality": "severely_impaired",
                "speaker_overlap_present": True,
                "long_silence_present": True,
                "confidence": 1.0,
            }
        ),
    )

    def inference_stub(path):
        return hashlib.sha256(path.read_bytes()).hexdigest()

    without_labels = process_batch(
        parse_manifest_for_inference(empty_manifest), tmp_path, inference_stub
    )
    with_altered_labels = process_batch(
        parse_manifest_for_inference(altered_manifest), tmp_path, inference_stub
    )

    assert without_labels.items == with_altered_labels.items
    assert not hasattr(parse_manifest_for_inference(altered_manifest)[0], "result_json")