""" test_auto_merge.py auto_merge.py TDD 테스트 (토르 작성) 테스트 항목: 1. test_scan_done_files: .done 파일 감지, .done.clear 제외 2. test_try_claim_success: 원자적 선점 성공 3. test_try_claim_already_claimed: 이미 선점된 경우 False 4. test_parse_done_valid: 정상 JSON 파싱 5. test_parse_done_invalid: 비정상 JSON 처리 6. test_parse_done_legacy_format: 구형 포맷 (merge_needed 없음) 처리 7. test_analyze_report: mock으로 parse_report 결과 검증 8. test_resolve_project_path_from_report: 보고서에서 프로젝트 경로 추출 9. test_execute_merge_success: 머지 성공 시나리오 (subprocess mock) 10. test_execute_merge_conflict: 머지 충돌 시나리오 11. test_run_tests_pytest: pytest 실행 성공 12. test_run_tests_timeout: 타임아웃 처리 13. test_revert_merge: 되돌리기 테스트 14. test_log_result: merge-log.json 기록 15. test_dry_run: 드라이런 모드에서 실제 머지 없음 16. test_run_full_flow: 전체 흐름 통합 테스트 (mock 기반) 17. test_load_env_keys: .env.keys 파싱 """ import json import os import sys from pathlib import Path from unittest.mock import MagicMock, patch import pytest # workspace를 sys.path에 추가 WORKSPACE = Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")) if str(WORKSPACE) not in sys.path: sys.path.insert(0, str(WORKSPACE)) # scripts 경로 추가 SCRIPTS = WORKSPACE / "scripts" if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) from auto_merge import AutoMerger # pyright: ignore[reportMissingImports] # --------------------------------------------------------------------------- # 1. test_scan_done_files # --------------------------------------------------------------------------- def test_scan_done_files(tmp_path: Path): """미처리 .done 파일 스캔: .done.clear 없는 것만 반환한다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) # 처리 대상: .done만 있음 f1 = events_dir / "task-100.1.done" f1.write_text('{"task_id": "task-100.1", "status": "done"}', encoding="utf-8") # 이미 처리됨: .done.clear도 있음 f2 = events_dir / "task-101.1.done" f2.write_text('{"task_id": "task-101.1", "status": "done"}', encoding="utf-8") (events_dir / "task-101.1.done.clear").write_text("{}", encoding="utf-8") # .done이 없고 .done.clear만 있는 경우 (이미 삭제된 케이스) (events_dir / "task-102.1.done.clear").write_text("{}", encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) result = merger.scan_done_files() assert len(result) == 1 assert result[0] == f1 # --------------------------------------------------------------------------- # 2. test_try_claim_success # --------------------------------------------------------------------------- def test_try_claim_success(tmp_path: Path): """원자적 선점: .done.merging 파일을 생성하고 True를 반환한다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) done_file = events_dir / "task-200.1.done" done_file.write_text('{"task_id": "task-200.1"}', encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) result = merger.try_claim(done_file) assert result is True # .done.merging 파일 생성 확인 (.done.clear가 아님) assert (events_dir / "task-200.1.done.merging").exists() # .done.clear는 생성되지 않아야 함 (아누 프로토콜 보호) assert not (events_dir / "task-200.1.done.clear").exists() # --------------------------------------------------------------------------- # 3. test_try_claim_already_claimed # --------------------------------------------------------------------------- def test_try_claim_already_claimed(tmp_path: Path): """원자적 선점: .done.merging 파일이 이미 존재하면 False를 반환한다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) done_file = events_dir / "task-201.1.done" done_file.write_text('{"task_id": "task-201.1"}', encoding="utf-8") # 이미 선점됨 (.done.merging으로 선점 표시) (events_dir / "task-201.1.done.merging").write_text("{}", encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) result = merger.try_claim(done_file) assert result is False # --------------------------------------------------------------------------- # 4. test_parse_done_valid # --------------------------------------------------------------------------- def test_parse_done_valid(tmp_path: Path): """정상 JSON .done 파일 파싱: task_id, team_id, merge_needed 등 추출.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) done_file = events_dir / "task-391.1.done" done_data = { "task_id": "task-391.1", "status": "done", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "team_id": "dev1", } done_file.write_text(json.dumps(done_data), encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) result = merger.parse_done(done_file) assert result["task_id"] == "task-391.1" assert result["team_id"] == "dev1" assert result["merge_needed"] is True assert result["merge_branch"] == "task/task-391.1-dev1" # --------------------------------------------------------------------------- # 5. test_parse_done_invalid # --------------------------------------------------------------------------- def test_parse_done_invalid(tmp_path: Path): """비정상 JSON: escalate를 호출하고 예외를 발생시킨다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) done_file = events_dir / "task-bad.done" done_file.write_text("NOT VALID JSON{{{{", encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) with patch.object(merger, "escalate") as mock_escalate: with pytest.raises(ValueError, match="JSON"): merger.parse_done(done_file) mock_escalate.assert_called_once() # --------------------------------------------------------------------------- # 6. test_parse_done_legacy_format # --------------------------------------------------------------------------- def test_parse_done_legacy_format(tmp_path: Path): """구형 포맷: merge_needed 없는 파일도 올바르게 파싱한다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) done_file = events_dir / "task-old.done" legacy_data = { "team_id": "dev2", "end_time": "2026-03-01T10:00:00", "duration_seconds": 120.0, } done_file.write_text(json.dumps(legacy_data), encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) result = merger.parse_done(done_file) # merge_needed 없으면 기본값 False assert result.get("merge_needed", False) is False assert result["team_id"] == "dev2" # task_id는 파일명에서 추출 assert result["task_id"] == "task-old" # --------------------------------------------------------------------------- # 7. test_analyze_report # --------------------------------------------------------------------------- def test_analyze_report(tmp_path: Path): """analyze_report: parse_report를 mock으로 대체하여 결과 검증.""" reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) # 더미 보고서 파일 생성 report_file = reports_dir / "task-391.1.md" report_file.write_text("# task-391.1 보고서\n", encoding="utf-8") mock_report = { "task_id": "task-391.1", "team": "dev1", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "merge_worktree": "/home/jay/projects/MyApp/.worktrees/task-391.1-dev1", "files": [], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): result = merger.analyze_report("task-391.1") assert result["merge_needed"] is True assert result["merge_branch"] == "task/task-391.1-dev1" assert result["merge_worktree"] == "/home/jay/projects/MyApp/.worktrees/task-391.1-dev1" # --------------------------------------------------------------------------- # 8. test_resolve_project_path_from_report # --------------------------------------------------------------------------- def test_resolve_project_path_from_report(tmp_path: Path): """보고서 데이터에서 /home/jay/projects/... 패턴 경로를 추출한다.""" merger = AutoMerger(workspace_path=str(tmp_path)) report_data = { "merge_worktree": "/home/jay/projects/ThreadAuto/.worktrees/task-391.1-dev1", "files": ["/home/jay/projects/ThreadAuto/main.py"], } result = merger.resolve_project_path("task-391.1", report_data) assert result == "/home/jay/projects/ThreadAuto" def test_resolve_project_path_from_worktree(tmp_path: Path): """merge_worktree에서 프로젝트 경로를 역추적한다.""" merger = AutoMerger(workspace_path=str(tmp_path)) report_data = { "merge_worktree": "/home/jay/projects/MyProject/.worktrees/task-391.1-dev1", "files": [], } result = merger.resolve_project_path("task-391.1", report_data) assert result == "/home/jay/projects/MyProject" def test_resolve_project_path_none(tmp_path: Path): """경로를 찾을 수 없을 때 None을 반환한다.""" merger = AutoMerger(workspace_path=str(tmp_path)) report_data = { "merge_worktree": None, "files": [], } result = merger.resolve_project_path("task-999.9", report_data) assert result is None # --------------------------------------------------------------------------- # 9. test_execute_merge_success # --------------------------------------------------------------------------- def test_execute_merge_success(tmp_path: Path): """머지 성공: worktree_manager.py subprocess 호출 결과 파싱.""" merger = AutoMerger(workspace_path=str(tmp_path)) mock_output = json.dumps({"status": "merged", "branch": "task/task-391.1-dev1"}) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = mock_output mock_result.stderr = "" with patch("subprocess.run", return_value=mock_result): result = merger.execute_merge("/home/jay/projects/MyApp", "task-391.1", "dev1") assert result["status"] == "merged" assert result["branch"] == "task/task-391.1-dev1" # --------------------------------------------------------------------------- # 10. test_execute_merge_conflict # --------------------------------------------------------------------------- def test_execute_merge_conflict(tmp_path: Path): """머지 충돌: subprocess 오류 시 RuntimeError를 발생시킨다.""" merger = AutoMerger(workspace_path=str(tmp_path)) mock_error_output = json.dumps( {"status": "error", "message": "Merge conflict detected while merging 'task/task-391.1-dev1'"} ) mock_result = MagicMock() mock_result.returncode = 1 mock_result.stdout = mock_error_output mock_result.stderr = "conflict" with patch("subprocess.run", return_value=mock_result): with pytest.raises(RuntimeError, match="conflict|error"): merger.execute_merge("/home/jay/projects/MyApp", "task-391.1", "dev1") # --------------------------------------------------------------------------- # 11. test_run_tests_pytest # --------------------------------------------------------------------------- def test_run_tests_pytest(tmp_path: Path): """pytest 실행 성공: passed=True, output 포함.""" merger = AutoMerger(workspace_path=str(tmp_path)) # pytest가 있는 프로젝트 디렉토리 시뮬레이션 project_dir = tmp_path / "myproject" project_dir.mkdir() mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "5 passed in 1.23s" mock_result.stderr = "" with patch("subprocess.run", return_value=mock_result): with patch("shutil.which", return_value="/usr/bin/pytest"): result = merger.run_tests(str(project_dir)) assert result["passed"] is True assert "passed" in result["output"] assert isinstance(result["duration"], float) # --------------------------------------------------------------------------- # 12. test_run_tests_timeout # --------------------------------------------------------------------------- def test_run_tests_timeout(tmp_path: Path): """테스트 타임아웃: passed=False, output에 timeout 메시지.""" import subprocess merger = AutoMerger(workspace_path=str(tmp_path)) project_dir = tmp_path / "myproject" project_dir.mkdir() with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("pytest", 300)): with patch("shutil.which", return_value="/usr/bin/pytest"): result = merger.run_tests(str(project_dir)) assert result["passed"] is False assert "timeout" in result["output"].lower() # --------------------------------------------------------------------------- # 13. test_revert_merge # --------------------------------------------------------------------------- def test_revert_merge(tmp_path: Path): """머지 되돌리기: git reset --hard HEAD~1 호출 확인.""" merger = AutoMerger(workspace_path=str(tmp_path)) mock_result = MagicMock() mock_result.returncode = 0 mock_result.stdout = "HEAD is now at abc1234" mock_result.stderr = "" with patch("subprocess.run", return_value=mock_result) as mock_run: result = merger.revert_merge("/home/jay/projects/MyApp") assert result is True # git reset --hard HEAD~1 호출 확인 called_args = mock_run.call_args[0][0] assert "git" in called_args assert "reset" in called_args assert "--hard" in called_args # --------------------------------------------------------------------------- # 14. test_log_result # --------------------------------------------------------------------------- def test_log_result(tmp_path: Path): """merge-log.json에 결과를 추가한다. 기존 entries는 유지된다.""" memory_dir = tmp_path / "memory" memory_dir.mkdir() merge_log = memory_dir / "merge-log.json" # 기존 로그 existing = {"entries": [{"task_id": "task-100.1", "status": "success"}]} merge_log.write_text(json.dumps(existing), encoding="utf-8") merger = AutoMerger(workspace_path=str(tmp_path)) new_entry = { "task_id": "task-391.1", "timestamp": "2026-03-07T15:30:00+09:00", "merge_needed": True, "action": "auto_merged", "project": "/home/jay/projects/ThreadAuto", "branch": "task/task-391.1-dev1", "test_result": "pass", "duration_seconds": 12.5, "status": "success", } merger.log_result("task-391.1", new_entry) saved = json.loads(merge_log.read_text(encoding="utf-8")) assert len(saved["entries"]) == 2 assert saved["entries"][0]["task_id"] == "task-100.1" assert saved["entries"][1]["task_id"] == "task-391.1" def test_log_result_creates_file(tmp_path: Path): """merge-log.json이 없으면 새로 생성한다.""" memory_dir = tmp_path / "memory" memory_dir.mkdir() merger = AutoMerger(workspace_path=str(tmp_path)) new_entry = {"task_id": "task-500.1", "status": "success"} merger.log_result("task-500.1", new_entry) merge_log = memory_dir / "merge-log.json" assert merge_log.exists() saved = json.loads(merge_log.read_text(encoding="utf-8")) assert len(saved["entries"]) == 1 assert saved["entries"][0]["task_id"] == "task-500.1" # --------------------------------------------------------------------------- # 15. test_dry_run # --------------------------------------------------------------------------- def test_dry_run(tmp_path: Path): """드라이런 모드: 실제 머지 없이 분석만 수행한다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) # .done 파일 생성 done_file = events_dir / "task-391.1.done" done_data = { "task_id": "task-391.1", "status": "done", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "team_id": "dev1", } done_file.write_text(json.dumps(done_data), encoding="utf-8") # 보고서 파일 생성 report_file = reports_dir / "task-391.1.md" report_file.write_text("# task-391.1\n", encoding="utf-8") mock_report = { "task_id": "task-391.1", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "merge_worktree": "/home/jay/projects/TestApp/.worktrees/task-391.1-dev1", "files": ["/home/jay/projects/TestApp/main.py"], } merger = AutoMerger(workspace_path=str(tmp_path), dry_run=True) with patch("auto_merge.parse_report", return_value=mock_report): with patch.object(merger, "execute_merge") as mock_merge: with patch.object(merger, "run_tests") as mock_tests: result = merger.run() # 드라이런이므로 실제 머지 호출 없음 mock_merge.assert_not_called() mock_tests.assert_not_called() # processed 카운트는 있어야 함 assert result["processed"] >= 0 # --------------------------------------------------------------------------- # 16. test_run_full_flow # --------------------------------------------------------------------------- def test_run_full_flow(tmp_path: Path): """전체 흐름 통합 테스트: scan → claim → parse → analyze → merge → test → log.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) # .done 파일 생성 done_file = events_dir / "task-391.1.done" done_data = { "task_id": "task-391.1", "status": "done", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "team_id": "dev1", } done_file.write_text(json.dumps(done_data), encoding="utf-8") # 보고서 파일 생성 report_file = reports_dir / "task-391.1.md" report_file.write_text("# task-391.1\n", encoding="utf-8") mock_report = { "task_id": "task-391.1", "merge_needed": True, "merge_branch": "task/task-391.1-dev1", "merge_worktree": "/home/jay/projects/TestApp/.worktrees/task-391.1-dev1", "files": ["/home/jay/projects/TestApp/main.py"], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): with patch.object(merger, "execute_merge", return_value={"status": "merged", "branch": "task/task-391.1-dev1"}): with patch.object( merger, "run_tests", return_value={"passed": True, "output": "5 passed", "duration": 1.5}, ): result = merger.run() assert result["processed"] == 1 assert result["merged"] == 1 assert result["escalated"] == 0 # .done 파일은 삭제되지 않고 유지되어야 함 (아누가 보고 후 .done.clear로 변환) assert done_file.exists(), ".done 파일은 아누가 처리할 때까지 유지되어야 함" # merge-log.json 기록 확인 merge_log = tmp_path / "memory" / "merge-log.json" assert merge_log.exists() log_data = json.loads(merge_log.read_text(encoding="utf-8")) assert len(log_data["entries"]) == 1 assert log_data["entries"][0]["task_id"] == "task-391.1" def test_run_full_flow_test_failure(tmp_path: Path): """테스트 실패 시 머지 revert + escalate 호출 확인.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) done_file = events_dir / "task-392.1.done" done_data = { "task_id": "task-392.1", "status": "done", "merge_needed": True, "team_id": "dev2", } done_file.write_text(json.dumps(done_data), encoding="utf-8") report_file = reports_dir / "task-392.1.md" report_file.write_text("# task-392.1\n", encoding="utf-8") mock_report = { "task_id": "task-392.1", "merge_needed": True, "merge_branch": "task/task-392.1-dev2", "merge_worktree": "/home/jay/projects/TestApp/.worktrees/task-392.1-dev2", "files": ["/home/jay/projects/TestApp/main.py"], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): with patch.object(merger, "execute_merge", return_value={"status": "merged", "branch": "task/task-392.1-dev2"}): with patch.object( merger, "run_tests", return_value={"passed": False, "output": "FAILED 2 errors", "duration": 2.0}, ): with patch.object(merger, "revert_merge", return_value=True) as mock_revert: with patch.object(merger, "escalate") as mock_escalate: result = merger.run() mock_revert.assert_called_once() mock_escalate.assert_called_once() assert result["escalated"] == 1 # --------------------------------------------------------------------------- # 17. test_load_env_keys # --------------------------------------------------------------------------- def test_load_env_keys(tmp_path: Path): """.env.keys 파일을 파싱하여 dict로 반환한다.""" env_file = tmp_path / ".env.keys" env_file.write_text( "export COKACDIR_KEY_ANU=abc123def456\n" "export COKACDIR_KEY_DEV1=xyz789\n" "# 주석\n" "export WORKSPACE_ROOT=/home/jay/workspace\n", encoding="utf-8", ) merger = AutoMerger(workspace_path=str(tmp_path)) result = merger._load_env_keys() assert result["COKACDIR_KEY_ANU"] == "abc123def456" assert result["COKACDIR_KEY_DEV1"] == "xyz789" assert result["WORKSPACE_ROOT"] == "/home/jay/workspace" # 주석은 포함되지 않음 assert "#" not in str(result.keys()) # --------------------------------------------------------------------------- # 18. test_team_id_to_short (문제 2: team_id 매핑 버그) # --------------------------------------------------------------------------- def test_team_id_to_short_with_suffix(): """'dev1-team' → 'dev1' 변환.""" assert AutoMerger._team_id_to_short("dev1-team") == "dev1" assert AutoMerger._team_id_to_short("dev2-team") == "dev2" def test_team_id_to_short_already_short(): """이미 short 형식이면 그대로 반환.""" assert AutoMerger._team_id_to_short("dev1") == "dev1" assert AutoMerger._team_id_to_short("dev2") == "dev2" def test_team_id_to_short_edge_cases(): """빈 문자열, 'team'만 있는 경우 등 엣지 케이스.""" assert AutoMerger._team_id_to_short("") == "" assert AutoMerger._team_id_to_short("team") == "team" assert AutoMerger._team_id_to_short("custom-team") == "custom" # --------------------------------------------------------------------------- # 19. test_extract_team_short_from_report (문제 2) # --------------------------------------------------------------------------- def test_extract_team_short_from_branch(): """merge_branch에서 team_short 역추출.""" report_data = {"merge_branch": "task/task-391.1-dev1", "merge_worktree": None} assert AutoMerger._extract_team_short_from_report(report_data) == "dev1" def test_extract_team_short_from_worktree(): """merge_worktree에서 team_short 역추출.""" report_data = { "merge_branch": None, "merge_worktree": "/home/jay/projects/App/.worktrees/task-391.1-dev2", } assert AutoMerger._extract_team_short_from_report(report_data) == "dev2" def test_extract_team_short_none(): """둘 다 없으면 None 반환.""" report_data = {"merge_branch": None, "merge_worktree": None} assert AutoMerger._extract_team_short_from_report(report_data) is None # --------------------------------------------------------------------------- # 20. test_merge_needed_false_no_claim (문제 3: 경합 해소) # --------------------------------------------------------------------------- def test_merge_needed_false_no_claim(tmp_path: Path): """merge_needed=False인 경우 .done.clear를 생성하지 않고 .done 파일도 건드리지 않는다.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) # .done 파일 생성 (merge_needed=False) done_file = events_dir / "task-500.1.done" done_data = { "task_id": "task-500.1", "status": "done", "merge_needed": False, } done_file.write_text(json.dumps(done_data), encoding="utf-8") # 보고서 파일 생성 report_file = reports_dir / "task-500.1.md" report_file.write_text("# task-500.1\n머지 불필요\n", encoding="utf-8") mock_report = { "task_id": "task-500.1", "merge_needed": False, "merge_branch": None, "merge_worktree": None, "files": [], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): result = merger.run() # .done.clear가 생성되지 않았어야 함 clear_file = events_dir / "task-500.1.done.clear" assert not clear_file.exists(), ".done.clear가 생성되면 안 됨 (경합 방지)" # .done 파일이 유지되어야 함 assert done_file.exists(), ".done 파일이 유지되어야 함 (notify-completion에 위임)" assert result["skipped"] == 1 # --------------------------------------------------------------------------- # 21. test_notify_anu_called_after_merge (문제 3: 아누 통보) # --------------------------------------------------------------------------- def test_notify_anu_called_after_merge(tmp_path: Path): """머지 성공 후 notify_anu()가 호출되는지 확인.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) done_file = events_dir / "task-501.1.done" done_data = { "task_id": "task-501.1", "status": "done", "merge_needed": True, "merge_branch": "task/task-501.1-dev1", "team_id": "dev1-team", } done_file.write_text(json.dumps(done_data), encoding="utf-8") report_file = reports_dir / "task-501.1.md" report_file.write_text("# task-501.1\n", encoding="utf-8") mock_report = { "task_id": "task-501.1", "merge_needed": True, "merge_branch": "task/task-501.1-dev1", "merge_worktree": "/home/jay/projects/TestApp/.worktrees/task-501.1-dev1", "files": [], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): with patch.object( merger, "execute_merge", return_value={"status": "merged", "branch": "task/task-501.1-dev1"}, ): with patch.object( merger, "run_tests", return_value={"passed": True, "output": "5 passed", "duration": 1.0}, ): with patch.object(merger, "notify_anu") as mock_notify: result = merger.run() mock_notify.assert_called_once_with("task-501.1") assert result["merged"] == 1 # --------------------------------------------------------------------------- # 22. test_team_short_used_in_execute_merge (문제 2: 변환된 team_short 사용) # --------------------------------------------------------------------------- def test_team_short_used_in_execute_merge(tmp_path: Path): """execute_merge에 team_id가 아닌 team_short(dev1)가 전달되는지 확인.""" events_dir = tmp_path / "memory" / "events" events_dir.mkdir(parents=True) reports_dir = tmp_path / "memory" / "reports" reports_dir.mkdir(parents=True) done_file = events_dir / "task-502.1.done" done_data = { "task_id": "task-502.1", "status": "done", "merge_needed": True, "team_id": "dev1-team", # team_id에 '-team' suffix 포함 } done_file.write_text(json.dumps(done_data), encoding="utf-8") report_file = reports_dir / "task-502.1.md" report_file.write_text("# task-502.1\n", encoding="utf-8") mock_report = { "task_id": "task-502.1", "merge_needed": True, "merge_branch": "task/task-502.1-dev1", "merge_worktree": "/home/jay/projects/TestApp/.worktrees/task-502.1-dev1", "files": ["/home/jay/projects/TestApp/main.py"], } merger = AutoMerger(workspace_path=str(tmp_path)) with patch("auto_merge.parse_report", return_value=mock_report): with patch.object( merger, "execute_merge", return_value={"status": "merged", "branch": "task/task-502.1-dev1"}, ) as mock_merge: with patch.object( merger, "run_tests", return_value={"passed": True, "output": "ok", "duration": 0.5}, ): with patch.object(merger, "notify_anu"): merger.run() # execute_merge의 team_id 인수가 'dev1'이어야 함 ('dev1-team'이 아닌) mock_merge.assert_called_once() call_args = mock_merge.call_args[0] actual_team_short = call_args[2] assert actual_team_short == "dev1", f"team_short가 'dev1'이어야 하는데 '{actual_team_short}'"