Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 00c95cdd53 | |||
| 7830ee9355 |
@@ -1,4 +1,5 @@
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
@@ -11,6 +12,20 @@ import server
|
||||
client = TestClient(server.app)
|
||||
|
||||
|
||||
class FakeTemporaryDirectory:
|
||||
def __init__(self, root: Path, prefix: str, created_paths: list[Path]) -> None:
|
||||
self.path = root / f"{prefix}{len(created_paths)}"
|
||||
self.created_paths = created_paths
|
||||
|
||||
def __enter__(self) -> str:
|
||||
self.path.mkdir(parents=True, exist_ok=False)
|
||||
self.created_paths.append(self.path)
|
||||
return str(self.path)
|
||||
|
||||
def __exit__(self, exc_type, exc, tb) -> None:
|
||||
shutil.rmtree(self.path, ignore_errors=True)
|
||||
|
||||
|
||||
def test_validate_output_format_rejects_unknown() -> None:
|
||||
try:
|
||||
server.validate_output_format("docx")
|
||||
@@ -21,6 +36,13 @@ def test_validate_output_format_rejects_unknown() -> None:
|
||||
|
||||
|
||||
def test_transcriptions_returns_generated_artifact(monkeypatch, tmp_path: Path) -> None:
|
||||
created_paths: list[Path] = []
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"TemporaryDirectory",
|
||||
lambda prefix="": FakeTemporaryDirectory(tmp_path, prefix, created_paths),
|
||||
)
|
||||
|
||||
def fake_run(command: list[str], check: bool, capture_output: bool, text: bool, timeout: int):
|
||||
output_dir = Path(command[command.index("--output_dir") + 1])
|
||||
(output_dir / "clip.txt").write_text("hello world", encoding="utf-8")
|
||||
@@ -43,9 +65,18 @@ def test_transcriptions_returns_generated_artifact(monkeypatch, tmp_path: Path)
|
||||
assert response.status_code == 200
|
||||
assert response.text == "hello world"
|
||||
assert response.headers["x-whisper-output-format"] == "txt"
|
||||
assert created_paths
|
||||
assert all(not path.exists() for path in created_paths)
|
||||
|
||||
|
||||
def test_transcriptions_maps_subprocess_failure(monkeypatch) -> None:
|
||||
def test_transcriptions_maps_subprocess_failure(monkeypatch, tmp_path: Path) -> None:
|
||||
created_paths: list[Path] = []
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"TemporaryDirectory",
|
||||
lambda prefix="": FakeTemporaryDirectory(tmp_path, prefix, created_paths),
|
||||
)
|
||||
|
||||
def fake_run(command: list[str], check: bool, capture_output: bool, text: bool, timeout: int):
|
||||
class Result:
|
||||
returncode = 1
|
||||
@@ -64,9 +95,18 @@ def test_transcriptions_maps_subprocess_failure(monkeypatch) -> None:
|
||||
|
||||
assert response.status_code == 502
|
||||
assert response.json()["detail"] == "bad whisper day"
|
||||
assert created_paths
|
||||
assert all(not path.exists() for path in created_paths)
|
||||
|
||||
|
||||
def test_transcriptions_maps_subprocess_timeout(monkeypatch) -> None:
|
||||
def test_transcriptions_maps_subprocess_timeout(monkeypatch, tmp_path: Path) -> None:
|
||||
created_paths: list[Path] = []
|
||||
monkeypatch.setattr(
|
||||
server,
|
||||
"TemporaryDirectory",
|
||||
lambda prefix="": FakeTemporaryDirectory(tmp_path, prefix, created_paths),
|
||||
)
|
||||
|
||||
def fake_run(command: list[str], check: bool, capture_output: bool, text: bool, timeout: int):
|
||||
raise server.subprocess.TimeoutExpired(cmd=command, timeout=timeout)
|
||||
|
||||
@@ -83,3 +123,5 @@ def test_transcriptions_maps_subprocess_timeout(monkeypatch) -> None:
|
||||
response.json()["detail"]
|
||||
== f"Whisper CLI timed out after {server.WHISPER_PROCESS_TIMEOUT_SECONDS}s and was terminated."
|
||||
)
|
||||
assert created_paths
|
||||
assert all(not path.exists() for path in created_paths)
|
||||
|
||||
+49
-1
@@ -1,14 +1,15 @@
|
||||
import os
|
||||
from argparse import Namespace
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
|
||||
|
||||
import main
|
||||
import httpx
|
||||
|
||||
|
||||
def test_resolve_server_from_env(monkeypatch) -> None:
|
||||
@@ -61,3 +62,50 @@ def test_format_request_error_network_failure() -> None:
|
||||
message
|
||||
== "Request to http://localhost:8000/transcriptions failed: connection refused"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeResponse:
|
||||
content: bytes
|
||||
text: str
|
||||
|
||||
def raise_for_status(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, response: FakeResponse) -> None:
|
||||
self.response = response
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb) -> None:
|
||||
return None
|
||||
|
||||
def post(self, endpoint: str, data: dict[str, str], files: dict[str, tuple[str, object, str]]):
|
||||
return self.response
|
||||
|
||||
|
||||
def test_main_writes_transcript_to_file(monkeypatch, tmp_path: Path, capsys) -> None:
|
||||
input_file = tmp_path / "clip.wav"
|
||||
input_file.write_bytes(b"audio")
|
||||
destination = tmp_path / "saved" / "clip.txt"
|
||||
|
||||
monkeypatch.setenv("WHISPER_REMOTE", "http://localhost:8000")
|
||||
monkeypatch.setattr(
|
||||
main.httpx,
|
||||
"Client",
|
||||
lambda timeout: FakeClient(FakeResponse(b"hello world", "hello world")),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
["whisper-remote", str(input_file), "--model", "base", "--to-file", str(destination)],
|
||||
)
|
||||
|
||||
exit_code = main.main()
|
||||
|
||||
assert exit_code == 0
|
||||
assert destination.read_bytes() == b"hello world"
|
||||
assert capsys.readouterr().out == f"{destination}\n"
|
||||
|
||||
Reference in New Issue
Block a user