test: add unit tests for handler-base
- tests/conftest.py: Pytest fixtures and configuration - tests/unit/test_config.py: Settings tests - tests/unit/test_nats_client.py: NATS client tests - tests/unit/test_health.py: Health server tests - tests/unit/test_clients.py: Service client tests - pytest.ini: Pytest configuration
This commit is contained in:
76
tests/conftest.py
Normal file
76
tests/conftest.py
Normal file
@@ -0,0 +1,76 @@
|
||||
"""
|
||||
Pytest configuration and fixtures.
|
||||
"""
|
||||
import asyncio
|
||||
import os
|
||||
from typing import AsyncGenerator
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# Set test environment variables before importing handler_base
|
||||
os.environ.setdefault("NATS_URL", "nats://localhost:4222")
|
||||
os.environ.setdefault("REDIS_URL", "redis://localhost:6379")
|
||||
os.environ.setdefault("MILVUS_HOST", "localhost")
|
||||
os.environ.setdefault("OTEL_ENABLED", "false")
|
||||
os.environ.setdefault("MLFLOW_ENABLED", "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 settings():
|
||||
"""Create test settings."""
|
||||
from handler_base.config import Settings
|
||||
return Settings(
|
||||
service_name="test-service",
|
||||
service_version="1.0.0-test",
|
||||
otel_enabled=False,
|
||||
mlflow_enabled=False,
|
||||
nats_url="nats://localhost:4222",
|
||||
redis_url="redis://localhost:6379",
|
||||
milvus_host="localhost",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_httpx_client():
|
||||
"""Create a mock httpx AsyncClient."""
|
||||
client = AsyncMock()
|
||||
client.post = AsyncMock()
|
||||
client.get = AsyncMock()
|
||||
client.aclose = AsyncMock()
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_nats_message():
|
||||
"""Create a mock NATS message."""
|
||||
msg = MagicMock()
|
||||
msg.subject = "test.subject"
|
||||
msg.reply = "test.reply"
|
||||
msg.data = b'\x82\xa8query\xa5hello\xaarequest_id\xa4test' # msgpack
|
||||
return msg
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_embedding():
|
||||
"""Sample embedding vector."""
|
||||
return [0.1] * 1024
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_documents():
|
||||
"""Sample documents for testing."""
|
||||
return [
|
||||
{"text": "Python is a programming language.", "source": "doc1"},
|
||||
{"text": "Machine learning is a subset of AI.", "source": "doc2"},
|
||||
{"text": "Deep learning uses neural networks.", "source": "doc3"},
|
||||
]
|
||||
Reference in New Issue
Block a user