zioinfo-mail/workspace/guardia-itsm/test_f2f3_cache.py

"""F-2 Redis 캐시 / F-3 Rate Limiting 테스트"""
import sys, ast, os, asyncio

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-cache-secret-32bytes-pad!!!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_cache.db")
os.environ["CACHE_ENABLED"] = "true"

print("=== 1. 구문 검사 ===")
files = ["core/cache.py", "core/ratelimit.py", "routers/analytics.py", "main.py"]
ok = True
for f in files:
    try:
        with open(f, encoding="utf-8") as fh:
            src = fh.read()
        ast.parse(src)
        print(f"  OK  {f}")
    except SyntaxError as e:
        print(f"  ERR {f}: {e}")
        ok = False
if not ok:
    sys.exit(1)

print("\n=== 2. 인메모리 캐시 단위 테스트 ===")
from core.cache import _MemoryCache

mc = _MemoryCache(maxsize=5)

# set / get
mc.set("key1", {"data": 42}, ttl=60)
val = mc.get("key1")
assert val == {"data": 42}, f"Expected dict, got {val}"
print("  OK  set/get")

# TTL 만료 테스트 (0초 TTL)
import time
mc.set("key_exp", "will expire", ttl=0)
time.sleep(0.01)
assert mc.get("key_exp") is None, "TTL 0 항목이 만료되지 않음"
print("  OK  TTL 만료")

# delete
mc.set("del_key", "to delete", ttl=60)
mc.delete("del_key")
assert mc.get("del_key") is None
print("  OK  delete")

# LRU 초과 시 가장 오래된 항목 제거 (maxsize=5)
for i in range(6):
    mc.set(f"lru_{i}", i, ttl=60)
# lru_0이 제거됨
assert mc.get("lru_0") is None, "LRU 항목이 제거되지 않음"
assert mc.get("lru_5") == 5, "최신 항목이 제거됨"
print("  OK  LRU 제거")

# prefix 키 조회
for i in range(3):
    mc.set(f"prefix:item:{i}", i, ttl=60)
keys = mc.keys_with_prefix("prefix:")
assert len(keys) == 3, f"Expected 3 keys, got {len(keys)}"
print(f"  OK  prefix 키 조회: {len(keys)}개")

print("\n=== 3. 비동기 캐시 API 테스트 (인메모리) ===")
# Redis 없이 메모리 캐시만 사용
os.environ["REDIS_URL"] = "redis://localhost:99999/0"  # 연결 불가 주소

async def test_async_cache():
    from core.cache import cache_get, cache_set, cache_delete, cache_invalidate_prefix, make_cache_key

    # set/get
    await cache_set("test:item1", {"hello": "world"}, ttl=60)
    val = await cache_get("test:item1")
    assert val == {"hello": "world"}, f"Expected dict, got {val}"
    print("  OK  async cache_set/cache_get")

    # 없는 키
    val_none = await cache_get("nonexistent:key")
    assert val_none is None
    print("  OK  cache_get None for missing key")

    # delete
    await cache_set("test:del", "value", ttl=60)
    await cache_delete("test:del")
    assert await cache_get("test:del") is None
    print("  OK  async cache_delete")

    # invalidate prefix
    for i in range(3):
        await cache_set(f"test:pfx:{i}", i, ttl=60)
    count = await cache_invalidate_prefix("test:pfx:")
    assert count == 3, f"Expected 3 deletions, got {count}"
    print(f"  OK  cache_invalidate_prefix: {count}개 삭제")

    # make_cache_key
    k1 = make_cache_key("tasks", status="OPEN", skip=0)
    k2 = make_cache_key("tasks", status="OPEN", skip=0)
    k3 = make_cache_key("tasks", status="CLOSED", skip=0)
    assert k1 == k2, "동일 인자 → 동일 키"
    assert k1 != k3, "다른 인자 → 다른 키"
    print(f"  OK  make_cache_key: {k1}")

    # cache_info
    from core.cache import cache_info
    info = await cache_info()
    assert "backend" in info
    assert "app_stats" in info
    print(f"  OK  cache_info: backend={info['backend']}")

asyncio.run(test_async_cache())

print("\n=== 4. Rate Limiter 임포트/초기화 테스트 ===")
from core.ratelimit import (
    create_limiter, limiter, setup_rate_limiting,
    DEFAULT_LIMIT, LOGIN_LIMIT, AI_LIMIT, UPLOAD_LIMIT,
    _get_user_key, _DummyLimiter,
)

print(f"  OK  limiter type: {type(limiter).__name__}")
assert DEFAULT_LIMIT == "120/minute"
assert LOGIN_LIMIT == "10/minute"
assert AI_LIMIT == "10/minute"
print(f"  OK  제한 상수: DEFAULT={DEFAULT_LIMIT}, LOGIN={LOGIN_LIMIT}, AI={AI_LIMIT}")

# 더미 리미터 작동 확인
dummy = _DummyLimiter()
@dummy.limit("10/minute")
async def dummy_fn():
    return "ok"
result = asyncio.run(dummy_fn())
assert result == "ok", "DummyLimiter 데코레이터 실패"
print(f"  OK  DummyLimiter 데코레이터 (no-op)")

print("\n=== 5. analytics.py 캐시/레이트리밋 엔드포인트 확인 ===")
with open("routers/analytics.py", encoding="utf-8") as f:
    src = f.read()
for endpoint in ["/admin/cache/info", "/admin/cache/flush", "/admin/ratelimit/info"]:
    status = "OK" if endpoint in src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {endpoint}")

print("\n=== 6. main.py 통합 확인 ===")
with open("main.py", encoding="utf-8") as f:
    main_src = f.read()
checks = [
    ("setup_rate_limiting", "Rate Limiting 미들웨어"),
    ("close_redis",         "Redis 종료 훅"),
]
for sym, desc in checks:
    status = "OK" if sym in main_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc} ({sym})")

print("\n=== F-2/F-3 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)