""" test_group_chat.py group_chat.py 단위 테스트 (아르고스 작성) 테스트 항목: 1. load_env_keys: 환경변수 로드 정상 작동 2. load_bot_token: 환경변수 / .env.keys 폴백 / RuntimeError 3. load_personas: 조직도 우선, personas.json 폴백, DEFAULT_PERSONAS 폴백 4. load_personas_from_org: 조직도 파싱 상세 검증 5. format_persona_tag: 태그 포맷 검증 6. EMOJI_MAP: 이모지 매핑 검증 7. read_trigger: 트리거 파일 읽기 + 삭제 확인 8. dump_session / load_session: 세션 상태 파일 덤프/복구 9. GroupChatSession.start: 입장 시퀀스 정상 동작 (send 호출 횟수) 10. GroupChatSession.end: 퇴장 시퀀스 + 세션 파일 정리 11. GroupChatSession.add_user_input: auto_turns 리셋, 히스토리 추가 12. select_next_speaker: 직전 발화자 제외, 폴백 로직 13. MAX_AUTO_TURNS 제한: auto_turns >= 6 시 대기 """ import json import os import sys import time import types from pathlib import Path from unittest.mock import MagicMock, mock_open, patch import pytest # --------------------------------------------------------------------------- # 모듈 임포트 전에 requests 스텁을 sys.modules에 주입 # group_chat.py 최상단에서 `import requests`를 하기 때문에 # 실제 패키지가 없는 환경에서도 테스트가 동작하도록 스텁을 미리 삽입한다. # --------------------------------------------------------------------------- def _inject_stubs(): """requests 스텁을 sys.modules에 등록한다.""" # ── requests 스텁 ─────────────────────────────────────────────────────── if "requests" not in sys.modules: requests_stub = types.ModuleType("requests") requests_stub.post = MagicMock(return_value=MagicMock(ok=True)) # type: ignore[attr-defined] requests_stub.get = MagicMock(return_value=MagicMock(ok=True)) # type: ignore[attr-defined] sys.modules["requests"] = requests_stub else: # 이미 등록된 스텁에 get/post 속성이 없으면 추가 stub = sys.modules["requests"] if not hasattr(stub, "get"): stub.get = MagicMock(return_value=MagicMock(ok=True)) # type: ignore[attr-defined] if not hasattr(stub, "post"): stub.post = MagicMock(return_value=MagicMock(ok=True)) # type: ignore[attr-defined] _inject_stubs() # --------------------------------------------------------------------------- # 워크스페이스 경로 및 모듈 로드 # --------------------------------------------------------------------------- WORKSPACE_ROOT = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) ORG_STRUCTURE_PATH = WORKSPACE_ROOT / "memory" / "organization-structure.json" if str(WORKSPACE_ROOT) not in sys.path: sys.path.insert(0, str(WORKSPACE_ROOT)) def _count_expected_org_personas() -> int: """organization-structure.json에서 load_personas_from_org와 동일한 로직으로 활성 인원 수를 계산한다.""" if not ORG_STRUCTURE_PATH.exists(): return -1 # 파일이 없으면 비교 불가 with open(ORG_STRUCTURE_PATH, encoding="utf-8") as f: org = json.load(f) seen: set = set() structure = org.get("structure", {}) # columns (수직조직 팀) for team in structure.get("columns", {}).get("teams", []): if team.get("status") != "active": continue if team.get("team_id") == "development-office": # 개발실: sub_teams 순회 for sub_team in team.get("sub_teams", []): if sub_team.get("status") != "active": continue lead = sub_team.get("lead") or {} if lead.get("id") and lead["id"] != "anu" and lead.get("status") == "active": seen.add(lead["id"]) for member in sub_team.get("members", []): if member.get("id") == "anu": continue if member.get("status") in ("active", "available"): seen.add(member["id"]) else: lead = team.get("lead") or {} if lead.get("id") and lead["id"] != "anu" and lead.get("name"): lead_status = lead.get("status", "active") if lead_status in ("active", "available"): seen.add(lead["id"]) for member in team.get("members", []): if member.get("id") == "anu": continue member_status = member.get("status", "active") if member_status in ("active", "available"): seen.add(member["id"]) # rows (횡단조직 센터) for center in structure.get("rows", {}).get("centers", []): if center.get("status") != "active": continue lead = center.get("lead") or {} if lead.get("id") and lead["id"] != "anu" and lead.get("status") in ("active", "available"): seen.add(lead["id"]) for member in center.get("members", []): if member.get("id") == "anu": continue member_status = member.get("status", "active") if member_status in ("active", "available"): seen.add(member["id"]) return len(seen) def _fresh_module(): """sys.modules에서 group_chat을 제거해 매 호출마다 새로운 상태로 임포트한다.""" _inject_stubs() # 스텁이 항상 등록되어 있도록 보장 for key in list(sys.modules.keys()): if key == "group_chat": del sys.modules[key] import group_chat as gc return gc # --------------------------------------------------------------------------- # 1. load_env_keys # --------------------------------------------------------------------------- class TestLoadEnvKeys: """load_env_keys 함수 테스트.""" def test_skips_when_token_already_set(self): """GROUP_CHAT_BOT_TOKEN이 이미 있으면 subprocess를 호출하지 않는다.""" gc = _fresh_module() with patch.dict(os.environ, {"GROUP_CHAT_BOT_TOKEN": "already-set"}, clear=False): with patch("subprocess.run") as mock_run: gc.load_env_keys() mock_run.assert_not_called() def test_loads_token_from_env_file(self, tmp_path): """GROUP_CHAT_BOT_TOKEN이 없을 때 .env.keys에서 읽어 환경변수에 설정한다.""" gc = _fresh_module() env_file = tmp_path / ".env.keys" env_file.write_text("export GROUP_CHAT_BOT_TOKEN=test-token-from-file\n") fake_result = MagicMock() fake_result.stdout = "GROUP_CHAT_BOT_TOKEN=test-token-from-file\nOTHER_VAR=foo\n" env_without_token = {k: v for k, v in os.environ.items() if k != "GROUP_CHAT_BOT_TOKEN"} with patch.dict(os.environ, env_without_token, clear=True): with patch.object(gc, "ENV_KEYS_FILE", str(env_file)): with patch("subprocess.run", return_value=fake_result): gc.load_env_keys() assert os.environ.get("GROUP_CHAT_BOT_TOKEN") == "test-token-from-file" def test_no_token_in_env_file(self, tmp_path): """subprocess 출력에 GROUP_CHAT_BOT_TOKEN이 없으면 환경변수가 설정되지 않는다.""" gc = _fresh_module() env_file = tmp_path / ".env.keys" env_file.write_text("export SOME_OTHER_VAR=value\n") fake_result = MagicMock() fake_result.stdout = "SOME_OTHER_VAR=value\n" env_without_token = {k: v for k, v in os.environ.items() if k != "GROUP_CHAT_BOT_TOKEN"} with patch.dict(os.environ, env_without_token, clear=True): with patch.object(gc, "ENV_KEYS_FILE", str(env_file)): with patch("subprocess.run", return_value=fake_result): gc.load_env_keys() assert "GROUP_CHAT_BOT_TOKEN" not in os.environ def test_missing_env_file_does_not_raise(self, tmp_path): """env.keys 파일이 없어도 예외 없이 반환한다.""" gc = _fresh_module() missing_path = tmp_path / "nonexistent.env.keys" env_without_token = {k: v for k, v in os.environ.items() if k != "GROUP_CHAT_BOT_TOKEN"} with patch.dict(os.environ, env_without_token, clear=True): with patch.object(gc, "ENV_KEYS_FILE", str(missing_path)): gc.load_env_keys() # --------------------------------------------------------------------------- # 2. load_bot_token (완전 재작성) # --------------------------------------------------------------------------- class TestLoadBotToken: """load_bot_token 함수 테스트 - GROUP_CHAT_BOT_TOKEN 환경변수 기반.""" def test_loads_from_env_var(self): """GROUP_CHAT_BOT_TOKEN 환경변수에서 토큰을 로드한다.""" gc = _fresh_module() with patch.dict(os.environ, {"GROUP_CHAT_BOT_TOKEN": "1234567:ENV_TOKEN"}, clear=False): token = gc.load_bot_token() assert token == "1234567:ENV_TOKEN" def test_loads_from_env_keys_fallback(self, tmp_path): """환경변수 없을 때 .env.keys에서 폴백 로드한다.""" gc = _fresh_module() env_file = tmp_path / ".env.keys" env_file.write_text("export GROUP_CHAT_BOT_TOKEN=9999:FALLBACK_TOKEN\n") fake_result = MagicMock() fake_result.stdout = "GROUP_CHAT_BOT_TOKEN=9999:FALLBACK_TOKEN\nOTHER=x\n" env_without_token = {k: v for k, v in os.environ.items() if k != "GROUP_CHAT_BOT_TOKEN"} with patch.dict(os.environ, env_without_token, clear=True): with patch.object(gc, "ENV_KEYS_FILE", str(env_file)): with patch("subprocess.run", return_value=fake_result): token = gc.load_bot_token() assert token == "9999:FALLBACK_TOKEN" def test_raises_when_no_token_found(self, tmp_path): """환경변수도 없고 .env.keys에서도 못 찾으면 RuntimeError를 발생시킨다.""" gc = _fresh_module() env_file = tmp_path / ".env.keys" env_file.write_text("export OTHER_VAR=something\n") fake_result = MagicMock() fake_result.stdout = "OTHER_VAR=something\n" env_without_token = {k: v for k, v in os.environ.items() if k != "GROUP_CHAT_BOT_TOKEN"} with patch.dict(os.environ, env_without_token, clear=True): with patch.object(gc, "ENV_KEYS_FILE", str(env_file)): with patch("subprocess.run", return_value=fake_result): with pytest.raises(RuntimeError): gc.load_bot_token() def test_env_var_takes_priority(self, tmp_path): """환경변수와 .env.keys 모두 있을 때 환경변수가 우선한다.""" gc = _fresh_module() env_file = tmp_path / ".env.keys" env_file.write_text("export GROUP_CHAT_BOT_TOKEN=FALLBACK_TOKEN\n") fake_result = MagicMock() fake_result.stdout = "GROUP_CHAT_BOT_TOKEN=FALLBACK_TOKEN\n" with patch.dict(os.environ, {"GROUP_CHAT_BOT_TOKEN": "ENV_PRIORITY_TOKEN"}, clear=False): with patch.object(gc, "ENV_KEYS_FILE", str(env_file)): with patch("subprocess.run", return_value=fake_result) as mock_run: token = gc.load_bot_token() # 환경변수가 있으므로 subprocess는 호출되지 않아야 한다 mock_run.assert_not_called() assert token == "ENV_PRIORITY_TOKEN" # --------------------------------------------------------------------------- # 3. load_personas (재작성) # --------------------------------------------------------------------------- class TestLoadPersonas: """load_personas 함수 테스트 - 조직도 우선, personas.json 폴백.""" def test_loads_from_org_structure(self): """조직도 파일에서 페르소나를 로드한다 (실제 파일 기반 통합 테스트).""" gc = _fresh_module() # 실제 조직도 파일이 존재하는 경우에만 실행 if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas() assert isinstance(result, dict) assert len(result) > 0 def test_excludes_anu_from_org(self): """anu가 조직도 로드 결과에서 제외된다 (실제 파일 기반).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas() assert "anu" not in result def test_excludes_planned_status(self): """planned 상태인 항목은 조직도 로드 결과에서 제외된다.""" gc = _fresh_module() # planned 상태만 있는 가짜 조직도 fake_org = { "structure": { "columns": { "teams": [ { "team_id": "strategy-team", "team_name": "전략팀", "status": "planned", "lead": { "id": "zeus", "name": "제우스", "status": "planned", "role": "팀장", }, "members": [], } ] }, "rows": {"centers": []}, } } with patch("builtins.open", mock_open(read_data=json.dumps(fake_org))): with patch("pathlib.Path.exists", return_value=True): result = gc.load_personas_from_org() assert "zeus" not in result def test_falls_back_to_personas_json(self, tmp_path): """조직도 로드 실패 시 personas.json으로 폴백한다.""" gc = _fresh_module() custom = { "custom_bot": { "name": "커스텀봇", "team": "테스트팀", "role": "테스터", "expertise": "QA", "personality": "꼼꼼함", "persona_desc": "", } } personas_file = tmp_path / "personas.json" personas_file.write_text(json.dumps(custom, ensure_ascii=False), encoding="utf-8") with patch.object(gc, "ORG_STRUCTURE_FILE", "/nonexistent/path/org.json"): with patch.object(gc, "PERSONAS_FILE", str(personas_file)): result = gc.load_personas() assert "custom_bot" in result assert result["custom_bot"]["name"] == "커스텀봇" def test_falls_back_to_defaults(self, tmp_path): """모든 소스 실패 시 DEFAULT_PERSONAS를 반환한다.""" gc = _fresh_module() missing_personas = tmp_path / "no_personas.json" with patch.object(gc, "ORG_STRUCTURE_FILE", "/nonexistent/path/org.json"): with patch.object(gc, "PERSONAS_FILE", str(missing_personas)): result = gc.load_personas() assert result is gc.DEFAULT_PERSONAS def test_org_loads_expected_count(self): """조직도에서 기대 인원 수를 로드한다 (anu 제외, planned 제외, 동적 계산).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() expected_count = _count_expected_org_personas() assert expected_count > 0, "organization-structure.json에서 인원 수 계산 실패" assert len(result) == expected_count def test_default_personas_contain_expected_keys(self): """DEFAULT_PERSONAS에 필수 페르소나가 포함되어 있다.""" gc = _fresh_module() expected_keys = {"hermes", "athena", "thor", "vulcan", "iris", "odin", "loki"} assert expected_keys.issubset(set(gc.DEFAULT_PERSONAS.keys())) def test_default_personas_have_new_fields(self): """DEFAULT_PERSONAS 각 항목이 team, persona_desc 필드를 가지고 emoji 필드가 없다.""" gc = _fresh_module() for key, persona in gc.DEFAULT_PERSONAS.items(): assert "team" in persona, f"{key}에 team 필드 없음" assert "persona_desc" in persona, f"{key}에 persona_desc 필드 없음" assert "emoji" not in persona, f"{key}에 emoji 필드가 있으면 안 됨" # --------------------------------------------------------------------------- # 4. TestLoadPersonasFromOrg (새 클래스) # --------------------------------------------------------------------------- class TestLoadPersonasFromOrg: """load_personas_from_org 함수 상세 테스트.""" def test_parses_dev_teams(self): """개발1팀, 개발2팀, 개발3팀 멤버를 파싱한다 (실제 파일 기반).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() # 개발1팀 멤버 assert "hermes" in result assert result["hermes"]["team"] == "개발1팀" assert "vulcan" in result assert result["vulcan"]["team"] == "개발1팀" assert "iris" in result assert result["iris"]["team"] == "개발1팀" assert "athena" in result assert result["athena"]["team"] == "개발1팀" # 개발2팀 멤버 assert "odin" in result assert result["odin"]["team"] == "개발2팀" assert "thor" in result assert result["thor"]["team"] == "개발2팀" # 개발8팀 멤버 assert "ra" in result assert result["ra"]["team"] == "개발8팀" assert "anubis" in result assert result["anubis"]["team"] == "개발8팀" def test_parses_security_team(self): """보안팀 리더(로키)를 파싱한다 (실제 파일 기반).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() assert "loki" in result assert result["loki"]["team"] == "보안팀" assert result["loki"]["role"] == "보안팀장 (Primary)" def test_parses_centers(self): """횡단조직(QC 센터, DevOps 센터, 디자인 센터)을 파싱한다 (실제 파일 기반).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() assert "maat" in result assert result["maat"]["team"] == "QC 센터" assert "janus" in result assert result["janus"]["team"] == "DevOps 센터" assert "venus" in result assert result["venus"]["team"] == "Gemini 센터" def test_persona_has_required_fields(self): """각 페르소나에 name, team, role, expertise, personality, persona_desc 필드가 있다.""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() required_fields = {"name", "team", "role", "expertise", "personality", "persona_desc"} for key, persona in result.items(): missing = required_fields - set(persona.keys()) assert not missing, f"{key}에 누락된 필드: {missing}" def test_name_parsing_strips_parenthetical(self): """'헤르메스 (Hermes)' 형태의 이름을 '헤르메스'로 파싱한다.""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() # 헤르메스: 원본은 "헤르메스 (Hermes)" assert result["hermes"]["name"] == "헤르메스" # 로키: 원본은 "로키 (Loki)" assert result["loki"]["name"] == "로키" # 마아트: 원본은 "마아트 (Ma'at)" assert result["maat"]["name"] == "마아트" def test_excludes_anu(self): """anu는 결과에서 제외된다.""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() assert "anu" not in result def test_excludes_planned_teams(self): """planned 상태인 팀은 파싱하지 않는다 (인원 수 동적 검증).""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") result = gc.load_personas_from_org() # strategy-team, marketing-team, insurance-team은 planned → 해당 팀원 없어야 함 # finance-center도 planned → 제외 # 이 팀들에 실제 멤버가 없으므로 키가 없어야 함 (간접 검증) expected_count = _count_expected_org_personas() assert expected_count > 0, "organization-structure.json에서 인원 수 계산 실패" assert len(result) == expected_count def test_raises_on_missing_org_file(self, tmp_path): """조직도 파일이 없으면 예외를 발생시킨다.""" gc = _fresh_module() with patch.object(gc, "ORG_STRUCTURE_FILE", str(tmp_path / "nonexistent.json")): with pytest.raises(Exception): gc.load_personas_from_org() # --------------------------------------------------------------------------- # 5. TestFormatPersonaTag (새 클래스) # --------------------------------------------------------------------------- class TestFormatPersonaTag: """format_persona_tag 함수 테스트.""" def test_format_with_team_and_role(self): """team과 role이 모두 있으면 '⚡ 헤르메스(개발1팀/개발1팀장)' 형태를 반환한다.""" gc = _fresh_module() personas_data = { "hermes": { "name": "헤르메스", "team": "개발1팀", "role": "개발1팀장", } } result = gc.format_persona_tag("hermes", personas_data) assert result == "⚡ 헤르메스(개발1팀/개발1팀장)" def test_format_with_role_only(self): """team이 없고 role만 있으면 '이모지 이름(역할)' 형태를 반환한다.""" gc = _fresh_module() personas_data = { "hermes": { "name": "헤르메스", "team": "", "role": "개발1팀장", } } result = gc.format_persona_tag("hermes", personas_data) assert result == "⚡ 헤르메스(개발1팀장)" def test_format_unknown_persona_uses_default_emoji(self): """EMOJI_MAP에 없는 키는 '💬' 이모지를 사용한다.""" gc = _fresh_module() personas_data = { "unknown_bot": { "name": "알수없는봇", "team": "테스트팀", "role": "봇", } } result = gc.format_persona_tag("unknown_bot", personas_data) assert result.startswith("💬") assert "알수없는봇" in result def test_format_missing_persona_key(self): """personas_data에 없는 키를 처리할 때 키 자체를 이름으로 사용한다.""" gc = _fresh_module() personas_data = {} # KeyError 없이 처리되어야 한다 result = gc.format_persona_tag("nonexistent", personas_data) assert isinstance(result, str) assert len(result) > 0 def test_format_odin_tag(self): """오딘의 태그 포맷을 검증한다.""" gc = _fresh_module() personas_data = { "odin": { "name": "오딘", "team": "개발2팀", "role": "개발2팀장", } } result = gc.format_persona_tag("odin", personas_data) assert result == "👁️ 오딘(개발2팀/개발2팀장)" def test_format_loki_tag(self): """로키의 태그 포맷을 검증한다.""" gc = _fresh_module() personas_data = { "loki": { "name": "로키", "team": "보안팀", "role": "보안팀 리더", } } result = gc.format_persona_tag("loki", personas_data) assert result == "🎭 로키(보안팀/보안팀 리더)" def test_format_with_real_org_data(self): """실제 조직도 데이터로 포맷을 검증한다.""" gc = _fresh_module() if not ORG_STRUCTURE_PATH.exists(): pytest.skip("organization-structure.json 파일이 없습니다.") personas_data = gc.load_personas_from_org() result = gc.format_persona_tag("hermes", personas_data) # "⚡ 헤르메스(개발1팀/개발1팀장)" 형태여야 한다 assert "헤르메스" in result assert "개발1팀" in result assert "개발1팀장" in result # --------------------------------------------------------------------------- # 6. TestEmojiMap (새 클래스) # --------------------------------------------------------------------------- class TestEmojiMap: """EMOJI_MAP 상수 테스트.""" def test_known_personas_have_emoji(self): """주요 페르소나가 EMOJI_MAP에 있는지 확인한다.""" gc = _fresh_module() expected_keys = { "hermes", "odin", "ra", "vulcan", "iris", "athena", "argos", "thor", "freya", "mimir", "heimdall", "anubis", "isis", "sobek", "horus", "loki", "venus", "janus", "maat", } assert expected_keys.issubset(set(gc.EMOJI_MAP.keys())) def test_hermes_emoji(self): """헤르메스의 이모지가 '⚡'이다.""" gc = _fresh_module() assert gc.EMOJI_MAP["hermes"] == "⚡" def test_loki_emoji(self): """로키의 이모지가 '🎭'이다.""" gc = _fresh_module() assert gc.EMOJI_MAP["loki"] == "🎭" def test_emoji_map_is_dict(self): """EMOJI_MAP이 딕셔너리 타입이다.""" gc = _fresh_module() assert isinstance(gc.EMOJI_MAP, dict) def test_emoji_map_values_are_strings(self): """EMOJI_MAP의 모든 값이 문자열이다.""" gc = _fresh_module() for key, emoji in gc.EMOJI_MAP.items(): assert isinstance(emoji, str), f"{key}의 이모지가 문자열이 아님" assert len(emoji) > 0, f"{key}의 이모지가 빈 문자열" # --------------------------------------------------------------------------- # 7. read_trigger # --------------------------------------------------------------------------- class TestReadTrigger: """read_trigger 함수 테스트.""" def test_returns_none_when_file_absent(self, tmp_path): """트리거 파일이 없으면 None을 반환한다.""" gc = _fresh_module() missing = tmp_path / "group_chat_trigger.json" with patch.object(gc, "TRIGGER_FILE", str(missing)): result = gc.read_trigger() assert result is None def test_reads_and_deletes_file(self, tmp_path): """트리거 파일을 읽고 삭제한다.""" gc = _fresh_module() trigger_data = {"action": "start", "topic": "배포 논의"} trigger_file = tmp_path / "group_chat_trigger.json" trigger_file.write_text(json.dumps(trigger_data, ensure_ascii=False), encoding="utf-8") with patch.object(gc, "TRIGGER_FILE", str(trigger_file)): result = gc.read_trigger() assert result == trigger_data # 파일이 삭제되어야 한다 assert not trigger_file.exists() def test_returns_none_on_invalid_json(self, tmp_path): """트리거 파일의 JSON이 깨져 있으면 None을 반환한다.""" gc = _fresh_module() trigger_file = tmp_path / "group_chat_trigger.json" trigger_file.write_text("{broken json}", encoding="utf-8") with patch.object(gc, "TRIGGER_FILE", str(trigger_file)): result = gc.read_trigger() assert result is None # --------------------------------------------------------------------------- # 8. dump_session / load_session # --------------------------------------------------------------------------- class TestDumpAndLoadSession: """dump_session / load_session 함수 테스트.""" def test_dump_creates_file(self, tmp_path): """dump_session은 세션 데이터를 JSON 파일로 저장한다.""" gc = _fresh_module() session_file = tmp_path / "group_chat_session.json" session_data = { "active": True, "topic": "테스트 주제", "personas": ["hermes", "athena"], "chat_id": "123", "history": [], "last_activity": 1700000000.0, "auto_turns": 3, "speak_counts": {"hermes": 2, "athena": 1}, } with patch.object(gc, "SESSION_FILE", str(session_file)): gc.dump_session(session_data) assert session_file.exists() loaded = json.loads(session_file.read_text(encoding="utf-8")) assert loaded["topic"] == "테스트 주제" assert loaded["active"] is True def test_load_session_returns_active_session(self, tmp_path): """active=True인 세션 파일이 있으면 데이터를 반환한다.""" gc = _fresh_module() session_file = tmp_path / "group_chat_session.json" session_data = { "active": True, "topic": "복구 테스트", "personas": ["thor"], "chat_id": "456", "history": [], "last_activity": 1700000000.0, "auto_turns": 0, "speak_counts": {"thor": 0}, } session_file.write_text(json.dumps(session_data, ensure_ascii=False), encoding="utf-8") with patch.object(gc, "SESSION_FILE", str(session_file)): result = gc.load_session() assert result is not None assert result["topic"] == "복구 테스트" def test_load_session_returns_none_when_inactive(self, tmp_path): """active=False인 세션 파일은 None을 반환한다.""" gc = _fresh_module() session_file = tmp_path / "group_chat_session.json" session_data = {"active": False, "topic": "종료된 세션"} session_file.write_text(json.dumps(session_data, ensure_ascii=False), encoding="utf-8") with patch.object(gc, "SESSION_FILE", str(session_file)): result = gc.load_session() assert result is None def test_load_session_returns_none_when_file_missing(self, tmp_path): """세션 파일이 없으면 None을 반환한다.""" gc = _fresh_module() missing = tmp_path / "no_session.json" with patch.object(gc, "SESSION_FILE", str(missing)): result = gc.load_session() assert result is None def test_dump_and_load_roundtrip(self, tmp_path): """dump_session → load_session 왕복 테스트.""" gc = _fresh_module() session_file = tmp_path / "group_chat_session.json" session_data = { "active": True, "topic": "왕복 테스트", "personas": ["loki", "odin"], "chat_id": "789", "history": [{"speaker": "loki", "speaker_name": "로키", "content": "안녕"}], "last_activity": 1700001234.5, "auto_turns": 2, "speak_counts": {"loki": 1, "odin": 1}, } with patch.object(gc, "SESSION_FILE", str(session_file)): gc.dump_session(session_data) result = gc.load_session() assert result is not None assert result["topic"] == "왕복 테스트" assert result["personas"] == ["loki", "odin"] assert len(result["history"]) == 1 # --------------------------------------------------------------------------- # 공통 픽스처: GroupChatSession 인스턴스 # --------------------------------------------------------------------------- @pytest.fixture() def gc_module(): """group_chat 모듈 (픽스처에서 공유).""" return _fresh_module() @pytest.fixture() def session_obj(gc_module): """격리된 GroupChatSession 인스턴스. DEFAULT_PERSONAS 구조 (team, persona_desc 필드 포함, emoji 필드 없음)를 사용한다. """ sess = gc_module.GroupChatSession( token="fake-token", personas_data=gc_module.DEFAULT_PERSONAS, ) return sess # --------------------------------------------------------------------------- # 9. GroupChatSession.start # --------------------------------------------------------------------------- class TestGroupChatSessionStart: """GroupChatSession.start 입장 시퀀스 테스트.""" def test_start_calls_send_correct_number_of_times(self, session_obj): """start()는 입장 안내 1회 + 페르소나별 1회 + 시작 안내 1회 = N+2 회 send를 호출한다.""" trigger = { "action": "start", "topic": "배포 전략", "personas": ["hermes", "athena"], "chat_id": "999", } with patch.object(session_obj, "send") as mock_send: with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) # 페르소나 2명 → send 호출 = 1(입장 안내) + 2(페르소나 인사) + 1(시작 안내) = 4 assert mock_send.call_count == 4 def test_start_sets_active_true(self, session_obj): """start() 후 active가 True가 된다.""" trigger = { "action": "start", "topic": "테스트 주제", "personas": ["thor"], "chat_id": "100", } with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) assert session_obj.active is True def test_start_initializes_speak_counts(self, session_obj): """start() 후 speak_counts가 0으로 초기화된다.""" trigger = { "topic": "발화 카운트 초기화", "personas": ["hermes", "athena", "thor"], } with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) assert session_obj.speak_counts == {"hermes": 0, "athena": 0, "thor": 0} def test_start_with_user_message_adds_to_history(self, session_obj): """user_message가 있으면 히스토리에 추가된다.""" trigger = { "topic": "프로젝트 킥오프", "personas": ["vulcan"], "user_message": "안녕하세요 여러분!", } with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) assert len(session_obj.history) == 1 assert session_obj.history[0]["content"] == "안녕하세요 여러분!" assert session_obj.history[0]["speaker"] == "user" def test_start_first_message_content(self, session_obj): """start()의 첫 번째 send 메시지는 '잠깐요' 안내 메시지다.""" trigger = { "topic": "긴급 회의", "personas": ["iris"], } sent_messages = [] with patch.object(session_obj, "send", side_effect=lambda t: sent_messages.append(t)): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) assert "잠깐요" in sent_messages[0] def test_start_last_message_contains_member_count(self, session_obj): """start()의 마지막 send 메시지는 참여 인원 수를 포함한다.""" personas = ["hermes", "athena", "thor"] trigger = {"topic": "팀 회의", "personas": personas} sent_messages = [] with patch.object(session_obj, "send", side_effect=lambda t: sent_messages.append(t)): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) last_msg = sent_messages[-1] assert "3" in last_msg assert "명" in last_msg def test_start_api_error_uses_fallback_greeting(self, session_obj): """CLI 오류 시 폴백 인사('잘 부탁드립니다.')가 포함된 메시지가 전송된다.""" trigger = { "topic": "폴백 테스트", "personas": ["loki"], } sent_messages = [] with patch.object(session_obj, "send", side_effect=lambda t: sent_messages.append(t)): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", side_effect=RuntimeError("CLI 오류")): session_obj.start(trigger) # 페르소나 입장 메시지 (인덱스 1)에 폴백 문자열이 포함되어야 한다 persona_msg = sent_messages[1] assert "잘 부탁드립니다." in persona_msg def test_start_persona_message_contains_new_format(self, session_obj): """페르소나 입장 메시지가 '이름(팀/역할)' 포맷을 포함한다.""" trigger = { "topic": "포맷 확인", "personas": ["hermes"], } sent_messages = [] with patch.object(session_obj, "send", side_effect=lambda t: sent_messages.append(t)): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="테스트 인사입니다."): session_obj.start(trigger) # 페르소나 입장 메시지 (인덱스 1)에 헤르메스와 개발1팀 정보가 있어야 한다 persona_msg = sent_messages[1] assert "헤르메스" in persona_msg assert "개발1팀" in persona_msg # --------------------------------------------------------------------------- # 10. GroupChatSession.end # --------------------------------------------------------------------------- class TestGroupChatSessionEnd: """GroupChatSession.end 퇴장 시퀀스 테스트.""" def _prepare(self, session_obj, personas=None): """테스트용 세션 상태 준비 헬퍼.""" if personas is None: personas = ["hermes", "athena"] session_obj.active = True session_obj.topic = "종료 테스트" session_obj.personas = personas session_obj.speak_counts = {p: 1 for p in personas} session_obj.history = [] def test_end_sets_active_false(self, session_obj, tmp_path): """end() 후 active가 False가 된다.""" self._prepare(session_obj) session_file = tmp_path / "group_chat_session.json" with patch.object(session_obj, "send"): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="수고하셨습니다."): session_obj.end("user_exit") assert session_obj.active is False def test_end_removes_session_file(self, session_obj, tmp_path): """end() 후 세션 파일이 삭제된다.""" self._prepare(session_obj) session_file = tmp_path / "group_chat_session.json" session_file.write_text(json.dumps({"active": True}), encoding="utf-8") with patch.object(session_obj, "send"): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="수고하셨습니다."): session_obj.end("user_exit") assert not session_file.exists() def test_end_sends_farewell_per_persona_plus_summary(self, session_obj, tmp_path): """end()는 페르소나별 작별 인사 1회 + 종료 요약 1회 = N+1 회 send를 호출한다.""" personas = ["hermes", "athena", "thor"] self._prepare(session_obj, personas) session_file = tmp_path / "no_session.json" sent_calls = [] with patch.object(session_obj, "send", side_effect=lambda t: sent_calls.append(t)): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="수고하셨습니다."): session_obj.end("user_exit") # 페르소나 3명 → 3회 + 종료 요약 1회 = 4회 assert len(sent_calls) == 4 def test_end_final_message_contains_topic(self, session_obj, tmp_path): """end()의 마지막 메시지에는 주제가 포함된다.""" self._prepare(session_obj, ["iris"]) session_obj.topic = "UI 개선 논의" sent_calls = [] session_file = tmp_path / "no_session.json" with patch.object(session_obj, "send", side_effect=lambda t: sent_calls.append(t)): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="수고하셨습니다."): session_obj.end("user_exit") assert "UI 개선 논의" in sent_calls[-1] def test_end_api_failure_uses_fallback_farewell(self, session_obj, tmp_path): """CLI 오류 시 폴백 작별 인사('수고하셨습니다, 제이회장님.')가 전송된다.""" self._prepare(session_obj, ["loki"]) sent_calls = [] session_file = tmp_path / "no_session.json" with patch.object(session_obj, "send", side_effect=lambda t: sent_calls.append(t)): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", side_effect=RuntimeError("CLI 오류")): session_obj.end("user_exit") assert session_obj.active is False farewell_msg = sent_calls[0] assert "수고하셨습니다" in farewell_msg def test_end_farewell_message_contains_new_format(self, session_obj, tmp_path): """end()의 작별 인사 메시지가 '이름(팀/역할)' 포맷을 포함한다.""" self._prepare(session_obj, ["hermes"]) sent_calls = [] session_file = tmp_path / "no_session.json" with patch.object(session_obj, "send", side_effect=lambda t: sent_calls.append(t)): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="수고하셨습니다."): session_obj.end("user_exit") # 첫 번째 메시지(페르소나 작별 인사)에 헤르메스와 팀 정보가 있어야 한다 farewell_msg = sent_calls[0] assert "헤르메스" in farewell_msg assert "개발1팀" in farewell_msg # --------------------------------------------------------------------------- # 11. GroupChatSession.add_user_input # --------------------------------------------------------------------------- class TestAddUserInput: """GroupChatSession.add_user_input 테스트.""" def test_resets_auto_turns(self, session_obj): """add_user_input() 호출 시 auto_turns가 0으로 리셋된다.""" session_obj.auto_turns = 5 session_obj.add_user_input("새 메시지") assert session_obj.auto_turns == 0 def test_appends_to_history(self, session_obj): """add_user_input() 호출 시 히스토리에 메시지가 추가된다.""" session_obj.history = [] session_obj.add_user_input("테스트 입력") assert len(session_obj.history) == 1 assert session_obj.history[0]["content"] == "테스트 입력" assert session_obj.history[0]["speaker"] == "user" assert session_obj.history[0]["speaker_name"] == "제이회장님" def test_updates_last_activity(self, session_obj): """add_user_input() 호출 시 last_activity가 갱신된다.""" session_obj.last_activity = 0.0 session_obj.add_user_input("활동 시간 갱신") assert session_obj.last_activity > 0.0 assert session_obj.last_activity <= time.time() def test_multiple_inputs_accumulate_in_history(self, session_obj): """여러 번 호출하면 히스토리가 누적된다.""" session_obj.history = [] session_obj.add_user_input("첫 번째") session_obj.add_user_input("두 번째") session_obj.add_user_input("세 번째") assert len(session_obj.history) == 3 assert session_obj.history[2]["content"] == "세 번째" def test_auto_turns_resets_regardless_of_previous_value(self, session_obj): """auto_turns가 어떤 값이든 0으로 리셋된다.""" for initial in [0, 1, 6, 100]: session_obj.auto_turns = initial session_obj.add_user_input("리셋 확인") assert session_obj.auto_turns == 0 # --------------------------------------------------------------------------- # 12. select_next_speaker # --------------------------------------------------------------------------- class TestSelectNextSpeaker: """select_next_speaker 함수 테스트.""" def test_excludes_last_speaker_in_fallback(self): """CLI 오류 시 폴백에서 직전 발화자를 제외한다.""" gc = _fresh_module() speak_counts = {"hermes": 3, "athena": 1, "thor": 2} persona_names = ["헤르메스", "아테나", "토르"] with patch("group_chat.call_claude", side_effect=RuntimeError("CLI 오류")): result = gc.select_next_speaker( persona_names=persona_names, history=[], last_speaker="athena", speak_counts=speak_counts, ) # 직전 발화자 athena는 제외되어야 한다 assert result != "athena" def test_fallback_picks_least_spoken(self): """CLI 오류 폴백에서 발화 횟수가 가장 적은 화자를 선택한다.""" gc = _fresh_module() # loki가 가장 적게 발화 (0회), hermes는 직전 발화자 speak_counts = {"hermes": 5, "loki": 0, "odin": 3} persona_names = ["헤르메스", "로키", "오딘"] with patch("group_chat.call_claude", side_effect=RuntimeError("CLI 오류")): result = gc.select_next_speaker( persona_names=persona_names, history=[], last_speaker="hermes", speak_counts=speak_counts, ) assert result == "loki" def test_uses_api_response_when_valid(self): """CLI가 유효한 JSON을 반환하면 그 결과를 사용한다.""" gc = _fresh_module() with patch("group_chat.call_claude", return_value='{"next_speaker": "odin", "reason": "자연스러운 순서"}'): result = gc.select_next_speaker( persona_names=["헤르메스", "오딘", "로키"], history=[], last_speaker="hermes", speak_counts={"hermes": 2, "odin": 1, "loki": 2}, ) assert result == "odin" def test_falls_back_when_api_returns_invalid_key(self): """CLI 응답의 persona_key가 speak_counts에 없으면 폴백을 사용한다.""" gc = _fresh_module() with patch("group_chat.call_claude", return_value='{"next_speaker": "zeus", "reason": "없는 키"}'): result = gc.select_next_speaker( persona_names=["헤르메스", "아테나"], history=[], last_speaker="hermes", speak_counts={"hermes": 3, "athena": 0}, ) # 폴백: hermes 제외, athena 선택 (0회 발화) assert result == "athena" def test_fallback_allows_last_speaker_when_only_one_candidate(self): """후보가 한 명뿐이고 직전 발화자와 같으면 그 화자를 선택한다.""" gc = _fresh_module() with patch("group_chat.call_claude", side_effect=RuntimeError("CLI 오류")): result = gc.select_next_speaker( persona_names=["헤르메스"], history=[], last_speaker="hermes", speak_counts={"hermes": 2}, ) # 후보가 없으므로 전체 목록으로 폴백 → hermes 선택 assert result == "hermes" def test_api_returns_json_with_surrounding_text(self): """CLI 응답이 JSON 앞뒤에 텍스트를 포함해도 올바르게 파싱한다.""" gc = _fresh_module() with patch( "group_chat.call_claude", return_value='다음 발화자를 선택했습니다. {"next_speaker": "thor", "reason": "순서"} 감사합니다.', ): result = gc.select_next_speaker( persona_names=["헤르메스", "토르", "아테나"], history=[], last_speaker="athena", speak_counts={"hermes": 1, "thor": 0, "athena": 2}, ) assert result == "thor" # --------------------------------------------------------------------------- # 13. MAX_AUTO_TURNS 제한 # --------------------------------------------------------------------------- class TestMaxAutoTurns: """MAX_AUTO_TURNS >= 6 시 대기 동작 테스트.""" def test_max_auto_turns_constant_is_six(self): """MAX_AUTO_TURNS 상수가 6이다.""" gc = _fresh_module() assert gc.MAX_AUTO_TURNS == 6 def test_run_loop_does_not_select_speaker_when_at_limit(self, session_obj): """auto_turns >= MAX_AUTO_TURNS 이면 select_next_speaker를 호출하지 않는다.""" session_obj.active = True session_obj.auto_turns = 6 # 한계치 session_obj.topic = "자동 턴 테스트" session_obj.personas = ["hermes"] session_obj.speak_counts = {"hermes": 6} session_obj.last_speaker = "hermes" session_obj.last_activity = time.time() call_count = {"n": 0} def fake_read_trigger(): call_count["n"] += 1 if call_count["n"] == 1: # 첫 번째 폴링: auto_turns 한계로 슬립 후 continue return None # 두 번째 폴링: 종료 트리거 return {"action": "end", "reason": "test_exit"} with patch("group_chat.read_trigger", side_effect=fake_read_trigger): with patch("group_chat.select_next_speaker") as mock_select: with patch("group_chat.save_session_log"): with patch.object(session_obj, "send"): with patch("time.sleep"): with patch("group_chat.SESSION_FILE", "/tmp/no_session_auto_turns.json"): session_obj.run_loop() # auto_turns >= MAX_AUTO_TURNS 구간에서는 select_next_speaker를 호출하지 않아야 한다 mock_select.assert_not_called() def test_auto_turns_increments_on_speak(self, session_obj): """speak() 호출마다 auto_turns가 1 증가한다.""" session_obj.active = True session_obj.topic = "증가 테스트" session_obj.personas = ["hermes"] session_obj.speak_counts = {"hermes": 0} session_obj.auto_turns = 0 session_obj.history = [] with patch("group_chat.generate_persona_response", return_value="테스트 응답"): with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): session_obj.speak("hermes") assert session_obj.auto_turns == 1 def test_auto_turns_increments_cumulatively(self, session_obj): """speak()를 여러 번 호출할수록 auto_turns가 누적된다.""" session_obj.active = True session_obj.topic = "누적 테스트" session_obj.personas = ["hermes"] session_obj.speak_counts = {"hermes": 0} session_obj.auto_turns = 0 session_obj.history = [] with patch("group_chat.generate_persona_response", return_value="테스트 응답"): with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): for _ in range(4): session_obj.speak("hermes") assert session_obj.auto_turns == 4 def test_add_user_input_resets_auto_turns_to_zero(self, session_obj): """add_user_input() 후에는 auto_turns가 0으로 리셋된다.""" session_obj.auto_turns = 6 # 한계 도달 상태 session_obj.add_user_input("유저가 새로 입력했어요") assert session_obj.auto_turns == 0 def test_run_loop_resumes_after_user_input_resets_auto_turns(self, session_obj): """유저 입력으로 auto_turns가 리셋되면 루프가 다시 발화자를 선택한다.""" session_obj.active = True session_obj.auto_turns = 6 # 한계치 session_obj.topic = "재개 테스트" session_obj.personas = ["athena"] session_obj.speak_counts = {"athena": 6} session_obj.last_speaker = "athena" session_obj.last_activity = time.time() call_count = {"n": 0} def fake_read_trigger(): call_count["n"] += 1 if call_count["n"] == 1: # 유저 입력 트리거 → auto_turns 리셋 return {"action": "user_input", "message": "계속해주세요"} if call_count["n"] == 2: # 종료 트리거 return {"action": "end", "reason": "done"} return None with patch("group_chat.read_trigger", side_effect=fake_read_trigger): with patch("group_chat.select_next_speaker", return_value="athena") as mock_select: with patch("group_chat.save_session_log"): with patch.object(session_obj, "send"): with patch("group_chat.generate_persona_response", return_value="응답"): with patch("group_chat.dump_session"): with patch("time.sleep"): with patch("group_chat.SESSION_FILE", "/tmp/no_session_resume.json"): session_obj.run_loop() # 유저 입력 후 auto_turns가 리셋되었으므로 select_next_speaker가 호출되어야 한다 mock_select.assert_called() # --------------------------------------------------------------------------- # 14. TestTelegramPoller # --------------------------------------------------------------------------- class TestTelegramPoller: """TelegramPoller 클래스 테스트.""" def test_init_sets_attributes(self): """__init__이 token, chat_id, base_url, last_update_id를 올바르게 초기화한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="test-token", chat_id="12345") assert poller.token == "test-token" assert poller.chat_id == "12345" assert poller.base_url == "https://api.telegram.org/bottest-token" assert poller.last_update_id == 0 def test_get_updates_returns_messages_on_success(self): """정상 응답 시 메시지 리스트를 반환한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") fake_resp = MagicMock() fake_resp.json.return_value = { "ok": True, "result": [ { "update_id": 101, "message": { "chat": {"id": 999}, "text": "안녕하세요", }, } ], } with patch("group_chat.requests.get", return_value=fake_resp): messages = poller.get_updates() assert messages == ["안녕하세요"] def test_get_updates_filters_by_chat_id(self): """chat_id가 불일치하는 메시지는 제외하고 빈 리스트를 반환한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") fake_resp = MagicMock() fake_resp.json.return_value = { "ok": True, "result": [ { "update_id": 102, "message": { "chat": {"id": 888}, # 다른 chat_id "text": "다른 채팅방", }, } ], } with patch("group_chat.requests.get", return_value=fake_resp): messages = poller.get_updates() assert messages == [] def test_get_updates_returns_empty_on_api_error(self): """API 응답의 ok가 False이면 빈 리스트를 반환한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") fake_resp = MagicMock() fake_resp.json.return_value = {"ok": False} with patch("group_chat.requests.get", return_value=fake_resp): messages = poller.get_updates() assert messages == [] def test_get_updates_returns_empty_on_exception(self): """requests.get이 예외를 발생시키면 빈 리스트를 반환한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") with patch("group_chat.requests.get", side_effect=Exception("Network error")): messages = poller.get_updates() assert messages == [] def test_get_updates_updates_last_update_id(self): """get_updates() 호출 후 last_update_id가 최신 update_id로 갱신된다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") assert poller.last_update_id == 0 fake_resp = MagicMock() fake_resp.json.return_value = { "ok": True, "result": [ { "update_id": 201, "message": {"chat": {"id": 999}, "text": "첫 메시지"}, }, { "update_id": 205, "message": {"chat": {"id": 999}, "text": "두 번째 메시지"}, }, ], } with patch("group_chat.requests.get", return_value=fake_resp): poller.get_updates() assert poller.last_update_id == 205 def test_get_updates_returns_empty_on_no_results(self): """result가 빈 리스트이면 빈 리스트를 반환하고 last_update_id는 변하지 않는다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") fake_resp = MagicMock() fake_resp.json.return_value = {"ok": True, "result": []} with patch("group_chat.requests.get", return_value=fake_resp): messages = poller.get_updates() assert messages == [] assert poller.last_update_id == 0 def test_get_updates_skips_messages_without_text(self): """text 필드가 없는 메시지(사진, 스티커 등)는 스킵한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") fake_resp = MagicMock() fake_resp.json.return_value = { "ok": True, "result": [ { "update_id": 301, "message": { "chat": {"id": 999}, # text 필드 없음 (사진 메시지 등) }, } ], } with patch("group_chat.requests.get", return_value=fake_resp): messages = poller.get_updates() # update_id는 갱신되지만 메시지 리스트는 비어야 한다 assert messages == [] assert poller.last_update_id == 301 def test_get_updates_passes_correct_offset(self): """get_updates()가 last_update_id + 1을 offset으로 전달한다.""" gc = _fresh_module() poller = gc.TelegramPoller(token="tok", chat_id="999") poller.last_update_id = 100 fake_resp = MagicMock() fake_resp.json.return_value = {"ok": True, "result": []} with patch("group_chat.requests.get", return_value=fake_resp) as mock_get: poller.get_updates() call_kwargs = mock_get.call_args params = ( call_kwargs[1].get("params") or call_kwargs[0][1] if len(call_kwargs[0]) > 1 else call_kwargs[1]["params"] ) assert params["offset"] == 101 # --------------------------------------------------------------------------- # 15. TestDetectIntent # --------------------------------------------------------------------------- class TestDetectIntent: """detect_intent 함수 테스트.""" # ── session_active=False (세션 없음) 상태 ────────────────── def test_start_keyword_team_gather(self): """'팀 모여' 키워드 → start_chat 인텐트 반환 (세션 없음).""" gc = _fresh_module() result = gc.detect_intent("팀 모여", session_active=False) assert result["intent"] == "start_chat" def test_start_keyword_have_meeting(self): """'회의하자' 키워드 → start_chat 인텐트 반환 (세션 없음).""" gc = _fresh_module() result = gc.detect_intent("회의하자", session_active=False) assert result["intent"] == "start_chat" def test_start_chat_result_has_topic_and_personas(self): """start_chat 인텐트에 topic과 personas 필드가 있다.""" gc = _fresh_module() result = gc.detect_intent("팀 모여", session_active=False) assert "topic" in result assert "personas" in result assert isinstance(result["personas"], list) def test_normal_message_calls_claude_when_no_session(self): """일반 메시지 '안녕' → Claude 호출 후 none 반환 (세션 없음).""" gc = _fresh_module() fake_response = '{"intent": "none"}' with patch("group_chat.call_claude", return_value=fake_response) as mock_claude: result = gc.detect_intent("안녕", session_active=False) mock_claude.assert_called_once() assert result["intent"] == "none" # ── session_active=True (세션 활성) 상태 ────────────────── def test_end_keyword_sleep(self): """'잘래' 키워드 → end_chat 인텐트 반환 (세션 활성).""" gc = _fresh_module() result = gc.detect_intent("잘래", session_active=True) assert result["intent"] == "end_chat" def test_end_keyword_bye(self): """'빠이' 키워드 → end_chat 인텐트 반환 (세션 활성).""" gc = _fresh_module() result = gc.detect_intent("빠이", session_active=True) assert result["intent"] == "end_chat" def test_normal_question_calls_claude_when_session_active(self): """일반 질문 → Claude 호출 → user_input 반환 (세션 활성).""" gc = _fresh_module() fake_response = '{"intent": "user_input", "message": "오늘 일정 어때?"}' with patch("group_chat.call_claude", return_value=fake_response) as mock_claude: result = gc.detect_intent("오늘 일정 어때?", session_active=True) mock_claude.assert_called_once() assert result["intent"] == "user_input" def test_claude_fallback_when_session_active(self): """Claude 호출 실패 시 session_active=True → user_input 폴백.""" gc = _fresh_module() with patch("group_chat.call_claude", side_effect=Exception("Claude 실패")): result = gc.detect_intent("어떻게 생각해?", session_active=True) assert result["intent"] == "user_input" assert result["message"] == "어떻게 생각해?" def test_claude_fallback_when_session_inactive(self): """Claude 호출 실패 시 session_active=False → none 폴백.""" gc = _fresh_module() with patch("group_chat.call_claude", side_effect=Exception("Claude 실패")): result = gc.detect_intent("안녕", session_active=False) assert result["intent"] == "none" def test_claude_returns_invalid_json_fallback_active(self): """Claude가 유효하지 않은 JSON 반환 시 session_active=True → user_input 폴백.""" gc = _fresh_module() with patch("group_chat.call_claude", return_value="유효하지 않은 응답"): result = gc.detect_intent("어떤 질문", session_active=True) assert result["intent"] == "user_input" def test_claude_returns_invalid_json_fallback_inactive(self): """Claude가 유효하지 않은 JSON 반환 시 session_active=False → none 폴백.""" gc = _fresh_module() with patch("group_chat.call_claude", return_value="유효하지 않은 응답"): result = gc.detect_intent("어떤 질문", session_active=False) assert result["intent"] == "none" def test_start_keywords_not_matched_when_session_active(self): """세션 활성 중에는 시작 키워드가 end_chat을 유발하지 않는다.""" gc = _fresh_module() with patch("group_chat.call_claude", return_value='{"intent": "user_input", "message": "팀 모여"}'): result = gc.detect_intent("팀 모여", session_active=True) # 세션 활성 중 시작 키워드는 종료 처리가 아닌 user_input 으로 처리됨 assert result["intent"] != "end_chat" def test_end_keywords_not_matched_when_session_inactive(self): """세션 비활성 중에는 종료 키워드가 즉각 end_chat을 반환하지 않는다.""" gc = _fresh_module() # 세션 비활성이면 END_KEYWORDS 체크를 안 하고 Claude 호출 with patch("group_chat.call_claude", return_value='{"intent": "none"}'): result = gc.detect_intent("잘래", session_active=False) # start_chat이나 none 중 하나 (end_chat이 아님) assert result["intent"] != "end_chat" # --------------------------------------------------------------------------- # 16. TestIsActive # --------------------------------------------------------------------------- class TestIsActive: """GroupChatSession.is_active() 메서드 테스트.""" def test_is_active_returns_false_by_default(self, session_obj): """초기 상태에서 is_active()는 False를 반환한다.""" assert session_obj.is_active() is False def test_is_active_returns_false_when_active_false(self, session_obj): """active=False로 설정하면 is_active()가 False를 반환한다.""" session_obj.active = False assert session_obj.is_active() is False def test_is_active_returns_true_after_start(self, session_obj): """start() 호출 후 is_active()가 True를 반환한다.""" trigger = {"topic": "테스트", "personas": ["hermes"]} with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="인사"): session_obj.start(trigger) assert session_obj.is_active() is True def test_is_active_returns_false_after_end(self, session_obj, tmp_path): """start() 후 end() 호출하면 is_active()가 다시 False를 반환한다.""" trigger = {"topic": "테스트", "personas": ["hermes"]} session_file = tmp_path / "session.json" with patch.object(session_obj, "send"): with patch("group_chat.dump_session"): with patch("group_chat.call_claude", return_value="인사"): session_obj.start(trigger) assert session_obj.is_active() is True with patch.object(session_obj, "send"): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="작별"): session_obj.end("user_exit") assert session_obj.is_active() is False def test_is_active_reflects_active_attribute(self, session_obj): """is_active()가 self.active 속성과 일치하는 값을 반환한다.""" session_obj.active = True assert session_obj.is_active() is True session_obj.active = False assert session_obj.is_active() is False # --------------------------------------------------------------------------- # 17. TestRunOneTurn # --------------------------------------------------------------------------- class TestRunOneTurn: """GroupChatSession.run_one_turn() 메서드 테스트.""" def _make_active_session(self, session_obj): """테스트용 활성 세션 상태를 만드는 헬퍼.""" session_obj.active = True session_obj.topic = "런원턴 테스트" session_obj.personas = ["hermes", "athena"] session_obj.speak_counts = {"hermes": 0, "athena": 0} session_obj.last_speaker = "" session_obj.last_activity = time.time() session_obj.auto_turns = 0 session_obj.history = [] def test_does_nothing_when_inactive(self, session_obj): """active=False이면 아무 동작도 하지 않는다.""" session_obj.active = False with patch("group_chat.select_next_speaker") as mock_select: with patch.object(session_obj, "speak") as mock_speak: session_obj.run_one_turn() mock_select.assert_not_called() mock_speak.assert_not_called() def test_calls_end_on_timeout(self, session_obj, tmp_path): """last_activity가 SESSION_TIMEOUT 초과 시 end()를 호출한다.""" session_obj.active = True session_obj.topic = "타임아웃 테스트" session_obj.personas = ["hermes"] session_obj.speak_counts = {"hermes": 0} session_obj.last_speaker = "" # 타임아웃보다 오래된 활동 시간 설정 session_obj.last_activity = time.time() - 400 # SESSION_TIMEOUT(300) 초과 session_obj.auto_turns = 0 session_obj.history = [] session_file = tmp_path / "session.json" with patch.object(session_obj, "send"): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="작별"): session_obj.run_one_turn() assert session_obj.active is False def test_does_not_speak_when_max_auto_turns_reached(self, session_obj): """auto_turns >= MAX_AUTO_TURNS이면 speak()를 호출하지 않는다.""" self._make_active_session(session_obj) session_obj.auto_turns = 6 # MAX_AUTO_TURNS 도달 with patch("group_chat.select_next_speaker") as mock_select: with patch.object(session_obj, "speak") as mock_speak: session_obj.run_one_turn() mock_select.assert_not_called() mock_speak.assert_not_called() def test_calls_speak_on_normal_turn(self, session_obj): """정상 상태에서 run_one_turn()은 speak()를 호출한다.""" self._make_active_session(session_obj) with patch("group_chat.select_next_speaker", return_value="hermes"): with patch.object(session_obj, "speak", return_value=True) as mock_speak: session_obj.run_one_turn() mock_speak.assert_called_once_with("hermes") def test_fallback_speaker_on_select_error(self, session_obj): """select_next_speaker 오류 시 personas[0]를 폴백으로 사용해 speak()를 호출한다.""" self._make_active_session(session_obj) with patch("group_chat.select_next_speaker", side_effect=Exception("선택 오류")): with patch.object(session_obj, "speak", return_value=True) as mock_speak: session_obj.run_one_turn() # 폴백으로 personas[0]("hermes")이 선택되어야 한다 mock_speak.assert_called_once_with("hermes") def test_ends_session_on_speak_exception(self, session_obj, tmp_path): """speak() 중 예외 발생 시 세션을 종료한다.""" self._make_active_session(session_obj) session_file = tmp_path / "session.json" with patch("group_chat.select_next_speaker", return_value="hermes"): with patch.object(session_obj, "speak", side_effect=Exception("발화 오류")): with patch.object(session_obj, "send"): with patch("group_chat.SESSION_FILE", str(session_file)): with patch("group_chat.save_session_log"): with patch("group_chat.call_claude", return_value="작별"): session_obj.run_one_turn() assert session_obj.active is False # --------------------------------------------------------------------------- # 18. TestCheckTriggerFile # --------------------------------------------------------------------------- class TestCheckTriggerFile: """check_trigger_file 함수 테스트.""" def test_does_nothing_when_no_trigger(self): """트리거 파일이 없으면 아무 동작도 하지 않는다.""" gc = _fresh_module() session = MagicMock() with patch("group_chat.read_trigger", return_value=None): gc.check_trigger_file(session, "fake-token") session.start.assert_not_called() session.add_user_input.assert_not_called() session.end.assert_not_called() def test_action_start_calls_session_start(self): """action=start 트리거 → session.start() 호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = False trigger = {"action": "start", "topic": "배포 논의", "personas": ["hermes"]} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.start.assert_called_once_with(trigger) def test_action_user_input_with_active_session(self): """action=user_input + 활성 세션 → session.add_user_input() 호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = True trigger = {"action": "user_input", "message": "좋은 의견이에요"} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.add_user_input.assert_called_once_with("좋은 의견이에요") def test_action_end_with_active_session(self): """action=end + 활성 세션 → session.end() 호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = True trigger = {"action": "end", "reason": "user_exit"} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.end.assert_called_once_with("user_exit") def test_action_user_input_with_inactive_session_is_ignored(self): """action=user_input + 비활성 세션 → add_user_input() 미호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = False trigger = {"action": "user_input", "message": "무시될 메시지"} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.add_user_input.assert_not_called() def test_action_end_with_inactive_session_is_ignored(self): """action=end + 비활성 세션 → end() 미호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = False trigger = {"action": "end", "reason": "user_exit"} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.end.assert_not_called() def test_action_start_ends_existing_session_first(self): """action=start + 이미 활성 세션 → 기존 세션 end() 후 start() 호출.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = True # 이미 활성 세션 trigger = {"action": "start", "topic": "새 회의", "personas": ["athena"]} with patch("group_chat.read_trigger", return_value=trigger): gc.check_trigger_file(session, "fake-token") session.end.assert_called_once_with("restart") session.start.assert_called_once_with(trigger) def test_action_start_error_sends_telegram_notification(self): """session.start() 오류 시 Telegram 알림을 전송한다.""" gc = _fresh_module() session = MagicMock() session.is_active.return_value = False session.start.side_effect = Exception("시작 오류") trigger = {"action": "start", "topic": "오류 테스트", "chat_id": "999"} with patch("group_chat.read_trigger", return_value=trigger): with patch("group_chat.send_telegram") as mock_send_tg: gc.check_trigger_file(session, "fake-token") mock_send_tg.assert_called_once() # --------------------------------------------------------------------------- # 19. TestNewConstants # --------------------------------------------------------------------------- class TestNewConstants: """새로 추가된 상수 검증 테스트.""" def test_start_keywords_exists(self): """START_KEYWORDS 상수가 존재한다.""" gc = _fresh_module() assert hasattr(gc, "START_KEYWORDS") def test_start_keywords_is_list(self): """START_KEYWORDS가 리스트 타입이다.""" gc = _fresh_module() assert isinstance(gc.START_KEYWORDS, list) def test_start_keywords_contains_expected_items(self): """START_KEYWORDS에 '팀 모여', '회의하자'가 포함된다.""" gc = _fresh_module() assert "팀 모여" in gc.START_KEYWORDS assert "회의하자" in gc.START_KEYWORDS def test_end_keywords_exists(self): """END_KEYWORDS 상수가 존재한다.""" gc = _fresh_module() assert hasattr(gc, "END_KEYWORDS") def test_end_keywords_is_list(self): """END_KEYWORDS가 리스트 타입이다.""" gc = _fresh_module() assert isinstance(gc.END_KEYWORDS, list) def test_end_keywords_contains_expected_items(self): """END_KEYWORDS에 '잘래', '빠이'가 포함된다.""" gc = _fresh_module() assert "잘래" in gc.END_KEYWORDS assert "빠이" in gc.END_KEYWORDS def test_all_persona_ids_exists(self): """ALL_PERSONA_IDS 상수가 존재한다.""" gc = _fresh_module() assert hasattr(gc, "ALL_PERSONA_IDS") def test_all_persona_ids_is_list(self): """ALL_PERSONA_IDS가 리스트 타입이다.""" gc = _fresh_module() assert isinstance(gc.ALL_PERSONA_IDS, list) def test_all_persona_ids_has_19_members(self): """ALL_PERSONA_IDS에 정확히 19명이 있다.""" gc = _fresh_module() assert len(gc.ALL_PERSONA_IDS) == 19 def test_all_persona_ids_contains_expected_members(self): """ALL_PERSONA_IDS에 주요 페르소나가 포함된다.""" gc = _fresh_module() expected = { "hermes", "vulcan", "iris", "athena", "argos", "odin", "thor", "freya", "mimir", "heimdall", "ra", "anubis", "isis", "sobek", "horus", "loki", "maat", "janus", "venus", } assert expected == set(gc.ALL_PERSONA_IDS) def test_all_persona_ids_no_duplicates(self): """ALL_PERSONA_IDS에 중복이 없다.""" gc = _fresh_module() assert len(gc.ALL_PERSONA_IDS) == len(set(gc.ALL_PERSONA_IDS)) # --------------------------------------------------------------------------- # 20. TestDetectIntentControl # --------------------------------------------------------------------------- class TestDetectIntentControl: """세션 활성 중 제어 명령 감지 테스트 (session_active=True).""" def test_limit_personas_10명만(self): """'10명만 얘기해' → control/limit_personas/count=10.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("10명만 얘기해", session_active=True) assert result["intent"] == "control" assert result["type"] == "limit_personas" assert result["count"] == 10 def test_limit_personas_5명까지만(self): """'5명까지만' → control/limit_personas/count=5.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("5명까지만", session_active=True) assert result["intent"] == "control" assert result["type"] == "limit_personas" assert result["count"] == 5 def test_remove_persona_로키(self): """'로키 빠져' → control/remove_persona/persona='loki'.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("로키 빠져", session_active=True) assert result["intent"] == "control" assert result["type"] == "remove_persona" assert result["persona"] == "loki" def test_remove_persona_나가(self): """'아테나 나가' → control/remove_persona/persona='athena'.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("아테나 나가", session_active=True) assert result["intent"] == "control" assert result["type"] == "remove_persona" assert result["persona"] == "athena" def test_add_persona_불러(self): """'비너스 불러' → control/add_persona/persona='venus'.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("비너스 불러", session_active=True) assert result["intent"] == "control" assert result["type"] == "add_persona" assert result["persona"] == "venus" def test_add_persona_합류(self): """'로키 합류시켜' → control/add_persona/persona='loki'.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("로키 합류시켜", session_active=True) assert result["intent"] == "control" assert result["type"] == "add_persona" assert result["persona"] == "loki" def test_filter_by_role_백엔드(self): """'백엔드만 남아' → control/filter_by_role/personas=['vulcan','thor','anubis'].""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("백엔드만 남아", session_active=True) assert result["intent"] == "control" assert result["type"] == "filter_by_role" assert set(result["personas"]) == {"vulcan", "thor", "anubis"} def test_filter_by_team_1팀(self): """'1팀만' → control/filter_by_team.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("1팀만", session_active=True) assert result["intent"] == "control" assert result["type"] == "filter_by_team" assert set(result["personas"]) == {"hermes", "vulcan", "iris", "athena", "argos"} def test_set_auto_turns_3턴(self): """'3턴까지만 자동으로' → control/set_auto_turns/count=3.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("3턴까지만 자동으로", session_active=True) assert result["intent"] == "control" assert result["type"] == "set_auto_turns" assert result["count"] == 3 def test_set_auto_turns_계속(self): """'계속 얘기해' → control/set_auto_turns/count=99.""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("계속 얘기해", session_active=True) assert result["intent"] == "control" assert result["type"] == "set_auto_turns" assert result["count"] == 99 def test_end_chat_still_works(self): """'끝' → end_chat (기존 기능 유지).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("끝", session_active=True) assert result["intent"] == "end_chat" def test_normal_user_input(self): """일반 메시지는 제어 명령이 아닌 경우 Claude 호출 후 user_input 반환.""" gc = _fresh_module() fake_response = '{"intent": "user_input", "message": "오늘 어떻게 진행할까요?"}' with patch("group_chat.call_claude", return_value=fake_response): result = gc.detect_intent("오늘 어떻게 진행할까요?", session_active=True) assert result["intent"] == "user_input" # --------------------------------------------------------------------------- # 21. TestDetectIntentStartPersonas # --------------------------------------------------------------------------- class TestDetectIntentStartPersonas: """시작 시 자연어 인원 지정 테스트 (session_active=False).""" def test_전원_집합(self): """'전원 집합' → personas 19명 (ALL_PERSONA_IDS 전체).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("전원 집합", session_active=False) assert result["intent"] == "start_chat" assert set(result["personas"]) == set(gc.ALL_PERSONA_IDS) assert len(result["personas"]) == 19 def test_백엔드만_모여(self): """'백엔드만 소집' → personas=['vulcan','thor','anubis'] (소집 키워드 포함).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("백엔드만 소집", session_active=False) assert result["intent"] == "start_chat" assert set(result["personas"]) == {"vulcan", "thor", "anubis"} def test_1팀_모여(self): """'1팀 모여봐' → personas=개발1팀 5명 (모여봐 키워드 포함).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("1팀 모여봐", session_active=False) assert result["intent"] == "start_chat" assert set(result["personas"]) == {"hermes", "vulcan", "iris", "athena", "argos"} def test_팀장들_모여(self): """'팀장 소집' → personas=['hermes','odin','ra'] (소집 키워드 포함).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("팀장 소집", session_active=False) assert result["intent"] == "start_chat" assert set(result["personas"]) == {"hermes", "odin", "ra"} def test_5명만_모여(self): """'전원 집합 5명만' → personas 5명으로 제한 (전원 집합 키워드 + 인원 제한).""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("전원 집합 5명만", session_active=False) assert result["intent"] == "start_chat" assert len(result["personas"]) == 5 def test_기본_소집(self): """'팀 모여' → 기본 3명 ['hermes','athena','thor'].""" gc = _fresh_module() with patch("group_chat.call_claude"): result = gc.detect_intent("팀 모여", session_active=False) assert result["intent"] == "start_chat" assert result["personas"] == ["hermes", "athena", "thor"] # --------------------------------------------------------------------------- # 22. TestHandleControl # --------------------------------------------------------------------------- # 전체 페르소나 personas_data (handle_control 테스트용) _ALL_PERSONAS_DATA = { "hermes": { "name": "헤르메스", "team": "개발1팀", "role": "개발1팀장", "expertise": "", "personality": "", "persona_desc": "", }, "vulcan": { "name": "불칸", "team": "개발1팀", "role": "백엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "iris": { "name": "이리스", "team": "개발1팀", "role": "프론트엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "athena": { "name": "아테나", "team": "개발1팀", "role": "UX/UI 설계자", "expertise": "", "personality": "", "persona_desc": "", }, "argos": { "name": "아르고스", "team": "개발1팀", "role": "QA 엔지니어", "expertise": "", "personality": "", "persona_desc": "", }, "odin": { "name": "오딘", "team": "개발2팀", "role": "개발2팀장", "expertise": "", "personality": "", "persona_desc": "", }, "thor": { "name": "토르", "team": "개발2팀", "role": "백엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "freya": { "name": "프레이야", "team": "개발2팀", "role": "프론트엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "mimir": { "name": "미미르", "team": "개발2팀", "role": "UX 설계자", "expertise": "", "personality": "", "persona_desc": "", }, "heimdall": { "name": "헤임달", "team": "개발2팀", "role": "QA 엔지니어", "expertise": "", "personality": "", "persona_desc": "", }, "ra": { "name": "라", "team": "개발3팀", "role": "개발3팀장", "expertise": "", "personality": "", "persona_desc": "", }, "anubis": { "name": "아누비스", "team": "개발3팀", "role": "백엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "isis": { "name": "이시스", "team": "개발3팀", "role": "프론트엔드 개발자", "expertise": "", "personality": "", "persona_desc": "", }, "sobek": { "name": "소베크", "team": "개발3팀", "role": "UX 설계자", "expertise": "", "personality": "", "persona_desc": "", }, "horus": { "name": "호루스", "team": "개발3팀", "role": "QA 엔지니어", "expertise": "", "personality": "", "persona_desc": "", }, "loki": { "name": "로키", "team": "레드팀", "role": "레드팀 리더", "expertise": "", "personality": "", "persona_desc": "", }, "maat": { "name": "마아트", "team": "QC 센터", "role": "QC 센터장", "expertise": "", "personality": "", "persona_desc": "", }, "janus": { "name": "야누스", "team": "DevOps 센터", "role": "DevOps 센터장", "expertise": "", "personality": "", "persona_desc": "", }, "venus": { "name": "비너스", "team": "디자인 센터", "role": "디자인 센터장", "expertise": "", "personality": "", "persona_desc": "", }, } class TestHandleControl: """handle_control 메서드 테스트.""" def _make_session(self, personas=None, speak_counts=None): """테스트용 GroupChatSession 인스턴스를 생성하는 헬퍼.""" gc = _fresh_module() session = gc.GroupChatSession(token="test-token", personas_data=_ALL_PERSONAS_DATA) session.active = True session.chat_id = "999" if personas is not None: session.personas = list(personas) else: session.personas = list(_ALL_PERSONAS_DATA.keys()) # 19명 if speak_counts is not None: session.speak_counts = dict(speak_counts) else: session.speak_counts = {p: 0 for p in session.personas} return session def test_limit_personas(self): """19명에서 5명으로 축소하고 send_system_message를 호출한다.""" session = self._make_session() assert len(session.personas) == 19 with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "limit_personas", "count": 5}) assert len(session.personas) == 5 assert len(session.speak_counts) == 5 mock_send_tg.assert_called_once() # 전송된 메시지에 ⚙️ 이모지가 포함되어야 한다 sent_text = mock_send_tg.call_args[0][2] assert "⚙️" in sent_text def test_limit_personas_zero_ignored(self): """count=0이면 무시되고 personas가 변경되지 않는다.""" session = self._make_session(personas=["hermes", "athena", "thor"]) with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "limit_personas", "count": 0}) assert len(session.personas) == 3 mock_send_tg.assert_not_called() def test_remove_persona(self): """특정 인원을 퇴장시키고 speak_counts에서도 제거한다.""" session = self._make_session( personas=["hermes", "athena", "thor"], speak_counts={"hermes": 2, "athena": 1, "thor": 3}, ) with patch("group_chat.send_telegram"): session.handle_control({"type": "remove_persona", "persona": "athena"}) assert "athena" not in session.personas assert "athena" not in session.speak_counts assert len(session.personas) == 2 def test_remove_persona_last_one(self): """1명만 남았을 때 제거 시도 → 무시하고 그대로 유지한다.""" session = self._make_session( personas=["hermes"], speak_counts={"hermes": 0}, ) with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "remove_persona", "persona": "hermes"}) assert "hermes" in session.personas assert len(session.personas) == 1 mock_send_tg.assert_not_called() def test_add_persona(self): """새 인원을 합류시키고 speak_counts에 0으로 추가한다.""" session = self._make_session( personas=["hermes", "athena"], speak_counts={"hermes": 1, "athena": 2}, ) with patch("group_chat.send_telegram"): session.handle_control({"type": "add_persona", "persona": "loki"}) assert "loki" in session.personas assert session.speak_counts["loki"] == 0 assert len(session.personas) == 3 def test_add_persona_duplicate(self): """이미 있는 인원을 추가하면 무시하고 중복 추가하지 않는다.""" session = self._make_session( personas=["hermes", "athena"], speak_counts={"hermes": 1, "athena": 2}, ) with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "add_persona", "persona": "hermes"}) assert session.personas.count("hermes") == 1 assert len(session.personas) == 2 mock_send_tg.assert_not_called() def test_filter_by_role(self): """역할 필터를 적용해 personas를 교체한다.""" session = self._make_session( personas=["hermes", "athena", "loki"], speak_counts={"hermes": 1, "athena": 0, "loki": 2}, ) with patch("group_chat.send_telegram"): session.handle_control( { "type": "filter_by_role", "personas": ["vulcan", "thor", "anubis"], "role": "백엔드", } ) assert set(session.personas) == {"vulcan", "thor", "anubis"} # 기존 hermes/athena/loki speak_counts가 정리되어야 함 assert "hermes" not in session.speak_counts assert "athena" not in session.speak_counts assert "loki" not in session.speak_counts # 새 인원 speak_counts 초기화 assert session.speak_counts["vulcan"] == 0 assert session.speak_counts["thor"] == 0 assert session.speak_counts["anubis"] == 0 def test_filter_by_team(self): """팀 필터를 적용해 personas를 교체한다.""" session = self._make_session( personas=["hermes", "athena", "loki"], speak_counts={"hermes": 1, "athena": 0, "loki": 2}, ) with patch("group_chat.send_telegram"): session.handle_control( { "type": "filter_by_team", "personas": ["odin", "thor", "freya", "mimir", "heimdall"], "team": "2팀", } ) assert set(session.personas) == {"odin", "thor", "freya", "mimir", "heimdall"} assert "hermes" not in session.speak_counts assert session.speak_counts.get("odin", -1) == 0 def test_set_auto_turns(self): """max_auto_turns를 설정하고 auto_turns를 리셋한다.""" session = self._make_session() session.auto_turns = 5 with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "set_auto_turns", "count": 3}) assert session.max_auto_turns == 3 assert session.auto_turns == 0 mock_send_tg.assert_called_once() sent_text = mock_send_tg.call_args[0][2] assert "3" in sent_text def test_set_auto_turns_unlimited(self): """count=99 → '계속 진행' 메시지를 전송한다.""" session = self._make_session() session.auto_turns = 4 with patch("group_chat.send_telegram") as mock_send_tg: session.handle_control({"type": "set_auto_turns", "count": 99}) assert session.max_auto_turns == 99 assert session.auto_turns == 0 mock_send_tg.assert_called_once() sent_text = mock_send_tg.call_args[0][2] assert "계속" in sent_text # --------------------------------------------------------------------------- # 23. TestSendSystemMessage # --------------------------------------------------------------------------- class TestSendSystemMessage: """send_system_message 메서드 테스트.""" def test_sends_with_gear_emoji(self): """send_system_message는 '⚙️ 텍스트' 형식으로 Telegram에 전송한다.""" gc = _fresh_module() session = gc.GroupChatSession( token="test-token", personas_data={ "hermes": { "name": "헤르메스", "team": "개발1팀", "role": "개발1팀장", "expertise": "", "personality": "", "persona_desc": "", } }, ) session.chat_id = "12345" with patch("group_chat.send_telegram") as mock_send_tg: session.send_system_message("인원 조정 완료") mock_send_tg.assert_called_once_with("test-token", "12345", "⚙️ 인원 조정 완료")