54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
import pytest
|
|
import respx
|
|
import httpx
|
|
|
|
from app.service_proxy import fetch_service_logs
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_fetch_service_logs_filters_by_path_prefix():
|
|
# lotto 컨테이너 응답: lotto + personal 섞임
|
|
respx.get("http://lotto:8000/logs/recent").mock(
|
|
return_value=httpx.Response(200, json={
|
|
"logs": [
|
|
{"ts": "2026-05-28T10:00:00Z", "source": "access",
|
|
"method": "GET", "path": "/api/lotto/recommend",
|
|
"status": 200, "ms": 12,
|
|
"message": "GET /api/lotto/recommend → 200 (12ms)"},
|
|
{"ts": "2026-05-28T10:00:01Z", "source": "access",
|
|
"method": "GET", "path": "/api/blog/posts",
|
|
"status": 200, "ms": 5,
|
|
"message": "GET /api/blog/posts → 200 (5ms)"},
|
|
{"ts": "2026-05-28T10:00:02Z", "source": "log",
|
|
"logger": "lotto", "level": "info",
|
|
"message": "성과 통계 캐시 갱신"},
|
|
]
|
|
})
|
|
)
|
|
|
|
result = await fetch_service_logs("lotto", limit=50)
|
|
# lotto path 와 모든 log 이벤트만 통과
|
|
paths = [x.get("path") for x in result]
|
|
assert "/api/lotto/recommend" in paths
|
|
assert "/api/blog/posts" not in paths
|
|
# 비즈니스 로그도 포함
|
|
assert any(x["source"] == "log" and x["message"] == "성과 통계 캐시 갱신"
|
|
for x in result)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_service_logs_unknown_agent_returns_empty():
|
|
result = await fetch_service_logs("nonexistent", limit=50)
|
|
assert result == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_fetch_service_logs_handles_connection_error():
|
|
respx.get("http://lotto:8000/logs/recent").mock(
|
|
side_effect=httpx.ConnectError("connection refused")
|
|
)
|
|
result = await fetch_service_logs("lotto", limit=50)
|
|
assert result == []
|