149 lines
4.7 KiB
Python
149 lines
4.7 KiB
Python
"""전수 검증: unified prompt → JSON parse → NC handler."""
|
|
|
|
import asyncio
|
|
import io
|
|
import json
|
|
import sys
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
|
sys.path.insert(0, ".")
|
|
|
|
|
|
# ── Step 1: _parse_unified_response 검증 ──
|
|
|
|
def _parse_unified_response(raw: str) -> dict:
|
|
import re
|
|
m = re.search(r"```json\s*\n(.+?)```", raw, re.DOTALL)
|
|
if m:
|
|
try:
|
|
return json.loads(m.group(1))
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
start = raw.find("{")
|
|
if start != -1:
|
|
depth = 0
|
|
for i in range(start, len(raw)):
|
|
if raw[i] == "{":
|
|
depth += 1
|
|
elif raw[i] == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
try:
|
|
return json.loads(raw[start:i + 1])
|
|
except json.JSONDecodeError as e:
|
|
print(f" JSON decode error: {e}")
|
|
print(f" Attempted: {raw[start:i+1][:200]}")
|
|
break
|
|
|
|
return {"mode": "chat", "response": raw}
|
|
|
|
|
|
# 테스트 케이스들
|
|
test_cases = [
|
|
# (1) 일반 JSON
|
|
'{\n "mode": "nextcloud",\n "tool": "files",\n "op": "list",\n "params": {\n "path": "/"\n },\n "summary": "test"\n}',
|
|
# (2) 끝에 ``` 붙은 경우
|
|
'{\n "mode": "nextcloud",\n "tool": "files",\n "op": "list",\n "params": {\n "path": "/"\n },\n "summary": "test"\n}\n```',
|
|
# (3) ```json 블록
|
|
'```json\n{\n "mode": "chat",\n "response": "안녕하세요"\n}\n```',
|
|
# (4) 텍스트 + JSON
|
|
'Here is the result:\n{\n "mode": "nextcloud",\n "tool": "mail",\n "op": "unread",\n "params": {"limit": 5},\n "summary": "test"\n}',
|
|
# (5) 빈 문자열
|
|
'',
|
|
# (6) mode 없는 JSON
|
|
'{"error": "something"}',
|
|
]
|
|
|
|
print("=== Step 1: JSON 파서 테스트 ===\n")
|
|
for i, tc in enumerate(test_cases, 1):
|
|
result = _parse_unified_response(tc)
|
|
mode = result.get("mode", "?")
|
|
status = "✅" if mode != "chat" or "response" in result else "❌"
|
|
print(f" [{i}] {status} mode={mode} keys={list(result.keys())}")
|
|
if mode == "nextcloud":
|
|
print(f" tool={result.get('tool')} op={result.get('op')} params={result.get('params')}")
|
|
|
|
|
|
# ── Step 2: NC handler 검증 (실제 API 호출) ──
|
|
|
|
async def test_nc_handler():
|
|
print("\n=== Step 2: NC handler 메서드 검증 ===\n")
|
|
|
|
from handlers.nc_handler import NCHandler
|
|
handler = NCHandler()
|
|
|
|
# files.list_dir
|
|
print(" [1] files.list_dir('')...")
|
|
try:
|
|
files = await handler.files.list_dir("")
|
|
print(f" ✅ {len(files)}건: {', '.join(f.name for f in files[:5])}")
|
|
except Exception as e:
|
|
print(f" ❌ {e}")
|
|
|
|
# files.search
|
|
print(" [2] files.search('pdf')...")
|
|
try:
|
|
files = await handler.files.search("pdf")
|
|
print(f" ✅ {len(files)}건: {', '.join(f.name for f in files[:5])}")
|
|
except Exception as e:
|
|
print(f" ❌ {e}")
|
|
|
|
# calendar.get_today (method exists?)
|
|
print(" [3] calendar 메서드 확인...")
|
|
cal = handler.calendar
|
|
has_today = hasattr(cal, "get_today")
|
|
has_week = hasattr(cal, "get_week")
|
|
has_events = hasattr(cal, "get_events")
|
|
print(f" get_today={has_today} get_week={has_week} get_events={has_events}")
|
|
|
|
# mail.get_unread
|
|
print(" [4] mail.get_unread(3)...")
|
|
try:
|
|
msgs = await handler.mail.get_unread(3)
|
|
print(f" ✅ {len(msgs)}건")
|
|
except Exception as e:
|
|
print(f" ❌ {e}")
|
|
|
|
|
|
# ── Step 3: NC handler.handle() 모의 호출 검증 ──
|
|
|
|
async def test_handle_dispatch():
|
|
print("\n=== Step 3: handle() 디스패치 검증 ===\n")
|
|
|
|
from handlers.nc_handler import NCHandler
|
|
handler = NCHandler()
|
|
|
|
# handle() 내부 _handle_files 호출 경로 확인
|
|
action = {"mode": "nextcloud", "tool": "files", "op": "list", "params": {"path": ""}}
|
|
|
|
class FakeChannel:
|
|
"""Discord channel 모의 객체."""
|
|
sent = []
|
|
|
|
async def send(self, content=None, embed=None):
|
|
if embed:
|
|
self.sent.append(f"[EMBED] title={embed.title}, desc_len={len(embed.description or '')}")
|
|
elif content:
|
|
self.sent.append(f"[TEXT] {content[:100]}")
|
|
print(f" → send: {self.sent[-1]}")
|
|
|
|
ch = FakeChannel()
|
|
try:
|
|
await handler.handle(action, ch)
|
|
if ch.sent:
|
|
print(f" ✅ {len(ch.sent)}건 전송됨")
|
|
else:
|
|
print(f" ❌ 전송 없음")
|
|
except Exception as e:
|
|
print(f" ❌ handle() 오류: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|
|
|
|
async def main():
|
|
await test_nc_handler()
|
|
await test_handle_dispatch()
|
|
|
|
|
|
asyncio.run(main())
|