""" test_notify_completion.py scripts/notify-completion.py 단위 테스트 (task-902.1 반영) 테스트 항목: - argparse 파싱 정상 동작 - send_telegram_notification 기반 완료 통보 검증 (requests.post 방식) - subprocess 호출 mock (실제 외부 전송 방지) - 셸 인젝션 방지 (shell=False 패치) - .done.notified 마커 생성 (O_EXCL) """ import json import os import sys from pathlib import Path from unittest.mock import MagicMock, patch _TEST_CHAT_ID = os.environ.get("COKACDIR_CHAT_ID", "6937032012") _WORKSPACE = os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace") import pytest # scripts 디렉토리를 import path에 추가 _SCRIPTS_DIR = Path(__file__).parent.parent sys.path.insert(0, str(_SCRIPTS_DIR)) # notify-completion.py는 하이픈이 있으므로 importlib으로 임포트 import importlib.util _MODULE_PATH = _SCRIPTS_DIR / "notify-completion.py" spec = importlib.util.spec_from_file_location("notify_completion", _MODULE_PATH) assert spec is not None notify_completion = importlib.util.module_from_spec(spec) assert spec.loader is not None spec.loader.exec_module(notify_completion) class TestArgparse: """argparse 파싱 테스트""" def test_help_flag(self): """--help 플래그 정상 동작""" with pytest.raises(SystemExit) as exc_info: notify_completion.main.__wrapped__ if hasattr(notify_completion.main, "__wrapped__") else None import argparse parser = argparse.ArgumentParser(description="팀장 → 아누 완료 통보") parser.add_argument("task_id") parser.add_argument("--chat-id", default=_TEST_CHAT_ID) parser.add_argument("--anu-key", default=None) parser.parse_args(["--help"]) assert exc_info.value.code == 0 class TestSendTelegramNotificationDirect: """send_telegram_notification (requests.post 방식) 직접 테스트""" def test_success_returns_true(self): """200 응답 시 True 반환""" mock_resp = MagicMock() mock_resp.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", return_value=mock_resp) as mock_post: result = notify_completion.send_telegram_notification("chat123", "테스트 메시지") assert result is True mock_post.assert_called_once() call_kwargs = mock_post.call_args assert "chat123" in str(call_kwargs) assert "테스트 메시지" in str(call_kwargs) def test_no_bot_token_returns_false(self): """ANU_BOT_TOKEN 미설정 시 False 반환""" env = {k: v for k, v in os.environ.items() if k != "ANU_BOT_TOKEN"} with patch.dict("os.environ", env, clear=True): result = notify_completion.send_telegram_notification("chat123", "메시지") assert result is False def test_non_200_returns_false(self): """200이 아닌 응답(400 등) 시 False 반환""" mock_resp = MagicMock() mock_resp.status_code = 400 mock_resp.headers = {} with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", return_value=mock_resp): result = notify_completion.send_telegram_notification("chat123", "메시지") assert result is False def test_429_retries_with_retry_after(self): """429 응답 시 Retry-After 헤더를 따라 재시도""" import requests as real_requests mock_429 = MagicMock() mock_429.status_code = 429 mock_429.headers = {"Retry-After": "1"} mock_200 = MagicMock() mock_200.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", side_effect=[mock_429, mock_200]) as mock_post: with patch("time.sleep"): result = notify_completion.send_telegram_notification("chat123", "메시지") assert result is True assert mock_post.call_count == 2 def test_request_exception_retries(self): """네트워크 에러 시 재시도 후 False 반환""" import requests as real_requests with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", side_effect=real_requests.RequestException("connection error")): with patch("time.sleep"): result = notify_completion.send_telegram_notification("chat123", "메시지") assert result is False def test_uses_correct_api_url(self): """올바른 Telegram API URL 사용""" mock_resp = MagicMock() mock_resp.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "mytoken123"}): with patch("requests.post", return_value=mock_resp) as mock_post: notify_completion.send_telegram_notification("chat123", "메시지") call_args = mock_post.call_args url = call_args[0][0] assert "mytoken123" in url assert "api.telegram.org" in url assert "sendMessage" in url def test_timeout_is_10_seconds(self): """timeout=10 필수""" mock_resp = MagicMock() mock_resp.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", return_value=mock_resp) as mock_post: notify_completion.send_telegram_notification("chat123", "메시지") call_kwargs = mock_post.call_args[1] assert call_kwargs.get("timeout") == 10 def test_parse_mode_markdown_in_payload(self): """requests.post payload에 parse_mode: 'Markdown'이 포함되어야 한다.""" mock_resp = MagicMock() mock_resp.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", return_value=mock_resp) as mock_post: notify_completion.send_telegram_notification("chat123", "테스트 메시지") call_kwargs = mock_post.call_args[1] payload = call_kwargs.get("json", {}) assert payload.get("parse_mode") == "Markdown" def test_markdown_parse_error_fallback(self): """Markdown 파싱 에러(400) 시 plain text fallback으로 2회 호출되어야 한다. 첫 번째 호출에는 parse_mode가 있고 두 번째에는 없어야 한다.""" mock_400 = MagicMock() mock_400.status_code = 400 mock_400.headers = {} mock_200 = MagicMock() mock_200.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", side_effect=[mock_400, mock_200]) as mock_post: notify_completion.send_telegram_notification("chat123", "테스트 메시지") assert mock_post.call_count == 2 first_call_kwargs = mock_post.call_args_list[0][1] first_payload = first_call_kwargs.get("json", {}) assert first_payload.get("parse_mode") == "Markdown" second_call_kwargs = mock_post.call_args_list[1][1] second_payload = second_call_kwargs.get("json", {}) assert "parse_mode" not in second_payload def test_fallback_returns_true_on_success(self): """첫 번째 Markdown 전송이 400일 때 fallback 성공 시 True를 반환해야 한다. (parse_mode=Markdown → 400 → parse_mode 없는 재전송 → 200 → True)""" mock_400 = MagicMock() mock_400.status_code = 400 mock_400.headers = {} mock_200 = MagicMock() mock_200.status_code = 200 with patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token"}): with patch("requests.post", side_effect=[mock_400, mock_200]) as mock_post: result = notify_completion.send_telegram_notification("chat123", "테스트 메시지") # 첫 번째 호출에 parse_mode=Markdown이 있어야 fallback이 의미를 가진다 first_payload = mock_post.call_args_list[0][1].get("json", {}) assert first_payload.get("parse_mode") == "Markdown" assert result is True class TestMainWithMock: """main() 함수 테스트 (requests.post를 mock)""" def _make_subprocess_side_effect(self, chain_result=None): """subprocess.run side_effect 공통 생성""" if chain_result is None: chain_result = {"in_chain": False, "is_last": False, "chain_id": None, "next_task_id": None} def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps(chain_result) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result return side_effect @patch("subprocess.run") def test_main_calls_requests_post(self, mock_run, tmp_path): """main()이 send_telegram_notification(requests.post)를 호출하는지 확인""" mock_run.side_effect = self._make_subprocess_side_effect() (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch("sys.argv", ["notify-completion.py", "task-999.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp) as mock_post, ): notify_completion.main() assert mock_post.call_count >= 1 call_args = mock_post.call_args # payload에 task_id와 완료 메시지가 있는지 확인 payload = call_args[1].get("json") or call_args[0][1] if len(call_args[0]) > 1 else call_args[1].get("json") if payload is None and call_args[1]: payload = call_args[1].get("json") text = payload.get("text", "") if payload else "" assert "task-999.1" in text, f"메시지에 task_id가 없음: {text}" assert "완료" in text, f"메시지에 '완료'가 없음: {text}" @patch("subprocess.run") def test_main_custom_chat_id(self, mock_run, tmp_path): """--chat-id 커스텀 값 전달 시 requests.post payload에 반영""" mock_run.side_effect = self._make_subprocess_side_effect() (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch("sys.argv", ["notify-completion.py", "task-1.1", "--chat-id", "12345"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp) as mock_post, ): notify_completion.main() assert mock_post.call_count >= 1 call_kwargs = mock_post.call_args[1] payload = call_kwargs.get("json", {}) assert payload.get("chat_id") == "12345", f"chat_id가 12345가 아님: {payload}" @patch("subprocess.run") def test_main_no_cokacdir_called(self, mock_run, tmp_path): """main() 실행 시 cokacdir 명령이 호출되지 않아야 함""" mock_run.side_effect = self._make_subprocess_side_effect() (tmp_path / "memory" / "events").mkdir(parents=True, exist_ok=True) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch("sys.argv", ["notify-completion.py", "task-999.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp), ): notify_completion.main() # cokacdir 호출이 없어야 함 cokacdir_calls = [c for c in mock_run.call_args_list if isinstance(c[0][0], list) and c[0][0][0] == "cokacdir"] assert len(cokacdir_calls) == 0, f"cokacdir 호출이 남아있음: {cokacdir_calls}" class TestDoneNotifiedMarker: """.done.notified 마커 파일 생성 테스트""" @patch("subprocess.run") def test_done_notified_marker_created_on_success(self, mock_run, tmp_path): """알림 성공 시 .done.notified 마커 파일 생성""" def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( {"in_chain": False, "is_last": False, "chain_id": None, "next_task_id": None} ) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result mock_run.side_effect = side_effect events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch("sys.argv", ["notify-completion.py", "task-777.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp), ): notify_completion.main() notified_path = events_dir / "task-777.1.done.notified" assert notified_path.exists(), f".done.notified 마커가 생성되지 않음: {notified_path}" @patch("subprocess.run") def test_done_notified_marker_not_created_on_failure(self, mock_run, tmp_path): """알림 실패 시 .done.notified 마커 파일 생성 안 함""" def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( {"in_chain": False, "is_last": False, "chain_id": None, "next_task_id": None} ) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result mock_run.side_effect = side_effect events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) # ANU_BOT_TOKEN 없으면 send_telegram_notification이 False 반환 env = {k: v for k, v in os.environ.items() if k != "ANU_BOT_TOKEN"} env["COKACDIR_KEY_ANU"] = "test-anu-key" # anu_key는 유지 with ( patch("sys.argv", ["notify-completion.py", "task-888.1"]), patch.dict("os.environ", env, clear=True), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), ): notify_completion.main() notified_path = events_dir / "task-888.1.done.notified" assert not notified_path.exists(), f".done.notified 마커가 불필요하게 생성됨: {notified_path}" @patch("subprocess.run") def test_done_notified_marker_oexcl_no_duplicate(self, mock_run, tmp_path): """이미 마커 존재 시 중복 생성 시도해도 에러 없이 처리""" def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( {"in_chain": False, "is_last": False, "chain_id": None, "next_task_id": None} ) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result mock_run.side_effect = side_effect events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) # 미리 마커 파일 생성 (이미 존재 상태) notified_path = events_dir / "task-555.1.done.notified" notified_path.touch() mock_resp = MagicMock() mock_resp.status_code = 200 # 예외 없이 실행되어야 함 with ( patch("sys.argv", ["notify-completion.py", "task-555.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp), ): notify_completion.main() # FileExistsError 예외 없이 통과해야 함 class TestDispatchNextPhaseWithTaskId: """dispatch_next_phase 결과에서 task_id를 받아 dispatch 명령에 --task-id를 추가하는 테스트""" def test_dispatch_next_phase_result_with_task_id(self): """dispatch_next_phase 결과에 task_id가 있으면 dispatch 명령에 --task-id 인자가 추가된다.""" dispatch_result = { "action": "dispatch", "task_file": "memory/tasks/task-566.2.md", "team": "dev1-team", "chain_id": "scoped-566", "task_id": "task-566.2", } task_file = dispatch_result.get("task_file") team = dispatch_result.get("team") level = dispatch_result.get("level", "normal") next_task_id = dispatch_result.get("task_id") workspace_root = _WORKSPACE task_id_arg = f" --task-id {next_task_id}" if next_task_id else "" dispatch_cmd = ( f"source {workspace_root}/.env.keys && python3 {workspace_root}/dispatch.py " f"--team {team} --task-file {task_file} --level {level}{task_id_arg}" ) assert "--task-id" in dispatch_cmd, f"dispatch 명령에 --task-id가 없음: {dispatch_cmd}" assert "task-566.2" in dispatch_cmd, f"dispatch 명령에 task-566.2가 없음: {dispatch_cmd}" def test_dispatch_next_phase_result_without_task_id(self): """dispatch_next_phase 결과에 task_id가 없으면 dispatch 명령에 --task-id가 없다.""" dispatch_result = { "action": "dispatch", "task_file": "memory/tasks/task-567.2.md", "team": "dev1-team", "chain_id": "scoped-567", # task_id 없음 (구버전 호환) } task_file = dispatch_result.get("task_file") team = dispatch_result.get("team") level = dispatch_result.get("level", "normal") next_task_id = dispatch_result.get("task_id") workspace_root = _WORKSPACE task_id_arg = f" --task-id {next_task_id}" if next_task_id else "" dispatch_cmd = ( f"source {workspace_root}/.env.keys && python3 {workspace_root}/dispatch.py " f"--team {team} --task-file {task_file} --level {level}{task_id_arg}" ) assert "--task-id" not in dispatch_cmd, f"task_id 없는데 --task-id가 있음: {dispatch_cmd}" def test_notify_completion_dispatch_cmd_format(self, tmp_path): """실제 notify_completion의 dispatch 호출이 --task-id를 포함하는지 확인 (shell=False 방식).""" import subprocess as real_subprocess # chain_manager.py next → dispatch action with task_id def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( {"in_chain": True, "is_last": False, "chain_id": "scoped-566", "next_task_id": "task-566.2"} ) elif "chain_manager.py" in cmd_str and "next" in cmd_str: result.stdout = json.dumps( { "action": "dispatch", "task_file": "memory/tasks/task-566.2.md", "team": "dev1-team", "chain_id": "scoped-566", "task_id": "task-566.2", } ) elif "dispatch.py" in cmd_str: result.stdout = json.dumps({"status": "dispatched", "task_id": "task-566.2"}) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) result.stderr = "" return result # events 디렉토리 생성 events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) dispatch_calls: list = [] def capturing_side_effect(cmd, *args, **kwargs): if isinstance(cmd, list) and "dispatch.py" in " ".join(str(c) for c in cmd): dispatch_calls.append(cmd) return side_effect(cmd, *args, **kwargs) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch.object(notify_completion, "subprocess") as mock_sub, patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("sys.argv", ["notify-completion.py", "task-568.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch("requests.post", return_value=mock_resp), ): mock_sub.run.side_effect = capturing_side_effect mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired notify_completion.main() assert len(dispatch_calls) >= 1, "dispatch.py 호출이 없음" dispatch_cmd_full = " ".join(str(x) for x in dispatch_calls[0]) assert "--task-id" in dispatch_cmd_full, f"dispatch 명령에 --task-id가 없음: {dispatch_cmd_full}" assert "task-566.2" in dispatch_cmd_full, f"dispatch 명령에 task-566.2가 없음: {dispatch_cmd_full}" class TestShellInjectionPrevention: """셸 인젝션 방지 테스트 (Fenrir P0 위협 1-B 대응)""" def test_dispatch_uses_list_not_bash_c(self, tmp_path): """dispatch 호출이 ['bash', '-c', ...] 대신 ['python3', ...] 리스트 방식인지 확인""" import subprocess as real_subprocess def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( { "in_chain": True, "is_last": False, "chain_id": "scoped-100", "next_task_id": "task-100.2", } ) elif "chain_manager.py" in cmd_str and "next" in cmd_str: result.stdout = json.dumps( { "action": "dispatch", "task_file": "memory/tasks/task-100.2.md", "team": "dev1-team", "level": "normal", "task_id": "task-100.2", } ) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result all_calls: list = [] def capturing_side_effect(cmd, *args, **kwargs): all_calls.append(cmd) return side_effect(cmd, *args, **kwargs) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch.object(notify_completion, "subprocess") as mock_sub, patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("sys.argv", ["notify-completion.py", "task-100.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch("requests.post", return_value=mock_resp), ): mock_sub.run.side_effect = capturing_side_effect mock_sub.TimeoutExpired = real_subprocess.TimeoutExpired notify_completion.main() # ["bash", "-c", ...] 형태 호출이 없어야 함 bash_c_calls = [c for c in all_calls if isinstance(c, list) and len(c) >= 2 and c[0] == "bash" and c[1] == "-c"] assert len(bash_c_calls) == 0, f"bash -c 호출이 남아있음 (셸 인젝션 위험): {bash_c_calls}" # ["python3", ..., "dispatch.py", ...] 형태 호출이 있어야 함 python3_dispatch_calls = [ c for c in all_calls if isinstance(c, list) and len(c) >= 2 and c[0] == "python3" and any("dispatch.py" in str(x) for x in c) ] assert len(python3_dispatch_calls) >= 1, "python3 dispatch.py 리스트 호출이 없음" def test_validate_dispatch_args_rejects_shell_metachar(self): """team에 셸 메타문자(;rm -rf /)가 있으면 ValueError 발생""" with pytest.raises(ValueError): notify_completion._validate_dispatch_args( team=";rm -rf /", task_file="memory/tasks/task-1.1.md", level="normal", next_task_id=None, ) def test_validate_dispatch_args_rejects_task_file_metachar(self): """task_file에 셸 메타문자가 있으면 ValueError 발생""" with pytest.raises(ValueError): notify_completion._validate_dispatch_args( team="dev1-team", task_file="memory/tasks/task-1.1.md; echo pwned", level="normal", next_task_id=None, ) def test_validate_dispatch_args_rejects_invalid_level(self): """level이 허용값(normal/critical/security) 외이면 ValueError 발생""" with pytest.raises(ValueError): notify_completion._validate_dispatch_args( team="dev1-team", task_file="memory/tasks/task-1.1.md", level="HACKED", next_task_id=None, ) def test_validate_dispatch_args_rejects_invalid_next_task_id(self): """next_task_id가 task-숫자.숫자 형식이 아니면 ValueError 발생""" with pytest.raises(ValueError): notify_completion._validate_dispatch_args( team="dev1-team", task_file="memory/tasks/task-1.1.md", level="normal", next_task_id="task-1.1; drop table tasks", ) def test_validate_dispatch_args_accepts_valid(self): """정상적인 인자값은 예외 없이 통과""" notify_completion._validate_dispatch_args( team="dev1-team", task_file="memory/tasks/task-566.2.md", level="normal", next_task_id="task-566.2", ) def test_validate_dispatch_args_accepts_valid_no_task_id(self): """next_task_id=None은 정상 통과""" notify_completion._validate_dispatch_args( team="dev1-team", task_file="memory/tasks/task-566.2.md", level="critical", next_task_id=None, ) def test_validate_dispatch_args_accepts_security_level(self): """level=security도 정상 통과""" notify_completion._validate_dispatch_args( team="alpha-team", task_file=f"{_WORKSPACE}/memory/tasks/task-1.1.md", level="security", next_task_id="task-1.2", ) def test_load_env_keys_parses_exports(self, tmp_path): """export KEY=VALUE 형식 파싱""" env_file = tmp_path / ".env.keys" env_file.write_text( "export API_KEY=abc123\n" 'export DB_PASS="secret"\n' "export TOKEN='mytoken'\n" "# 이것은 주석\n" "\n" "PLAIN_VAR=plainvalue\n", encoding="utf-8", ) result = notify_completion.load_env_keys(str(env_file)) assert result.get("API_KEY") == "abc123" assert result.get("DB_PASS") == "secret" assert result.get("TOKEN") == "mytoken" assert result.get("PLAIN_VAR") == "plainvalue" def test_load_env_keys_ignores_comments_and_blanks(self, tmp_path): """주석(#)과 빈 줄은 무시""" env_file = tmp_path / ".env.keys" env_file.write_text( "# 주석\n" "\n" "export VALID_KEY=validvalue\n", encoding="utf-8", ) result = notify_completion.load_env_keys(str(env_file)) assert "VALID_KEY" in result assert len(result) == 1 def test_load_env_keys_missing_file(self, tmp_path): """파일이 없으면 빈 딕셔너리 반환""" result = notify_completion.load_env_keys(str(tmp_path / "nonexistent.env")) assert result == {} class TestSaveCompletionMessage: """_save_completion_message 함수 테스트 (task-923.1)""" def test_saves_message_to_file(self, tmp_path): """메시지를 completion.txt 파일로 저장""" with patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)): events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) notify_completion._save_completion_message("task-100.1", "테스트 완료 메시지") saved = (tmp_path / "memory" / "events" / "task-100.1.completion.txt").read_text(encoding="utf-8") assert saved == "테스트 완료 메시지" def test_creates_parent_directory(self, tmp_path): """events 디렉토리가 없으면 자동 생성""" with patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)): notify_completion._save_completion_message("task-200.1", "메시지") assert (tmp_path / "memory" / "events" / "task-200.1.completion.txt").exists() def test_overwrites_existing_file(self, tmp_path): """기존 파일이 있으면 덮어쓰기""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) (events_dir / "task-300.1.completion.txt").write_text("이전 메시지", encoding="utf-8") with patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)): notify_completion._save_completion_message("task-300.1", "새 메시지") saved = (events_dir / "task-300.1.completion.txt").read_text(encoding="utf-8") assert saved == "새 메시지" def test_handles_oserror_gracefully(self, tmp_path): """OSError 발생 시 예외 없이 처리""" with patch.object(notify_completion, "WORKSPACE_ROOT", "/nonexistent/readonly/path"): with patch.object(Path, "mkdir", side_effect=OSError("Permission denied")): # 예외 없이 실행되어야 함 notify_completion._save_completion_message("task-400.1", "메시지") @patch("subprocess.run") def test_completion_file_created_in_main_normal_path(self, mock_run, tmp_path): """main() 일반 경로에서 .completion.txt가 생성되는지 확인""" def side_effect(cmd, *args, **kwargs): result = MagicMock() result.returncode = 0 result.stderr = "" if isinstance(cmd, list): cmd_str = " ".join(str(c) for c in cmd) if "chain_manager.py" in cmd_str and "check" in cmd_str: result.stdout = json.dumps( {"in_chain": False, "is_last": False, "chain_id": None, "next_task_id": None} ) else: result.stdout = json.dumps({"status": "ok"}) else: result.stdout = json.dumps({"status": "ok"}) return result mock_run.side_effect = side_effect events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True, exist_ok=True) mock_resp = MagicMock() mock_resp.status_code = 200 with ( patch("sys.argv", ["notify-completion.py", "task-923.1"]), patch.dict("os.environ", {"ANU_BOT_TOKEN": "test-token", "COKACDIR_KEY_ANU": "test-anu-key"}), patch.object(notify_completion, "WORKSPACE_ROOT", str(tmp_path)), patch("requests.post", return_value=mock_resp), ): notify_completion.main() completion_file = events_dir / "task-923.1.completion.txt" assert completion_file.exists(), f".completion.txt 파일이 생성되지 않음" content = completion_file.read_text(encoding="utf-8") assert "task-923.1" in content assert "완료" in content