Clean up audio probe temp files
This commit is contained in:
parent
14c26d037a
commit
13fe8cef17
|
|
@ -191,14 +191,15 @@ class AudioProcessor:
|
|||
return f"{filename}{suffix}"
|
||||
|
||||
def _probe_audio_duration_seconds(self, content: bytes, suffix: str) -> float:
|
||||
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
|
||||
temp_file.write(content)
|
||||
temp_path = Path(temp_file.name)
|
||||
|
||||
temp_path: Path | None = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp_file:
|
||||
temp_path = Path(temp_file.name)
|
||||
temp_file.write(content)
|
||||
return self.probe_audio_duration_seconds_from_path(temp_path)
|
||||
finally:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
if temp_path is not None:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
|
||||
def probe_audio_duration_seconds_from_path(self, path: Path) -> float:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -30,6 +30,37 @@ def test_audio_defaults_use_openai_whisper_without_backup():
|
|||
assert settings.AUDIO.MODEL == "whisper-1"
|
||||
|
||||
|
||||
def test_probe_audio_duration_cleans_up_temp_file_on_write_failure():
|
||||
processor = AudioProcessor()
|
||||
|
||||
class FailingTempFile:
|
||||
name: str = "/tmp/test-audio-probe.mp3"
|
||||
|
||||
def __enter__(self) -> "FailingTempFile":
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
self,
|
||||
exc_type: type[BaseException] | None,
|
||||
exc: BaseException | None,
|
||||
tb: TracebackType | None,
|
||||
) -> bool:
|
||||
return False
|
||||
|
||||
def write(self, _content: bytes) -> int:
|
||||
raise OSError("disk full")
|
||||
|
||||
with (
|
||||
patch("src.utils.files.tempfile.NamedTemporaryFile", return_value=FailingTempFile()),
|
||||
patch("src.utils.files.Path.unlink") as mock_unlink,
|
||||
pytest.raises(OSError, match="disk full"),
|
||||
):
|
||||
probe_audio_duration_seconds = getattr(processor, "_probe_audio_duration_seconds")
|
||||
probe_audio_duration_seconds(b"audio-bytes", ".mp3")
|
||||
|
||||
mock_unlink.assert_called_once_with(missing_ok=True)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_audio_upload_requires_openai_client_before_processing():
|
||||
file = UploadFile(
|
||||
|
|
|
|||
Loading…
Reference in New Issue