Files
stt-module/tests/conftest.py
Billy D. 55cd657364
Some checks failed
CI / Docker Build & Push (push) Has been skipped
CI / Notify (push) Successful in 1s
CI / Lint (push) Failing after 9s
CI / Test (push) Successful in 50s
CI / Release (push) Has been skipped
feat: CI pipeline, lint fixes, and Renovate config
- pyproject.toml with ruff/pytest config (setuptools<81 pin)
- Full test suite (26 tests)
- Gitea Actions CI (lint, test, docker, notify)
- Ruff lint/format fixes across source files
- Renovate config for automated dependency updates

Ref: ADR-0057
2026-02-13 15:33:35 -05:00

72 lines
1.7 KiB
Python

"""
Pytest configuration and fixtures for stt-module tests.
"""
import asyncio
import base64
import os
import struct
from unittest.mock import AsyncMock, MagicMock
import pytest
# Set test environment variables before importing
os.environ.setdefault("NATS_URL", "nats://localhost:4222")
os.environ.setdefault("WHISPER_URL", "http://localhost:8000")
os.environ.setdefault("OTEL_ENABLED", "false")
os.environ.setdefault("STT_ENABLE_VAD", "false")
@pytest.fixture(scope="session")
def event_loop():
"""Create event loop for async tests."""
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture
def silent_pcm_bytes():
"""16-bit PCM silence (1000 samples at 16kHz = 62.5ms)."""
return bytes([0x00] * 2000)
@pytest.fixture
def noisy_pcm_bytes():
"""16-bit PCM with some signal (sine-like pattern)."""
samples = []
for i in range(1000):
val = int(16000 * ((i % 50) / 50.0 - 0.5))
samples.append(struct.pack("<h", val))
return b"".join(samples)
@pytest.fixture
def sample_audio_b64(silent_pcm_bytes):
"""Base64-encoded silent PCM audio."""
return base64.b64encode(silent_pcm_bytes).decode()
@pytest.fixture
def mock_nats():
"""Mock NATS connection."""
nc = AsyncMock()
nc.publish = AsyncMock()
nc.subscribe = AsyncMock()
nc.close = AsyncMock()
nc.jetstream = MagicMock(return_value=AsyncMock())
return nc
@pytest.fixture
def mock_http_client():
"""Mock httpx async client."""
client = AsyncMock()
response = MagicMock()
response.status_code = 200
response.json.return_value = {"text": "Hello world"}
response.raise_for_status = MagicMock()
client.post = AsyncMock(return_value=response)
client.aclose = AsyncMock()
return client