130 lines
3.2 KiB
Python
130 lines
3.2 KiB
Python
import logging
|
|
import time
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from _shared.access_log import (
|
|
AccessLogMiddleware,
|
|
BufferLogHandler,
|
|
router as logs_router,
|
|
install,
|
|
_BUFFER,
|
|
)
|
|
|
|
|
|
def _reset_buffer():
|
|
_BUFFER.clear()
|
|
|
|
|
|
def test_access_middleware_records_request():
|
|
_reset_buffer()
|
|
app = FastAPI()
|
|
app.add_middleware(AccessLogMiddleware)
|
|
|
|
@app.get("/api/lotto/recommend")
|
|
def recommend():
|
|
return {"ok": True}
|
|
|
|
client = TestClient(app)
|
|
client.get("/api/lotto/recommend")
|
|
|
|
items = [x for x in _BUFFER if x["source"] == "access"]
|
|
assert len(items) == 1
|
|
assert items[0]["method"] == "GET"
|
|
assert items[0]["path"] == "/api/lotto/recommend"
|
|
assert items[0]["status"] == 200
|
|
assert items[0]["ms"] >= 0
|
|
|
|
|
|
def test_access_middleware_skips_health():
|
|
_reset_buffer()
|
|
app = FastAPI()
|
|
app.add_middleware(AccessLogMiddleware)
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"ok": True}
|
|
|
|
client = TestClient(app)
|
|
client.get("/health")
|
|
|
|
items = [x for x in _BUFFER if x["source"] == "access"]
|
|
assert items == []
|
|
|
|
|
|
def test_access_middleware_skips_options():
|
|
_reset_buffer()
|
|
app = FastAPI()
|
|
app.add_middleware(AccessLogMiddleware)
|
|
|
|
@app.get("/api/lotto/recommend")
|
|
def recommend():
|
|
return {"ok": True}
|
|
|
|
client = TestClient(app)
|
|
client.options("/api/lotto/recommend")
|
|
|
|
items = [x for x in _BUFFER if x["source"] == "access"]
|
|
assert items == []
|
|
|
|
|
|
def test_buffer_log_handler_captures_logger_info():
|
|
_reset_buffer()
|
|
root = logging.getLogger("")
|
|
handler = BufferLogHandler()
|
|
root.addHandler(handler)
|
|
try:
|
|
lg = logging.getLogger("lotto.test")
|
|
lg.setLevel(logging.INFO)
|
|
lg.info("뉴스 스크래핑 완료: 국내 12건")
|
|
finally:
|
|
root.removeHandler(handler)
|
|
|
|
items = [x for x in _BUFFER if x["source"] == "log"]
|
|
assert len(items) == 1
|
|
assert items[0]["message"] == "뉴스 스크래핑 완료: 국내 12건"
|
|
assert items[0]["level"] == "info"
|
|
assert items[0]["logger"] == "lotto.test"
|
|
|
|
|
|
def test_logs_recent_endpoint_returns_recent_items():
|
|
_reset_buffer()
|
|
app = FastAPI()
|
|
install(app)
|
|
|
|
@app.get("/api/lotto/recommend")
|
|
def recommend():
|
|
return {"ok": True}
|
|
|
|
client = TestClient(app)
|
|
client.get("/api/lotto/recommend")
|
|
client.get("/api/lotto/recommend")
|
|
client.get("/health") # 제외되어야 함
|
|
|
|
resp = client.get("/logs/recent")
|
|
assert resp.status_code == 200
|
|
logs = resp.json()["logs"]
|
|
access_items = [x for x in logs if x["source"] == "access"]
|
|
assert len(access_items) == 2
|
|
|
|
|
|
def test_logs_recent_with_since_filter():
|
|
_reset_buffer()
|
|
app = FastAPI()
|
|
install(app)
|
|
|
|
@app.get("/api/lotto/recommend")
|
|
def recommend():
|
|
return {"ok": True}
|
|
|
|
client = TestClient(app)
|
|
client.get("/api/lotto/recommend")
|
|
time.sleep(0.01)
|
|
cursor_resp = client.get("/logs/recent")
|
|
cursor_ts = cursor_resp.json()["logs"][-1]["ts"]
|
|
client.get("/api/lotto/recommend")
|
|
|
|
resp = client.get(f"/logs/recent?since={cursor_ts}")
|
|
items = [x for x in resp.json()["logs"] if x["source"] == "access"]
|
|
assert len(items) == 1
|