zioinfo-mail/itsm/test_a1_ws.py

"""A-1 WebSocket 실시간 대시보드 테스트"""
import sys, ast, os, asyncio, json

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-ws-secret-key-32bytes-pad!!!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_ws.db")

print("=== 1. 구문 검사 ===")
files = ["routers/ws.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. ConnectionManager 단위 테스트 ===")
# ws.py 임포트 (DB 의존성 없는 클래스만 테스트)
import importlib.util
spec = importlib.util.spec_from_file_location("ws_mod", "routers/ws.py")
ws_mod = importlib.util.module_from_spec(spec)
try:
    spec.loader.exec_module(ws_mod)
    manager = ws_mod.ConnectionManager()

    # 채널 정의 확인
    channels = ws_mod._CHANNELS
    assert "all" in channels
    assert "sr" in channels
    assert "sla" in channels
    assert "deploy" in channels
    assert "oncall" in channels
    assert "batch" in channels
    print(f"  OK  채널 정의: {sorted(channels)}")

    # _CHANNEL_EVENT_MAP 확인
    event_map = ws_mod._CHANNEL_EVENT_MAP
    assert "sr_created" in event_map["sr"]
    assert "sla_violation" in event_map["sla"]
    assert event_map["all"] is None  # all = 모든 이벤트
    print(f"  OK  채널 이벤트 매핑 확인")

    # connection_count 초기값
    assert manager.connection_count() == 0
    print(f"  OK  connection_count 초기값 = 0")

    # connections_info 빈 목록
    assert manager.connections_info() == []
    print(f"  OK  connections_info 빈 목록")

except Exception as e:
    print(f"  INFO  외부 의존성 에러 (정상): {type(e).__name__}: {str(e)[:80]}")

print("\n=== 3. 이벤트 브로드캐스트 단위 테스트 ===")

async def test_broadcast():
    from routers.ws import ConnectionManager, _CHANNEL_EVENT_MAP

    # 더미 WebSocket 객체 (실제 연결 없이)
    class MockWS:
        def __init__(self):
            self.messages = []
            self.closed = False

        async def accept(self):
            pass

        async def send_text(self, text):
            self.messages.append(json.loads(text))

        async def close(self, code=None, reason=None):
            self.closed = True

    mgr = ConnectionManager()

    # 연결
    ws1 = MockWS()
    ws2 = MockWS()
    await mgr.connect(ws1, "alice", "ADMIN", "all")
    await mgr.connect(ws2, "bob", "ENGINEER", "sr")
    assert mgr.connection_count() == 2
    print(f"  OK  2개 연결 등록")

    # 전체 채널 이벤트 브로드캐스트
    await mgr.broadcast("sr_created", {"sr_id": "SR-001", "title": "테스트"})
    assert len(ws1.messages) == 1, f"Expected 1 msg for alice, got {len(ws1.messages)}"
    assert len(ws2.messages) == 1, f"Expected 1 msg for bob, got {len(ws2.messages)}"
    assert ws1.messages[0]["type"] == "sr_created"
    assert ws1.messages[0]["sr_id"] == "SR-001"
    print(f"  OK  sr_created 브로드캐스트 (2명 수신)")

    # deploy 이벤트 → sr 채널 구독자(bob)는 수신 안함
    await mgr.broadcast("deploy_completed", {"session_id": 1})
    assert len(ws1.messages) == 2, f"alice should have 2 msgs, got {len(ws1.messages)}"
    assert len(ws2.messages) == 1, f"bob (sr channel) should still have 1 msg, got {len(ws2.messages)}"
    print(f"  OK  deploy_completed: alice 수신, bob(sr채널) 미수신")

    # sla_violation → sr 채널 (bob) 포함 여부 확인
    # sr 채널의 allowed = {"sr_created", "sr_updated", "sr_status_changed"}
    # sla_violation is NOT in sr channel events
    await mgr.broadcast("sla_violation", {"sr_id": "SR-001"})
    assert len(ws2.messages) == 1, f"bob (sr channel) should not receive sla_violation"
    print(f"  OK  sla_violation: bob(sr채널) 미수신")

    # disconnect
    mgr.disconnect(ws1)
    assert mgr.connection_count() == 1
    print(f"  OK  disconnect: 1명 남음")

asyncio.run(test_broadcast())

print("\n=== 4. main.py 등록 확인 ===")
with open("main.py", encoding="utf-8") as f:
    main_src = f.read()
checks = [
    ("ws_router", "ws 라우터 임포트"),
    ("ws_router.router", "ws 라우터 등록"),
    ("_integrate_with_sse_bus", "SSE 통합 패치"),
]
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=== A-1 WebSocket 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)