qzl
ad06fe7de4
Consolidate backend modules/tests under the backend package while syncing Supabase compose/env config and related plans.
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import socket
|
|
|
|
import pytest
|
|
|
|
from core.config.settings import Settings
|
|
from services.base.qdrant import QdrantService
|
|
from services.base.redis import RedisService
|
|
|
|
|
|
def _can_connect(host: str, port: int) -> bool:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.settimeout(0.2)
|
|
return sock.connect_ex((host, port)) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_redis_service_health_check_integration() -> None:
|
|
host = "127.0.0.1"
|
|
port = 6379
|
|
if not _can_connect(host, port):
|
|
pytest.skip("Redis is not running on localhost:6379")
|
|
|
|
config = Settings()
|
|
settings = config.redis.model_copy(update={"host": host, "port": port})
|
|
service = RedisService(settings=settings)
|
|
|
|
assert await service.initialize() is True
|
|
health = await service.health_check()
|
|
assert health["status"] == "healthy"
|
|
assert await service.close() is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_qdrant_service_health_check_integration() -> None:
|
|
host = "127.0.0.1"
|
|
port = 6333
|
|
if not _can_connect(host, port):
|
|
pytest.skip("Qdrant is not running on localhost:6333")
|
|
|
|
config = Settings()
|
|
settings = config.qdrant.model_copy(update={"host": host, "port": port})
|
|
service = QdrantService(settings=settings)
|
|
|
|
assert await service.initialize() is True
|
|
health = await service.health_check()
|
|
assert health["status"] == "healthy"
|
|
assert await service.close() is True
|