test_deepinfra.py

tests/test_deepinfra.py

1. """The DeepInfra engine against a local stand-in server: request shape, errors, normalization."""
3. import json
4. import threading
5. from http.server import BaseHTTPRequestHandler, HTTPServer
7. import pytest
9. from transcription_pipeline.engines import deepinfra
12. class Stub:
13. status = 200
14. body = {}
15. seen = {}
18. class Handler(BaseHTTPRequestHandler):
19. def do_POST(self):
20. n = int(self.headers["Content-Length"])
21. Stub.seen = {"auth": self.headers["Authorization"], "ctype": self.headers["Content-Type"],
22. "body": self.rfile.read(n), "path": self.path}
23. payload = json.dumps(Stub.body).encode()
24. self.send_response(Stub.status)
25. self.send_header("Content-Type", "application/json")
26. self.send_header("Content-Length", str(len(payload)))
27. self.end_headers()
28. self.wfile.write(payload)
30. def log_message(self, *a):
31. pass
34. @pytest.fixture
35. def server():
36. srv = HTTPServer(("127.0.0.1", 0), Handler)
37. threading.Thread(target=srv.serve_forever, daemon=True).start()
38. yield f"http://127.0.0.1:{srv.server_address[1]}/v1/inference/openai/whisper-large-v3-turbo"
39. srv.shutdown()
42. @pytest.fixture
43. def clip(tmp_path):
44. p = tmp_path / "call.flac"
45. p.write_bytes(b"fLaC-not-really")
46. return p
49. def test_request_shape_and_segments(server, clip):
50. Stub.status, Stub.body = 200, {"text": " Hi there. Bye.", "language": "en", "segments": [
51. {"start": 0.0, "end": 1.2, "text": " Hi there."}, {"start": 1.2, "end": 2.0, "text": " "},
52. {"start": 2.0, "end": 2.6, "text": " Bye."}]}
53. out = deepinfra.transcribe(clip, "tok-123", language="en", api_url=server)
54. assert Stub.seen["auth"] == "bearer tok-123"
55. assert Stub.seen["ctype"].startswith("multipart/form-data; boundary=")
56. body = Stub.seen["body"]
57. assert b'name="audio"; filename="call.flac"' in body and b"fLaC-not-really" in body
58. assert b'name="language"\r\n\r\nen\r\n' in body
59. assert out["segments"] == [{"start": 0.0, "end": 1.2, "text": "Hi there."},
60. {"start": 2.0, "end": 2.6, "text": "Bye."}]
61. assert out["language"] == "en" and "DeepInfra" in out["model"]
64. def test_text_only_response_becomes_one_segment(server, clip):
65. Stub.status, Stub.body = 200, {"text": "Just text.", "segments": None}
66. out = deepinfra.transcribe(clip, "t", api_url=server)
67. assert out["segments"] == [{"start": 0.0, "end": 0.0, "text": "Just text."}]
68. assert b'name="language"' not in Stub.seen["body"]
71. @pytest.mark.parametrize("code,needle", [(401, "rejected the token"), (403, "rejected the token"),
72. (429, "out of credit"), (413, "too large"),
73. (500, "DeepInfra error 500")])
74. def test_error_mapping(server, clip, code, needle):
75. Stub.status, Stub.body = code, {"detail": "nope"}
76. with pytest.raises(deepinfra.EngineError, match=needle):
77. deepinfra.transcribe(clip, "t", api_url=server)
80. def test_missing_token(clip):
81. with pytest.raises(deepinfra.EngineError, match="No DeepInfra token"):
82. deepinfra.transcribe(clip, None)
85. def test_unreachable(clip):
86. with pytest.raises(deepinfra.EngineError, match="Could not reach"):
87. deepinfra.transcribe(clip, "t", api_url="http://127.0.0.1:9/x", timeout=2)