Spaces:
Sleeping
Sleeping
File size: 3,032 Bytes
39eb052 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | import pytest
from httpx import AsyncClient, ASGITransport
from main import app
from unittest.mock import MagicMock, patch
@pytest.fixture
def mock_asr_pipeline():
with patch("main.get_asr_pipeline") as mock:
mock_pipeline = MagicMock()
mock_pipeline.return_value = {"text": "టెస్ట్ ట్రాన్స్క్రిప్షన్"}
mock.return_value = mock_pipeline
yield mock_pipeline
@pytest.fixture
def mock_ffmpeg():
with patch("subprocess.run") as mock:
mock_result = MagicMock()
mock_result.returncode = 0
mock.return_value = mock_result
yield mock
@pytest.fixture
def mock_requests_post():
with patch("requests.post") as mock:
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"id": "123", "status": "stored"}
mock.return_value = mock_resp
yield mock
@pytest.mark.asyncio
async def test_health():
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
response = await ac.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "ok"
@pytest.mark.asyncio
async def test_transcribe(mock_asr_pipeline, mock_ffmpeg):
# Create a dummy audio file
audio_content = b"fake audio content"
files = {"audio": ("test.webm", audio_content, "audio/webm")}
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
# We need to mock the file read as well if needed, but UploadFile handles it
response = await ac.post("/transcribe", files=files)
assert response.status_code == 200
assert "text" in response.json()
assert response.json()["text"] == "టెస్ట్ ట్రాన్స్క్రిప్షన్"
@pytest.mark.asyncio
async def test_transcribe_and_store(mock_asr_pipeline, mock_ffmpeg, mock_requests_post):
with patch("main.SWECHA_AUTH_TOKEN", "fake_token"):
audio_content = b"fake audio content"
files = {"audio": ("test.webm", audio_content, "audio/webm")}
data = {"title": "Test Title", "description": "Test Description"}
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
response = await ac.post("/transcribe-and-store", files=files, data=data)
assert response.status_code == 200
assert response.json()["text"] == "టెస్ట్ ట్రాన్స్క్రిప్షన్"
assert response.json()["swecha_response"]["id"] == "123"
@pytest.mark.asyncio
async def test_transcribe_empty_file():
files = {"audio": ("test.webm", b"", "audio/webm")}
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
response = await ac.post("/transcribe", files=files)
assert response.status_code == 400
assert response.json()["detail"] == "Empty audio file"
|