Spaces:
Running
Running
File size: 5,949 Bytes
76db545 | 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """
Integration tests for the FastAPI endpoints.
Uses httpx.AsyncClient with a mocked transcriber so GPU hardware is not required.
"""
from __future__ import annotations
import io
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
def _make_mock_app():
"""Create a test FastAPI app with mocked model state."""
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.api.routes import health, iot, transcribe
from src.engine.transcriber import TranscriptionResult
app = FastAPI()
app.include_router(health.router, prefix="/api/v1")
app.include_router(transcribe.router, prefix="/api/v1")
app.include_router(iot.router, prefix="/api/v1")
# Mock adapter manager
mock_adapter_manager = MagicMock()
mock_adapter_manager.get_active.return_value = "bam"
mock_adapter_manager.list_available.return_value = ["bam", "ful"]
mock_adapter_manager.list_loaded.return_value = ["bam"]
# Mock transcriber
mock_transcriber = MagicMock()
mock_transcriber.transcribe_file.return_value = TranscriptionResult(
text="bunding nɔgɔ foro",
language="bam",
duration_s=3.0,
processing_time_ms=250,
)
# Mock sensor bridge
mock_sensor_bridge = MagicMock()
from src.iot.sensor_bridge import SensorData
from datetime import datetime
mock_sensor_bridge.fetch = AsyncMock(
return_value=SensorData(
sensor_type="soil",
values={"moisture_pct": 42.0, "ph": 6.5},
timestamp=datetime.utcnow().isoformat(),
)
)
app.state.adapter_manager = mock_adapter_manager
app.state.transcriber = mock_transcriber
app.state.sensor_bridge = mock_sensor_bridge
return app, TestClient(app)
class TestHealthEndpoint:
def setup_method(self):
self.app, self.client = _make_mock_app()
def test_health_returns_200(self):
response = self.client.get("/api/v1/health")
assert response.status_code == 200
def test_health_response_structure(self):
data = self.client.get("/api/v1/health").json()
assert "status" in data
assert "model_loaded" in data
assert "adapters_available" in data
def test_health_model_loaded_true(self):
data = self.client.get("/api/v1/health").json()
assert data["model_loaded"] is True
def test_health_active_adapter(self):
data = self.client.get("/api/v1/health").json()
assert data["active_adapter"] == "bam"
class TestTranscribeEndpoint:
def setup_method(self):
self.app, self.client = _make_mock_app()
def _wav_bytes(self) -> bytes:
"""Minimal valid WAV file bytes for testing."""
import wave
import struct
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(16000)
wf.writeframes(struct.pack("<" + "h" * 160, *([0] * 160)))
return buf.getvalue()
def test_transcribe_returns_200(self):
response = self.client.post(
"/api/v1/transcribe",
data={"language": "bam"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
)
assert response.status_code == 200
def test_transcribe_response_has_text(self):
data = self.client.post(
"/api/v1/transcribe",
data={"language": "bam"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
).json()
assert "text" in data
assert isinstance(data["text"], str)
def test_invalid_language_returns_422(self):
response = self.client.post(
"/api/v1/transcribe",
data={"language": "xyz"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
)
assert response.status_code == 422
def test_unsupported_file_type_returns_422(self):
response = self.client.post(
"/api/v1/transcribe",
data={"language": "bam"},
files={"audio_file": ("test.txt", b"not audio", "text/plain")},
)
assert response.status_code == 422
class TestIoTQueryEndpoint:
def setup_method(self):
self.app, self.client = _make_mock_app()
def _wav_bytes(self) -> bytes:
import io
import struct
import wave
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(16000)
wf.writeframes(struct.pack("<" + "h" * 160, *([0] * 160)))
return buf.getvalue()
def test_query_returns_200(self):
response = self.client.post(
"/api/v1/query",
data={"language": "bam", "field_id": "field_001"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
)
assert response.status_code == 200
def test_query_response_structure(self):
data = self.client.post(
"/api/v1/query",
data={"language": "bam"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
).json()
assert "transcription" in data
assert "intent" in data
assert "sensor_data" in data
assert "voice_response" in data
def test_query_voice_response_is_french(self):
data = self.client.post(
"/api/v1/query",
data={"language": "bam"},
files={"audio_file": ("test.wav", self._wav_bytes(), "audio/wav")},
).json()
# French response should contain at least one French word
response_text = data["voice_response"]
french_indicators = ["du", "de", "le", "la", "les", "et", "Humidité", "sol", "pH"]
assert any(word in response_text for word in french_indicators)
|