feat(_shared): access_log 공용 모듈 추가 (ring buffer + middleware + /logs/recent)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

_shared/tests/test_access_log.py

@@ -0,0 +1,129 @@
+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