"""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)