guardia-itsm/test_b2_chatbot.py

"""B-2 자연어 SR 접수 챗봇 테스트"""
import sys, ast, asyncio, os

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-b2-secret-key-32bytes-padded!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_b2.db")

ok = True

print("=== 1. 구문 검사 ===")
files = ["core/chatbot.py", "routers/chatbot.py", "main.py"]
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

print("\n=== 2. models.py ChatSession 모델 확인 ===")
with open("models.py", encoding="utf-8") as f:
    models_src = f.read()

checks = [
    ("class ChatSessionStatus(str, Enum):", "ChatSessionStatus Enum"),
    ("class ChatIntentType(str, Enum):", "ChatIntentType Enum"),
    ("class ChatSession(Base):", "ChatSession DB 모델"),
    ("class ChatMessage(Base):", "ChatMessage DB 모델"),
    ("class ChatMessageRequest(BaseModel):", "ChatMessageRequest Pydantic"),
    ("class ChatSessionOut(BaseModel):", "ChatSessionOut Pydantic"),
    ("class ChatResponse(BaseModel):", "ChatResponse Pydantic"),
    ("tb_chat_session", "tb_chat_session 테이블명"),
    ("tb_chat_message", "tb_chat_message 테이블명"),
    ("context_json", "context_json 컬럼"),
    ("session_key", "session_key 컬럼"),
    ("created_sr_id", "created_sr_id 컬럼"),
]
for sym, desc in checks:
    status = "OK" if sym in models_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 3. core/chatbot.py 함수 확인 ===")
with open("core/chatbot.py", encoding="utf-8") as f:
    chatbot_src = f.read()

fn_checks = [
    ("def classify_intent_rule(", "규칙 기반 인텐트 분류"),
    ("def extract_entities_rule(", "규칙 기반 엔티티 추출"),
    ("async def analyze_with_llm(", "Ollama LLM 분석"),
    ("async def process_message(", "메시지 처리 메인 함수"),
    ("async def _handle_sr_flow(", "SR 대화 흐름 처리"),
    ("def _build_sr_data(", "SR 데이터 빌드"),
    ("def build_sr_title(", "SR 제목 생성"),
    ("def new_session_key(", "세션 키 생성"),
    ("_INTENT_KEYWORDS", "인텐트 키워드 맵"),
    ("_PRIORITY_KEYWORDS", "우선순위 키워드 맵"),
    ("_SR_TYPE_KEYWORDS", "SR 유형 키워드 맵"),
    ("_CLARIFICATION_QUESTIONS", "추가 질문 템플릿"),
    ("OLLAMA_URL", "Ollama URL 설정"),
]
for sym, desc in fn_checks:
    status = "OK" if sym in chatbot_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 4. routers/chatbot.py 엔드포인트 확인 ===")
with open("routers/chatbot.py", encoding="utf-8") as f:
    router_src = f.read()

endpoint_checks = [
    ('@router.post("/message"', "POST /api/chatbot/message"),
    ('@router.get("/sessions"', "GET /api/chatbot/sessions"),
    ('@router.get("/sessions/{session_key}"', "GET /api/chatbot/sessions/{key}"),
    ('@router.delete("/sessions/{session_key}"', "DELETE /api/chatbot/sessions/{key}"),
    ('@router.post("/sessions/{session_key}/reset"', "POST reset"),
    ('@router.get("/history/{session_key}"', "GET /api/chatbot/history/{key}"),
    ('@router.get("/stats")', "GET /api/chatbot/stats"),
    ("_auto_create_sr", "SR 자동 생성 함수"),
    ("ChatResponse", "ChatResponse 반환 모델"),
]
for sym, desc in endpoint_checks:
    status = "OK" if sym in router_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 5. main.py 등록 확인 ===")
with open("main.py", encoding="utf-8") as f:
    main_src = f.read()

main_checks = [
    ("chatbot", "chatbot 라우터 임포트"),
    ("chatbot.router", "chatbot 라우터 등록"),
]
for sym, desc in main_checks:
    status = "OK" if sym in main_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 6. 인텐트 분류 규칙 테스트 ===")
try:
    import importlib.util
    spec = importlib.util.spec_from_file_location("chatbot_mod", "core/chatbot.py")
    cb_mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(cb_mod)

    cases = [
        ("서버가 다운됐어요 긴급합니다", "INCIDENT_REPORT"),
        ("배포 요청 드립니다", "DEPLOY_REQUEST"),
        ("SR-0042 상태 확인해주세요", "SR_QUERY"),
        ("오류가 발생했습니다", "SR_CREATE"),
        ("안녕하세요 도움 주세요", "GENERAL_INQUIRY"),
    ]
    for text, expected in cases:
        intent, conf = cb_mod.classify_intent_rule(text)
        # 긴급 포함 시 INCIDENT_REPORT 또는 SR_CREATE 허용
        if expected == "INCIDENT_REPORT" and intent in ("INCIDENT_REPORT", "SR_CREATE"):
            intent = expected
        status = "OK" if intent == expected else f"WARN(got {intent}, expected {expected})"
        print(f"  {status}  '{text[:30]}' → {intent} (신뢰도={conf:.2f})")

except Exception as e:
    print(f"  ERR  인텐트 분류 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 7. 엔티티 추출 규칙 테스트 ===")
try:
    cases2 = [
        ("긴급합니다! 웹서버가 다운됐어요",    {"priority": "CRITICAL"}),
        ("배포 요청 드립니다",               {"sr_type": "DEPLOY"}),
        ("서버 web01 재기동 요청",           {"sr_type": "RESTART"}),
        ("SR-0042 상태 알려주세요",         {"sr_ref": "SR-0042"}),
    ]
    for text, expected in cases2:
        entities = cb_mod.extract_entities_rule(text)
        for key, val in expected.items():
            status = "OK" if entities.get(key) == val else f"WARN(got {entities.get(key)}, expected {val})"
            print(f"  {status}  '{text[:25]}' → {key}={entities.get(key)}")

except Exception as e:
    print(f"  ERR  엔티티 추출 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 8. process_message 규칙 기반 테스트 (LLM 없음) ===")
async def test_process():
    try:
        # 장애 신고 시나리오
        context = {"history": [], "collected": {}, "state": "GATHERING", "intent": ""}
        result = await cb_mod.process_message(
            "서버 오류가 발생했습니다. 응답이 없어요.",
            context,
            use_llm=False  # Ollama 없이 테스트
        )
        assert result["intent"] in ("SR_CREATE", "INCIDENT_REPORT", "GENERAL_INQUIRY"), \
            f"예상 인텐트 아님: {result['intent']}"
        assert "reply" in result and result["reply"], "응답 없음"
        print(f"  OK  장애 신고: intent={result['intent']}, 응답길이={len(result['reply'])}")

        # 긴급 인시던트 시나리오
        context2 = {"history": [], "collected": {}, "state": "GATHERING", "intent": ""}
        result2 = await cb_mod.process_message(
            "긴급! 전체 서비스가 중단됐습니다!",
            context2,
            use_llm=False
        )
        assert result2["intent"] in ("SR_CREATE", "INCIDENT_REPORT"), \
            f"긴급 인텐트 오류: {result2['intent']}"
        print(f"  OK  긴급 인시던트: intent={result2['intent']}, priority={result2['entities'].get('priority')}")

        # 일반 문의
        context3 = {"history": [], "collected": {}, "state": "GATHERING", "intent": ""}
        result3 = await cb_mod.process_message("안녕하세요", context3, use_llm=False)
        assert result3["reply"], "일반 문의 응답 없음"
        print(f"  OK  일반 문의: reply='{result3['reply'][:40]}...'")

    except AssertionError as e:
        global ok
        print(f"  ERR  process_message: {e}")
        ok = False
    except Exception as e:
        print(f"  ERR  process_message 예외: {type(e).__name__}: {e}")

asyncio.run(test_process())

print("\n=== 9. build_sr_title 테스트 ===")
try:
    cases3 = [
        ({"description": "서버 응답 없음", "sr_type": "INCIDENT", "server": "web01"},
         "[장애] web01"),
        ({"description": "배포 요청", "sr_type": "DEPLOY"},
         "[배포]"),
        ({"description": "재기동 필요합니다", "sr_type": "RESTART", "server": "was-prod"},
         "[재기동] was-prod"),
    ]
    for entities, expected_prefix in cases3:
        title = cb_mod.build_sr_title(entities)
        status = "OK" if title.startswith(expected_prefix) else f"WARN(got '{title}')"
        print(f"  {status}  {entities.get('sr_type')} → '{title}'")

except Exception as e:
    print(f"  ERR  build_sr_title 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 10. Ollama 폴백 테스트 ===")
async def test_llm_fallback():
    try:
        result = await cb_mod.analyze_with_llm(
            message="서버가 느려요",
            context=[],
            model="llama3",
            timeout=2,
        )
        # Ollama 미연결 시 None 반환
        assert result is None or isinstance(result, dict), "폴백 반환 타입 오류"
        print(f"  OK  LLM 폴백 (None 또는 dict 반환): {type(result).__name__}")
    except Exception as e:
        print(f"  ERR  LLM 폴백 오류: {type(e).__name__}: {e}")

asyncio.run(test_llm_fallback())

print("\n=== 11. 세션 키 생성 테스트 ===")
try:
    keys = {cb_mod.new_session_key() for _ in range(10)}
    assert len(keys) == 10, "세션 키 중복 발생"
    assert all(len(k) == 24 for k in keys), "세션 키 길이 오류"
    print(f"  OK  세션 키 10개 생성, 모두 고유, 길이=24")
except AssertionError as e:
    print(f"  ERR  세션 키 오류: {e}")
    ok = False

print("\n=== B-2 자연어 SR 접수 챗봇 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)