"""TDD RED Phase 5 테스트 모음. 대상 모듈 (아직 미구현): - orchestrator.token_ledger : TokenLedger 클래스 Phase 5 확장 메서드 - orchestrator.auto_orch : send_telegram_alert, check_stale_tasks, 확장된 함수들 작성자 : 모리건 (dev3-team tester) 날짜 : 2026-03-24 """ import datetime import json import os import sys import tempfile from pathlib import Path from unittest.mock import MagicMock, patch import pytest # --------------------------------------------------------------------------- # sys.path 세팅 # --------------------------------------------------------------------------- WORKSPACE_ROOT = "/home/jay/workspace" if WORKSPACE_ROOT not in sys.path: sys.path.insert(0, WORKSPACE_ROOT) # --------------------------------------------------------------------------- # 기존 함수 import (이미 구현된 것들) # --------------------------------------------------------------------------- from orchestrator.auto_orch import ( # noqa: E402 cmd_scan, cmd_status, update_health, ) from orchestrator.token_ledger import TokenLedger # noqa: E402 # --------------------------------------------------------------------------- # Phase 5 신규 함수 import (아직 미구현 — RED 단계이므로 ImportError 예상) # --------------------------------------------------------------------------- try: from orchestrator.auto_orch import ( # noqa: E402 check_stale_tasks, send_telegram_alert, ) except ImportError: send_telegram_alert = None # type: ignore[assignment] check_stale_tasks = None # type: ignore[assignment] # --------------------------------------------------------------------------- # Phase 5 상수 (token_ledger.py에 추가될 예정) # --------------------------------------------------------------------------- try: from orchestrator.token_ledger import ( # noqa: E402 CONSERVATIVE_MULTIPLIER, TOKENS_PER_MINUTE_OPUS, TOKENS_PER_MINUTE_SONNET, ) except ImportError: TOKENS_PER_MINUTE_OPUS = 5000 TOKENS_PER_MINUTE_SONNET = 3000 CONSERVATIVE_MULTIPLIER = 1.2 # --------------------------------------------------------------------------- # Phase 5 상수 (auto_orch.py에 추가될 예정) # --------------------------------------------------------------------------- try: from orchestrator.auto_orch import ( # noqa: E402 ALERTS_SENT_PATH, STALE_TASK_RUNNING_SECONDS, TASK_TIMERS_PATH, ) except ImportError: ALERTS_SENT_PATH = os.path.join(WORKSPACE_ROOT, "orchestrator/state/alerts_sent.json") TASK_TIMERS_PATH = os.path.join(WORKSPACE_ROOT, "memory/task-timers.json") STALE_TASK_RUNNING_SECONDS = 7200 # =========================================================================== # TestTokenLedgerPhase5 — TokenLedger 확장 메서드 테스트 # =========================================================================== class TestTokenLedgerPhase5: """TokenLedger Phase 5 확장 메서드 테스트 스위트.""" def test_estimate_session_tokens_sonnet(self, tmp_path): """Sonnet 세션(dev1-team, 600초) → 3000 * 10 * 1.2 = 36000 토큰.""" ledger_path = tmp_path / "token_ledger.json" ledger = TokenLedger(str(ledger_path)) result = ledger.estimate_session_tokens("dev1-team", 600.0) # 3000 tokens/min * (600 / 60) min * 1.2 = 36000 assert result == 36000, f"Sonnet 세션(600초) 추정값이 36000이 아님: {result}" def test_estimate_session_tokens_opus(self, tmp_path): """Opus 세션(팀장 포함 team_id, 120초) → Sonnet보다 높은 추정값 반환.""" ledger_path = tmp_path / "token_ledger.json" ledger = TokenLedger(str(ledger_path)) # Sonnet(120초): 3000 * 2 * 1.2 = 7200 # Opus(120초): 5000 * 2 * 1.2 = 12000 sonnet_result = ledger.estimate_session_tokens("dev1-team", 120.0) # 팀장 식별 가능한 team_id 사용 (예: "팀장" 포함 또는 구현에 따라 다름) # API 계약에 따르면 team_id 이름으로 Opus/Sonnet 판단 # 여기서는 명시적으로 Opus 식별 team_id를 사용 opus_result = ledger.estimate_session_tokens("팀장-dev1-team", 120.0) assert ( opus_result > sonnet_result ), f"Opus 세션 추정값({opus_result})이 Sonnet 세션 추정값({sonnet_result})보다 크지 않음" # Opus(120초): 5000 * 2 * 1.2 = 12000 assert opus_result == 12000, f"Opus 세션(120초) 추정값이 12000이 아님: {opus_result}" def test_estimate_session_tokens_zero_duration(self, tmp_path): """0초 세션 → 0 토큰 반환.""" ledger_path = tmp_path / "token_ledger.json" ledger = TokenLedger(str(ledger_path)) result = ledger.estimate_session_tokens("dev1-team", 0.0) assert result == 0, f"0초 세션 추정값이 0이 아님: {result}" def test_get_daily_usage_summary(self, tmp_path): """daily_total=450000 → {"today": 450000, "limit": 1000000, "remaining": 550000, "percentage": 45.0}.""" ledger_path = tmp_path / "token_ledger.json" # 사전 상태 파일 작성 state = { "date": datetime.date.today().isoformat(), "daily_total": 450000, "pipelines": {"pipe-001": 450000}, } ledger_path.write_text(json.dumps(state), encoding="utf-8") ledger = TokenLedger(str(ledger_path)) summary = ledger.get_daily_usage_summary() assert summary["today"] == 450000, f"today 값 불일치: {summary['today']}" assert summary["limit"] == 1_000_000, f"limit 값 불일치: {summary['limit']}" assert summary["remaining"] == 550000, f"remaining 값 불일치: {summary['remaining']}" assert summary["percentage"] == 45.0, f"percentage 값 불일치: {summary['percentage']}" def test_get_daily_usage_summary_empty(self, tmp_path): """신규 레저 → {"today": 0, "limit": 1000000, "remaining": 1000000, "percentage": 0.0}.""" ledger_path = tmp_path / "token_ledger_empty.json" ledger = TokenLedger(str(ledger_path)) summary = ledger.get_daily_usage_summary() assert summary["today"] == 0, f"신규 레저 today 값 불일치: {summary['today']}" assert summary["limit"] == 1_000_000, f"신규 레저 limit 값 불일치: {summary['limit']}" assert summary["remaining"] == 1_000_000, f"신규 레저 remaining 값 불일치: {summary['remaining']}" assert summary["percentage"] == 0.0, f"신규 레저 percentage 값 불일치: {summary['percentage']}" def test_check_warning_threshold_below(self, tmp_path): """daily_total=700000, threshold=0.8 → False (700000 < 800000).""" ledger_path = tmp_path / "token_ledger.json" state = { "date": datetime.date.today().isoformat(), "daily_total": 700000, "pipelines": {}, } ledger_path.write_text(json.dumps(state), encoding="utf-8") ledger = TokenLedger(str(ledger_path)) result = ledger.check_warning_threshold(threshold=0.8) assert result is False, f"700000 < 800000인데 check_warning_threshold가 True를 반환함: {result}" def test_check_warning_threshold_at(self, tmp_path): """daily_total=800000, threshold=0.8 → True (800000 == 800000).""" ledger_path = tmp_path / "token_ledger.json" state = { "date": datetime.date.today().isoformat(), "daily_total": 800000, "pipelines": {}, } ledger_path.write_text(json.dumps(state), encoding="utf-8") ledger = TokenLedger(str(ledger_path)) result = ledger.check_warning_threshold(threshold=0.8) assert result is True, f"800000 == 800000인데 check_warning_threshold가 False를 반환함: {result}" def test_check_warning_threshold_above(self, tmp_path): """daily_total=900000, threshold=0.8 → True (900000 > 800000).""" ledger_path = tmp_path / "token_ledger.json" state = { "date": datetime.date.today().isoformat(), "daily_total": 900000, "pipelines": {}, } ledger_path.write_text(json.dumps(state), encoding="utf-8") ledger = TokenLedger(str(ledger_path)) result = ledger.check_warning_threshold(threshold=0.8) assert result is True, f"900000 > 800000인데 check_warning_threshold가 False를 반환함: {result}" # =========================================================================== # TestTelegramAlert — send_telegram_alert 테스트 # =========================================================================== class TestTelegramAlert: """send_telegram_alert 함수 테스트 스위트.""" def test_send_telegram_alert_success(self, tmp_path, monkeypatch): """mock requests.post → 200 → True 반환, alerts_sent.json에 기록됨.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") alerts_path = tmp_path / "alerts_sent.json" monkeypatch.setenv("ANU_BOT_TOKEN", "test-bot-token-12345") monkeypatch.setenv("AUTO_ORCH_CHAT_ID", "987654321") monkeypatch.setattr("orchestrator.auto_orch.ALERTS_SENT_PATH", str(alerts_path)) mock_response = MagicMock() mock_response.status_code = 200 with patch("requests.post", return_value=mock_response) as mock_post: result = send_telegram_alert("토큰 경고: 80% 초과", "token_warning") assert result is True, f"200 응답인데 send_telegram_alert가 False를 반환함: {result}" assert mock_post.called, "requests.post가 호출되지 않음" # alerts_sent.json에 기록 확인 assert alerts_path.exists(), "alerts_sent.json이 생성되지 않음" data = json.loads(alerts_path.read_text(encoding="utf-8")) today = datetime.date.today().isoformat() assert today in data, f"alerts_sent.json에 오늘 날짜({today})가 없음: {data}" assert "token_warning" in data[today], f"alerts_sent.json에 alert_type이 없음: {data}" def test_send_telegram_alert_dedup(self, tmp_path, monkeypatch): """같은 날 같은 alert_type → 두 번째 호출 False 반환.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") alerts_path = tmp_path / "alerts_sent.json" monkeypatch.setenv("ANU_BOT_TOKEN", "test-bot-token-12345") monkeypatch.setenv("AUTO_ORCH_CHAT_ID", "987654321") monkeypatch.setattr("orchestrator.auto_orch.ALERTS_SENT_PATH", str(alerts_path)) mock_response = MagicMock() mock_response.status_code = 200 with patch("requests.post", return_value=mock_response): first_result = send_telegram_alert("첫 번째 경고", "token_warning") second_result = send_telegram_alert("두 번째 경고", "token_warning") assert first_result is True, f"첫 번째 호출이 True가 아님: {first_result}" assert second_result is False, f"중복 alert_type인데 두 번째 호출이 False가 아님: {second_result}" def test_send_telegram_alert_no_token(self, tmp_path, monkeypatch): """ANU_BOT_TOKEN 미설정 → False 반환.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") alerts_path = tmp_path / "alerts_sent.json" monkeypatch.delenv("ANU_BOT_TOKEN", raising=False) monkeypatch.setenv("AUTO_ORCH_CHAT_ID", "987654321") monkeypatch.setattr("orchestrator.auto_orch.ALERTS_SENT_PATH", str(alerts_path)) result = send_telegram_alert("경고 메시지", "token_warning") assert result is False, f"ANU_BOT_TOKEN 미설정인데 True를 반환함: {result}" def test_send_telegram_alert_new_day(self, tmp_path, monkeypatch): """날짜 변경 시 기존 기록 무시, 재발송 가능 → True 반환.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") alerts_path = tmp_path / "alerts_sent.json" # 어제 날짜로 이미 발송된 기록 작성 yesterday = (datetime.date.today() - datetime.timedelta(days=1)).isoformat() existing_data = {yesterday: ["token_warning"]} alerts_path.write_text(json.dumps(existing_data), encoding="utf-8") monkeypatch.setenv("ANU_BOT_TOKEN", "test-bot-token-12345") monkeypatch.setenv("AUTO_ORCH_CHAT_ID", "987654321") monkeypatch.setattr("orchestrator.auto_orch.ALERTS_SENT_PATH", str(alerts_path)) mock_response = MagicMock() mock_response.status_code = 200 with patch("requests.post", return_value=mock_response): result = send_telegram_alert("오늘 새 경고", "token_warning") assert result is True, f"날짜 변경 후 재발송인데 False를 반환함: {result}" # =========================================================================== # TestStaleTaskCheck — check_stale_tasks 테스트 # =========================================================================== class TestStaleTaskCheck: """check_stale_tasks 함수 테스트 스위트.""" def test_check_stale_tasks_found(self, tmp_path, monkeypatch): """task-timers.json에 2시간+ running 태스크 → 목록 반환.""" if check_stale_tasks is None: pytest.fail("check_stale_tasks가 orchestrator.auto_orch에 구현되지 않음") task_timers_path = tmp_path / "task-timers.json" monkeypatch.setattr("orchestrator.auto_orch.TASK_TIMERS_PATH", str(task_timers_path)) # 현재 시각 기준 3시간 전 시작된 태스크 (stale) now = datetime.datetime.now(datetime.timezone.utc) stale_start = (now - datetime.timedelta(hours=3)).isoformat() fresh_start = (now - datetime.timedelta(minutes=30)).isoformat() task_timers_data = { "stale-task-001": { "team_id": "dev1-team", "status": "running", "start_time": stale_start, }, "stale-task-002": { "team_id": "dev2-team", "status": "running", "start_time": stale_start, }, "fresh-task-001": { "team_id": "dev3-team", "status": "running", "start_time": fresh_start, }, "done-task-001": { "team_id": "dev1-team", "status": "done", "start_time": stale_start, }, } task_timers_path.write_text(json.dumps(task_timers_data), encoding="utf-8") result = check_stale_tasks() assert isinstance(result, list), f"check_stale_tasks 반환값이 list가 아님: {type(result)}" assert len(result) == 2, f"stale 태스크가 2개여야 하는데 {len(result)}개 반환됨: {result}" task_ids = [item["task_id"] for item in result] assert "stale-task-001" in task_ids, f"stale-task-001이 결과에 없음: {task_ids}" assert "stale-task-002" in task_ids, f"stale-task-002이 결과에 없음: {task_ids}" assert "fresh-task-001" not in task_ids, f"fresh 태스크가 stale 목록에 포함됨: {task_ids}" assert "done-task-001" not in task_ids, f"done 태스크가 stale 목록에 포함됨: {task_ids}" # 반환 구조 확인: {"task_id": str, "team_id": str, "duration_seconds": float} for item in result: assert "task_id" in item, f"task_id 키 없음: {item}" assert "team_id" in item, f"team_id 키 없음: {item}" assert "duration_seconds" in item, f"duration_seconds 키 없음: {item}" assert ( item["duration_seconds"] >= STALE_TASK_RUNNING_SECONDS ), f"duration_seconds({item['duration_seconds']})가 임계값({STALE_TASK_RUNNING_SECONDS})보다 작음" def test_check_stale_tasks_none(self, tmp_path, monkeypatch): """stale 태스크 없음 → 빈 리스트 반환.""" if check_stale_tasks is None: pytest.fail("check_stale_tasks가 orchestrator.auto_orch에 구현되지 않음") task_timers_path = tmp_path / "task-timers.json" monkeypatch.setattr("orchestrator.auto_orch.TASK_TIMERS_PATH", str(task_timers_path)) now = datetime.datetime.now(datetime.timezone.utc) fresh_start = (now - datetime.timedelta(minutes=30)).isoformat() task_timers_data = { "fresh-task-001": { "team_id": "dev1-team", "status": "running", "start_time": fresh_start, }, "done-task-001": { "team_id": "dev2-team", "status": "done", "start_time": (now - datetime.timedelta(hours=5)).isoformat(), }, } task_timers_path.write_text(json.dumps(task_timers_data), encoding="utf-8") result = check_stale_tasks() assert isinstance(result, list), f"check_stale_tasks 반환값이 list가 아님: {type(result)}" assert len(result) == 0, f"stale 태스크가 없어야 하는데 {len(result)}개 반환됨: {result}" # =========================================================================== # TestHealthJsonExtension — update_health 확장 테스트 # =========================================================================== class TestHealthJsonExtension: """update_health Phase 5 확장 테스트 스위트.""" def test_update_health_with_token_usage(self, tmp_path, monkeypatch): """token_usage 전달 시 health.json에 포함, version "1.1".""" health_path = tmp_path / "health.json" monkeypatch.setattr("orchestrator.auto_orch.HEALTH_PATH", str(health_path)) token_usage = { "today": 450000, "limit": 1_000_000, "percentage": 45.0, } update_health(active_pipelines=2, errors=0, token_usage=token_usage) assert health_path.exists(), "update_health 후 health.json이 생성되지 않음" data = json.loads(health_path.read_text(encoding="utf-8")) assert "token_usage" in data, f"health.json에 token_usage 키 없음: {data}" assert data["token_usage"]["today"] == 450000, f"token_usage.today 불일치: {data['token_usage']}" assert data["token_usage"]["limit"] == 1_000_000, f"token_usage.limit 불일치: {data['token_usage']}" assert data["token_usage"]["percentage"] == 45.0, f"token_usage.percentage 불일치: {data['token_usage']}" assert data.get("version") == "1.1", f"version이 '1.1'이 아님: {data.get('version')}" def test_update_health_without_token_usage(self, tmp_path, monkeypatch): """token_usage=None → 기존 포맷 유지 (하위호환).""" health_path = tmp_path / "health.json" monkeypatch.setattr("orchestrator.auto_orch.HEALTH_PATH", str(health_path)) update_health(active_pipelines=1, errors=0, token_usage=None) assert health_path.exists(), "update_health 후 health.json이 생성되지 않음" data = json.loads(health_path.read_text(encoding="utf-8")) # 기존 필드들이 유지되어야 함 assert "active_pipelines" in data, f"active_pipelines 키 없음: {data}" assert data["active_pipelines"] == 1, f"active_pipelines 값 불일치: {data}" assert "last_tick" in data, f"last_tick 키 없음: {data}" # token_usage가 없거나 None이어야 함 assert ( "token_usage" not in data or data["token_usage"] is None ), f"token_usage=None인데 health.json에 token_usage가 포함됨: {data}" # =========================================================================== # TestCmdStatusExtension — cmd_status Phase 5 출력 확장 테스트 # =========================================================================== class TestCmdStatusExtension: """cmd_status Phase 5 확장 출력 테스트 스위트.""" def _setup_state_and_ledger(self, tmp_path, monkeypatch): """공통 fixture: state 디렉터리와 token_ledger 설정.""" state_dir = tmp_path / "state" state_dir.mkdir() monkeypatch.setattr("orchestrator.auto_orch.STATE_DIR", str(state_dir)) ledger_path = state_dir / "token_ledger.json" monkeypatch.setattr("orchestrator.auto_orch.LEDGER_PATH", str(ledger_path)) ledger_state = { "date": datetime.date.today().isoformat(), "daily_total": 450000, "pipelines": {}, } ledger_path.write_text(json.dumps(ledger_state), encoding="utf-8") return state_dir, ledger_path def test_cmd_status_includes_token_usage(self, tmp_path, monkeypatch, capsys): """cmd_status 출력에 '토큰 사용량' 섹션이 포함되어야 한다.""" self._setup_state_and_ledger(tmp_path, monkeypatch) cmd_status() captured = capsys.readouterr() output = captured.out + captured.err assert "토큰 사용량" in output, f"cmd_status 출력에 '토큰 사용량' 섹션 없음:\n{output}" def test_cmd_status_includes_warning_threshold(self, tmp_path, monkeypatch, capsys): """cmd_status 출력에 '경고 임계값' 텍스트가 포함되어야 한다.""" self._setup_state_and_ledger(tmp_path, monkeypatch) cmd_status() captured = capsys.readouterr() output = captured.out + captured.err assert "경고 임계값" in output, f"cmd_status 출력에 '경고 임계값' 텍스트 없음:\n{output}" # =========================================================================== # TestCmdScanIntegration — cmd_scan Phase 5 통합 테스트 # =========================================================================== class TestCmdScanIntegration: """cmd_scan Phase 5 통합 테스트 — 알림 및 stale 태스크 감지.""" def _setup_scan_env(self, tmp_path, monkeypatch): """공통 fixture: cmd_scan 실행에 필요한 환경 설정.""" state_dir = tmp_path / "state" state_dir.mkdir() incoming_dir = tmp_path / "incoming" incoming_dir.mkdir() processed_dir = tmp_path / "processed" processed_dir.mkdir() health_path = tmp_path / "health.json" task_timers_path = tmp_path / "task-timers.json" alerts_path = tmp_path / "alerts_sent.json" ledger_path = state_dir / "token_ledger.json" monkeypatch.setattr("orchestrator.auto_orch.STATE_DIR", str(state_dir)) monkeypatch.setattr("orchestrator.auto_orch.INCOMING_DIR", str(incoming_dir)) monkeypatch.setattr("orchestrator.auto_orch.PROCESSED_DIR", str(processed_dir)) monkeypatch.setattr("orchestrator.auto_orch.HEALTH_PATH", str(health_path)) monkeypatch.setattr("orchestrator.auto_orch.TASK_TIMERS_PATH", str(task_timers_path)) monkeypatch.setattr("orchestrator.auto_orch.ALERTS_SENT_PATH", str(alerts_path)) monkeypatch.setattr("orchestrator.auto_orch.LEDGER_PATH", str(ledger_path)) return { "state_dir": state_dir, "incoming_dir": incoming_dir, "processed_dir": processed_dir, "health_path": health_path, "task_timers_path": task_timers_path, "alerts_path": alerts_path, } def test_cmd_scan_sends_alert_at_threshold(self, tmp_path, monkeypatch): """80% 초과 시 send_telegram_alert가 호출되어야 한다.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") paths = self._setup_scan_env(tmp_path, monkeypatch) # 토큰 사용량 90% (임계값 초과) ledger_path = paths["state_dir"] / "token_ledger.json" ledger_state = { "date": datetime.date.today().isoformat(), "daily_total": 900000, "pipelines": {}, } ledger_path.write_text(json.dumps(ledger_state), encoding="utf-8") # task-timers.json: 빈 파일 (stale 없음) paths["task_timers_path"].write_text(json.dumps({}), encoding="utf-8") with patch("orchestrator.auto_orch.send_telegram_alert") as mock_alert: mock_alert.return_value = True with patch("orchestrator.auto_orch.acquire_global_lock", return_value=42): with patch("orchestrator.auto_orch.release_global_lock"): cmd_scan() # send_telegram_alert가 최소 1번 호출되어야 함 assert mock_alert.called, "80% 초과인데 send_telegram_alert가 호출되지 않음" # 호출 인자 확인: alert_type이 토큰 관련이어야 함 call_args_list = mock_alert.call_args_list alert_types = [call.args[1] if len(call.args) > 1 else call.kwargs.get("alert_type") for call in call_args_list] token_alert_called = any(at is not None and "token" in str(at).lower() for at in alert_types) assert token_alert_called, f"token 관련 alert_type으로 호출되지 않음: {alert_types}" def test_cmd_scan_checks_stale_tasks(self, tmp_path, monkeypatch): """stale task 존재 시 send_telegram_alert가 호출되어야 한다.""" if send_telegram_alert is None: pytest.fail("send_telegram_alert가 orchestrator.auto_orch에 구현되지 않음") if check_stale_tasks is None: pytest.fail("check_stale_tasks가 orchestrator.auto_orch에 구현되지 않음") paths = self._setup_scan_env(tmp_path, monkeypatch) # 토큰 사용량 낮음 (경고 임계값 미달) ledger_path = paths["state_dir"] / "token_ledger.json" ledger_state = { "date": datetime.date.today().isoformat(), "daily_total": 100000, "pipelines": {}, } ledger_path.write_text(json.dumps(ledger_state), encoding="utf-8") # stale 태스크가 있는 task-timers.json now = datetime.datetime.now(datetime.timezone.utc) stale_start = (now - datetime.timedelta(hours=3)).isoformat() task_timers_data = { "stale-task-999": { "team_id": "dev1-team", "status": "running", "start_time": stale_start, } } paths["task_timers_path"].write_text(json.dumps(task_timers_data), encoding="utf-8") with patch("orchestrator.auto_orch.send_telegram_alert") as mock_alert: mock_alert.return_value = True with patch("orchestrator.auto_orch.acquire_global_lock", return_value=42): with patch("orchestrator.auto_orch.release_global_lock"): cmd_scan() # send_telegram_alert가 stale 관련으로 호출되어야 함 assert mock_alert.called, "stale 태스크 존재인데 send_telegram_alert가 호출되지 않음" call_args_list = mock_alert.call_args_list alert_types = [call.args[1] if len(call.args) > 1 else call.kwargs.get("alert_type") for call in call_args_list] stale_alert_called = any(at is not None and "stale" in str(at).lower() for at in alert_types) assert stale_alert_called, f"stale 관련 alert_type으로 호출되지 않음: {alert_types}"