#!/usr/bin/env python3 """test_collect_metrics.py — collect_metrics.py 테스트""" import json import sys from datetime import date from pathlib import Path import pytest # scripts/ 디렉토리를 sys.path에 추가 sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) import collect_metrics # pyright: ignore[reportMissingImports] @pytest.fixture def sample_task_timers(tmp_path): """테스트용 task-timers.json 생성""" data = { "tasks": { "task-1": { "task_id": "task-1", "team_id": "dev1-team", "status": "completed", "end_time": "2026-04-16T10:00:00", "token_usage": {"cost_estimate_usd": 1.5}, }, "task-2": { "task_id": "task-2", "team_id": "dev2-team", "status": "completed", "end_time": "2026-04-16T15:00:00", }, "task-3": { "task_id": "task-3", "team_id": "dev1-team", "status": "running", "end_time": None, }, } } p = tmp_path / "task-timers.json" p.write_text(json.dumps(data), encoding="utf-8") return p # --------------------------------------------------------------------------- # Test 1: load_task_timers # --------------------------------------------------------------------------- def test_load_task_timers(monkeypatch, sample_task_timers): """task-timers.json 파일이 존재할 때 정상 로드 + tasks 키 확인""" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", sample_task_timers) result = collect_metrics.load_task_timers() # 반환값은 "tasks" 딕셔너리의 내용 — dict여야 한다 assert isinstance(result, dict), "load_task_timers()는 dict를 반환해야 한다" # 파일에 담긴 3개 태스크가 모두 로드되어야 한다 assert len(result) == 3, "3개 task가 모두 로드되어야 한다" # 각 task_id가 키로 존재해야 한다 for task_id in ("task-1", "task-2", "task-3"): assert task_id in result, f"{task_id}가 결과에 없다" def test_load_task_timers_missing_file(tmp_path, monkeypatch): """파일이 없을 때 빈 dict 반환 (에러 없음)""" missing = tmp_path / "no-such-file.json" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", missing) result = collect_metrics.load_task_timers() assert result == {}, "파일 없을 때 빈 dict를 반환해야 한다" # --------------------------------------------------------------------------- # Test 2: collect_daily_completed # --------------------------------------------------------------------------- def test_collect_daily_completed(monkeypatch, sample_task_timers): """M-3: 특정 날짜 완료 task 수 + by_team 분류 검증""" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", sample_task_timers) target = date(2026, 4, 16) result = collect_metrics.collect_daily_completed(target) # task-1, task-2 모두 2026-04-16 완료 / task-3은 running → 미포함 assert result["value"] == 2, "2026-04-16에 완료된 task는 2개여야 한다" assert "by_team" in result, "by_team 키가 있어야 한다" assert result["by_team"].get("dev1-team") == 1, "dev1-team task 1개" assert result["by_team"].get("dev2-team") == 1, "dev2-team task 1개" def test_collect_daily_completed_no_match(monkeypatch, sample_task_timers): """날짜가 다른 경우 count=0""" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", sample_task_timers) target = date(2026, 4, 15) # 해당 날짜에 완료된 task 없음 result = collect_metrics.collect_daily_completed(target) assert result["value"] == 0, "해당 날짜 완료 task가 없으면 0이어야 한다" assert result["by_team"] == {}, "by_team도 비어 있어야 한다" # --------------------------------------------------------------------------- # Test 3: collect_ai_cost # --------------------------------------------------------------------------- def test_collect_ai_cost(monkeypatch, sample_task_timers): """M-7: token_usage 있는 task와 없는 task 혼재 시 정상 합산""" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", sample_task_timers) target = date(2026, 4, 16) result = collect_metrics.collect_ai_cost(target) # task-1만 token_usage 보유 (cost=1.5), task-2는 token_usage 없음 → 스킵 assert result["value"] == pytest.approx(1.5), "AI 비용 합산이 1.5 USD여야 한다" assert result["currency"] == "USD" assert result["task_count"] == 1, "비용이 기록된 task는 1개여야 한다" def test_collect_ai_cost_no_token_usage(tmp_path, monkeypatch): """token_usage 없는 task만 있을 때 cost=0, 에러 없음""" data = { "tasks": { "t-a": { "task_id": "t-a", "status": "completed", "end_time": "2026-04-16T09:00:00", # token_usage 키 아예 없음 }, "t-b": { "task_id": "t-b", "status": "completed", "end_time": "2026-04-16T11:00:00", "token_usage": None, # 명시적 None }, } } p = tmp_path / "task-timers-no-cost.json" p.write_text(json.dumps(data), encoding="utf-8") monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", p) target = date(2026, 4, 16) result = collect_metrics.collect_ai_cost(target) assert result["value"] == 0.0, "비용 없는 task만 있으면 합산은 0.0이어야 한다" assert result["task_count"] == 0 # --------------------------------------------------------------------------- # Test 4: collect_all — 반환 구조 검증 # --------------------------------------------------------------------------- def test_collect_all_structure(tmp_path, monkeypatch, sample_task_timers): """collect_all() 반환 dict의 최상위 키 및 6개 지표 키 존재 확인""" monkeypatch.setattr(collect_metrics, "TASK_TIMERS_PATH", sample_task_timers) # WORKSPACE_ROOT를 tmp_path로 교체하여 REPORTS_DIR / EVENTS_DIR 접근 격리 monkeypatch.setattr(collect_metrics, "WORKSPACE_ROOT", str(tmp_path)) monkeypatch.setattr(collect_metrics, "REPORTS_DIR", tmp_path / "memory" / "reports") monkeypatch.setattr(collect_metrics, "EVENTS_DIR", tmp_path / "memory" / "events") # git log subprocess mock monkeypatch.setattr(collect_metrics, "_run_git_log", lambda _since, _until: []) target = date(2026, 4, 16) result = collect_metrics.collect_all(target) # 최상위 키 검증 for key in ("date", "collected_at", "period", "metrics"): assert key in result, f"최상위 키 '{key}'가 없다" assert result["date"] == "2026-04-16" assert "start" in result["period"] and "end" in result["period"] # 6개 지표 키 검증 expected_metric_keys = { "M-1_merge_conflicts", "M-2_revert_count", "M-3_daily_completed_tasks", "M-5_qc_fail_count", "M-6_g1_fail_count", "M-7_total_ai_cost", } actual_keys = set(result["metrics"].keys()) assert actual_keys == expected_metric_keys, ( f"metrics 키 불일치: 기대={expected_metric_keys}, 실제={actual_keys}" ) # --------------------------------------------------------------------------- # Test 5: save_metrics — 파일 저장 검증 # --------------------------------------------------------------------------- def test_save_metrics(tmp_path, monkeypatch): """save_metrics()가 임시 디렉토리에 유효한 JSON 파일을 생성하는지 검증""" output_dir = tmp_path / "daily" monkeypatch.setattr(collect_metrics, "OUTPUT_DIR", output_dir) sample_data = { "date": "2026-04-16", "collected_at": "2026-04-16T12:00:00", "period": {"start": "2026-04-16", "end": "2026-04-16"}, "metrics": { "M-1_merge_conflicts": {"value": 0}, }, } target = date(2026, 4, 16) out_path = collect_metrics.save_metrics(sample_data, target) # 파일 존재 확인 assert out_path.exists(), "저장된 파일이 존재해야 한다" assert out_path.name == "metrics-2026-04-16.json" # JSON 유효성 확인 loaded = json.loads(out_path.read_text(encoding="utf-8")) assert loaded["date"] == "2026-04-16" assert "metrics" in loaded