cfe2901a55
- itsm/ -> workspace/guardia-itsm/ - manager/ -> workspace/guardia-manager/ - app/ -> workspace/guardia-messenger/ - manual/ -> workspace/guardia-docs/ workspace/zioinfo-web/ unchanged. git mv preserves full commit history. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
507 lines
19 KiB
Python
507 lines
19 KiB
Python
"""
|
|
테스트 관리 API — 단위/통합/UAT 테스트 계획·케이스·실행·결함.
|
|
|
|
엔드포인트:
|
|
Plans:
|
|
GET /api/si/projects/{pid}/test-plans — 테스트 계획 목록
|
|
POST /api/si/projects/{pid}/test-plans — 테스트 계획 등록
|
|
GET /api/si/projects/{pid}/test-plans/{id} — 상세
|
|
PATCH /api/si/projects/{pid}/test-plans/{id} — 수정
|
|
DELETE /api/si/projects/{pid}/test-plans/{id} — 삭제
|
|
|
|
Cases:
|
|
GET /api/si/projects/{pid}/test-plans/{pid2}/cases — 케이스 목록
|
|
POST /api/si/projects/{pid}/test-plans/{pid2}/cases — 케이스 등록
|
|
PATCH /api/si/projects/{pid}/test-plans/{pid2}/cases/{id} — 케이스 수정
|
|
DELETE /api/si/projects/{pid}/test-plans/{pid2}/cases/{id} — 케이스 삭제
|
|
|
|
Executions:
|
|
GET /api/si/test-cases/{id}/executions — 실행 이력
|
|
POST /api/si/test-cases/{id}/execute — 테스트 실행 (결과 입력)
|
|
|
|
Defects:
|
|
GET /api/si/projects/{pid}/defects — 결함 목록
|
|
POST /api/si/projects/{pid}/defects — 결함 등록
|
|
GET /api/si/projects/{pid}/defects/{id} — 결함 상세
|
|
PATCH /api/si/projects/{pid}/defects/{id} — 결함 수정
|
|
PATCH /api/si/projects/{pid}/defects/{id}/fix — 수정 완료
|
|
PATCH /api/si/projects/{pid}/defects/{id}/verify — 검증 완료
|
|
GET /api/si/projects/{pid}/defects/stats — 결함 통계
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import List, Optional
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select, desc
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import get_current_user
|
|
from database import get_db
|
|
from models import (
|
|
SiProject,
|
|
SiTestPlan, SiTestPlanCreate, SiTestPlanOut,
|
|
SiTestCase, SiTestCaseCreate, SiTestCaseOut,
|
|
SiTestExecution, SiTestExecutionCreate, SiTestExecutionOut,
|
|
SiDefect, SiDefectCreate, SiDefectOut, SiDefectUpdate,
|
|
SiTestResult, DefectSeverity, DefectStatus,
|
|
User, UserRole,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter(prefix="/api/si", tags=["si-tests"])
|
|
|
|
|
|
def _new_tc_id() -> str:
|
|
return f"TC-{datetime.now().strftime('%Y%m%d')}-{str(uuid4())[:5].upper()}"
|
|
|
|
|
|
def _new_defect_id() -> str:
|
|
return f"DEF-{datetime.now().strftime('%Y%m%d')}-{str(uuid4())[:5].upper()}"
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 테스트 계획
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/projects/{project_id}/test-plans", response_model=List[SiTestPlanOut])
|
|
async def list_test_plans(
|
|
project_id: int,
|
|
test_type: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
await _assert_project(project_id, db)
|
|
q = select(SiTestPlan).where(SiTestPlan.project_id == project_id)
|
|
if test_type:
|
|
q = q.where(SiTestPlan.test_type == test_type)
|
|
q = q.order_by(SiTestPlan.planned_start)
|
|
return (await db.execute(q)).scalars().all()
|
|
|
|
|
|
@router.post("/projects/{project_id}/test-plans", response_model=SiTestPlanOut, status_code=201)
|
|
async def create_test_plan(
|
|
project_id: int,
|
|
body: SiTestPlanCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM, UserRole.ENGINEER):
|
|
raise HTTPException(403, "엔지니어 이상 권한 필요")
|
|
await _assert_project(project_id, db)
|
|
|
|
plan = SiTestPlan(
|
|
project_id = project_id,
|
|
phase = body.phase,
|
|
test_type = body.test_type,
|
|
plan_name = body.plan_name,
|
|
description = body.description,
|
|
planned_start = body.planned_start,
|
|
planned_end = body.planned_end,
|
|
created_by = body.created_by or current_user.username,
|
|
)
|
|
db.add(plan)
|
|
await db.commit()
|
|
await db.refresh(plan)
|
|
return plan
|
|
|
|
|
|
@router.get("/projects/{project_id}/test-plans/{plan_id}", response_model=SiTestPlanOut)
|
|
async def get_test_plan(
|
|
project_id: int, plan_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
return await _get_plan_or_404(project_id, plan_id, db)
|
|
|
|
|
|
@router.patch("/projects/{project_id}/test-plans/{plan_id}", response_model=SiTestPlanOut)
|
|
async def update_test_plan(
|
|
project_id: int, plan_id: int,
|
|
actual_start: Optional[datetime] = Query(None),
|
|
actual_end: Optional[datetime] = Query(None),
|
|
is_completed: Optional[bool] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
plan = await _get_plan_or_404(project_id, plan_id, db)
|
|
if actual_start:
|
|
plan.actual_start = actual_start
|
|
if actual_end:
|
|
plan.actual_end = actual_end
|
|
if is_completed is not None:
|
|
plan.is_completed = is_completed
|
|
await db.commit()
|
|
await db.refresh(plan)
|
|
return plan
|
|
|
|
|
|
@router.delete("/projects/{project_id}/test-plans/{plan_id}", status_code=204)
|
|
async def delete_test_plan(
|
|
project_id: int, plan_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
if current_user.role not in (UserRole.ADMIN, UserRole.PM):
|
|
raise HTTPException(403, "PM 이상 권한 필요")
|
|
plan = await _get_plan_or_404(project_id, plan_id, db)
|
|
await db.delete(plan)
|
|
await db.commit()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 테스트 케이스
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/projects/{project_id}/test-plans/{plan_id}/cases", response_model=List[SiTestCaseOut])
|
|
async def list_test_cases(
|
|
project_id: int, plan_id: int,
|
|
last_result: Optional[str] = Query(None),
|
|
assigned_to: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
await _get_plan_or_404(project_id, plan_id, db)
|
|
q = select(SiTestCase).where(SiTestCase.plan_id == plan_id)
|
|
if last_result:
|
|
q = q.where(SiTestCase.last_result == last_result)
|
|
if assigned_to:
|
|
q = q.where(SiTestCase.assigned_to == assigned_to)
|
|
q = q.order_by(SiTestCase.tc_id)
|
|
return (await db.execute(q)).scalars().all()
|
|
|
|
|
|
@router.post("/projects/{project_id}/test-plans/{plan_id}/cases",
|
|
response_model=SiTestCaseOut, status_code=201)
|
|
async def create_test_case(
|
|
project_id: int, plan_id: int,
|
|
body: SiTestCaseCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
plan = await _get_plan_or_404(project_id, plan_id, db)
|
|
|
|
tc = SiTestCase(
|
|
tc_id = _new_tc_id(),
|
|
plan_id = plan_id,
|
|
req_id = body.req_id,
|
|
title = body.title,
|
|
precondition = body.precondition,
|
|
test_steps = body.test_steps,
|
|
expected_result = body.expected_result,
|
|
assigned_to = body.assigned_to,
|
|
priority = body.priority,
|
|
)
|
|
db.add(tc)
|
|
await db.flush()
|
|
|
|
# 계획 TC 수 업데이트
|
|
plan.total_cases += 1
|
|
await db.commit()
|
|
await db.refresh(tc)
|
|
return tc
|
|
|
|
|
|
@router.patch("/projects/{project_id}/test-plans/{plan_id}/cases/{tc_id}",
|
|
response_model=SiTestCaseOut)
|
|
async def update_test_case(
|
|
project_id: int, plan_id: int, tc_id: int,
|
|
body: SiTestCaseCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
tc = await _get_tc_or_404(plan_id, tc_id, db)
|
|
for field, val in body.model_dump(exclude_none=True).items():
|
|
setattr(tc, field, val)
|
|
await db.commit()
|
|
await db.refresh(tc)
|
|
return tc
|
|
|
|
|
|
@router.delete("/projects/{project_id}/test-plans/{plan_id}/cases/{tc_id}", status_code=204)
|
|
async def delete_test_case(
|
|
project_id: int, plan_id: int, tc_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
plan = await _get_plan_or_404(project_id, plan_id, db)
|
|
tc = await _get_tc_or_404(plan_id, tc_id, db)
|
|
await db.delete(tc)
|
|
plan.total_cases = max(0, plan.total_cases - 1)
|
|
await db.commit()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 테스트 실행
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/test-cases/{tc_id}/executions", response_model=List[SiTestExecutionOut])
|
|
async def list_executions(
|
|
tc_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
result = await db.execute(
|
|
select(SiTestExecution).where(SiTestExecution.tc_id == tc_id)
|
|
.order_by(desc(SiTestExecution.executed_at))
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("/test-cases/{tc_id}/execute", response_model=SiTestExecutionOut, status_code=201)
|
|
async def execute_test_case(
|
|
tc_id: int,
|
|
body: SiTestExecutionCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""테스트 실행 결과 기록 — TC 최종 결과 및 계획 통계 자동 갱신."""
|
|
tc = (await db.execute(select(SiTestCase).where(SiTestCase.id == tc_id))).scalars().first()
|
|
if not tc:
|
|
raise HTTPException(404, "테스트 케이스를 찾을 수 없습니다")
|
|
|
|
exec_record = SiTestExecution(
|
|
tc_id = tc_id,
|
|
executed_by = body.executed_by or current_user.username,
|
|
result = body.result,
|
|
actual_result = body.actual_result,
|
|
note = body.note,
|
|
build_version = body.build_version,
|
|
)
|
|
db.add(exec_record)
|
|
|
|
# TC 최종 결과 갱신
|
|
prev_result = tc.last_result
|
|
tc.last_result = body.result
|
|
|
|
# 테스트 계획 통계 재계산
|
|
plan = (await db.execute(
|
|
select(SiTestPlan).where(SiTestPlan.id == tc.plan_id)
|
|
)).scalars().first()
|
|
if plan:
|
|
await _recalc_plan_stats(plan, db)
|
|
|
|
await db.commit()
|
|
await db.refresh(exec_record)
|
|
return exec_record
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# 결함 관리
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@router.get("/projects/{project_id}/defects", response_model=List[SiDefectOut])
|
|
async def list_defects(
|
|
project_id: int,
|
|
severity: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
assigned_to: Optional[str] = Query(None),
|
|
phase: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
await _assert_project(project_id, db)
|
|
q = select(SiDefect).where(SiDefect.project_id == project_id)
|
|
if severity:
|
|
q = q.where(SiDefect.severity == severity)
|
|
if status:
|
|
q = q.where(SiDefect.status == status)
|
|
if assigned_to:
|
|
q = q.where(SiDefect.assigned_to == assigned_to)
|
|
if phase:
|
|
q = q.where(SiDefect.phase == phase)
|
|
q = q.order_by(desc(SiDefect.created_at))
|
|
return (await db.execute(q)).scalars().all()
|
|
|
|
|
|
@router.post("/projects/{project_id}/defects", response_model=SiDefectOut, status_code=201)
|
|
async def create_defect(
|
|
project_id: int,
|
|
body: SiDefectCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
await _assert_project(project_id, db)
|
|
|
|
defect = SiDefect(
|
|
defect_id = _new_defect_id(),
|
|
project_id = project_id,
|
|
tc_id = body.tc_id,
|
|
req_id = body.req_id,
|
|
title = body.title,
|
|
description = body.description,
|
|
severity = body.severity,
|
|
phase = body.phase,
|
|
reported_by = body.reported_by or current_user.username,
|
|
assigned_to = body.assigned_to,
|
|
build_found = body.build_found,
|
|
note = body.note,
|
|
)
|
|
db.add(defect)
|
|
await db.commit()
|
|
await db.refresh(defect)
|
|
return defect
|
|
|
|
|
|
@router.get("/projects/{project_id}/defects/{defect_id}", response_model=SiDefectOut)
|
|
async def get_defect(
|
|
project_id: int, defect_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
return await _get_defect_or_404(project_id, defect_id, db)
|
|
|
|
|
|
@router.patch("/projects/{project_id}/defects/{defect_id}", response_model=SiDefectOut)
|
|
async def update_defect(
|
|
project_id: int, defect_id: int,
|
|
body: SiDefectUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
defect = await _get_defect_or_404(project_id, defect_id, db)
|
|
for field, val in body.model_dump(exclude_none=True).items():
|
|
setattr(defect, field, val)
|
|
await db.commit()
|
|
await db.refresh(defect)
|
|
return defect
|
|
|
|
|
|
@router.patch("/projects/{project_id}/defects/{defect_id}/fix", response_model=SiDefectOut)
|
|
async def fix_defect(
|
|
project_id: int, defect_id: int,
|
|
build_fixed: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
defect = await _get_defect_or_404(project_id, defect_id, db)
|
|
defect.status = DefectStatus.FIXED
|
|
defect.fixed_by = current_user.username
|
|
defect.fixed_at = datetime.now()
|
|
if build_fixed:
|
|
defect.build_fixed = build_fixed
|
|
await db.commit()
|
|
await db.refresh(defect)
|
|
return defect
|
|
|
|
|
|
@router.patch("/projects/{project_id}/defects/{defect_id}/verify", response_model=SiDefectOut)
|
|
async def verify_defect(
|
|
project_id: int, defect_id: int,
|
|
is_verified: bool = Query(True),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""결함 검증 — 통과 시 CLOSED, 재현 시 OPEN으로 복귀."""
|
|
defect = await _get_defect_or_404(project_id, defect_id, db)
|
|
if defect.status != DefectStatus.FIXED:
|
|
raise HTTPException(400, "수정 완료 상태의 결함만 검증할 수 있습니다")
|
|
|
|
if is_verified:
|
|
defect.status = DefectStatus.VERIFIED
|
|
defect.verified_by = current_user.username
|
|
defect.verified_at = datetime.now()
|
|
else:
|
|
defect.status = DefectStatus.OPEN
|
|
|
|
await db.commit()
|
|
await db.refresh(defect)
|
|
return defect
|
|
|
|
|
|
@router.get("/projects/{project_id}/defects/stats")
|
|
async def get_defect_stats(
|
|
project_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
await _assert_project(project_id, db)
|
|
defects = (await db.execute(
|
|
select(SiDefect).where(SiDefect.project_id == project_id)
|
|
)).scalars().all()
|
|
|
|
by_severity = {}
|
|
by_status = {}
|
|
by_phase = {}
|
|
|
|
for d in defects:
|
|
by_severity[d.severity] = by_severity.get(d.severity, 0) + 1
|
|
by_status[d.status] = by_status.get(d.status, 0) + 1
|
|
if d.phase:
|
|
by_phase[d.phase] = by_phase.get(d.phase, 0) + 1
|
|
|
|
open_count = sum(1 for d in defects if d.status in ("OPEN", "ASSIGNED"))
|
|
critical_count = sum(1 for d in defects if d.severity == "CRITICAL"
|
|
and d.status not in ("CLOSED", "VERIFIED"))
|
|
|
|
return {
|
|
"project_id": project_id,
|
|
"total": len(defects),
|
|
"open": open_count,
|
|
"critical_open": critical_count,
|
|
"by_severity": by_severity,
|
|
"by_status": by_status,
|
|
"by_phase": by_phase,
|
|
}
|
|
|
|
|
|
# ── 내부 헬퍼 ─────────────────────────────────────────────────────────────────
|
|
|
|
async def _assert_project(project_id: int, db: AsyncSession) -> None:
|
|
if not (await db.execute(
|
|
select(SiProject).where(SiProject.id == project_id)
|
|
)).scalars().first():
|
|
raise HTTPException(404, f"SI 프로젝트 {project_id}를 찾을 수 없습니다")
|
|
|
|
|
|
async def _get_plan_or_404(project_id: int, plan_id: int, db: AsyncSession) -> SiTestPlan:
|
|
plan = (await db.execute(
|
|
select(SiTestPlan).where(
|
|
SiTestPlan.id == plan_id,
|
|
SiTestPlan.project_id == project_id,
|
|
)
|
|
)).scalars().first()
|
|
if not plan:
|
|
raise HTTPException(404, "테스트 계획을 찾을 수 없습니다")
|
|
return plan
|
|
|
|
|
|
async def _get_tc_or_404(plan_id: int, tc_id: int, db: AsyncSession) -> SiTestCase:
|
|
tc = (await db.execute(
|
|
select(SiTestCase).where(
|
|
SiTestCase.id == tc_id,
|
|
SiTestCase.plan_id == plan_id,
|
|
)
|
|
)).scalars().first()
|
|
if not tc:
|
|
raise HTTPException(404, "테스트 케이스를 찾을 수 없습니다")
|
|
return tc
|
|
|
|
|
|
async def _get_defect_or_404(project_id: int, defect_id: int, db: AsyncSession) -> SiDefect:
|
|
defect = (await db.execute(
|
|
select(SiDefect).where(
|
|
SiDefect.id == defect_id,
|
|
SiDefect.project_id == project_id,
|
|
)
|
|
)).scalars().first()
|
|
if not defect:
|
|
raise HTTPException(404, "결함을 찾을 수 없습니다")
|
|
return defect
|
|
|
|
|
|
async def _recalc_plan_stats(plan: SiTestPlan, db: AsyncSession) -> None:
|
|
"""테스트 계획 pass/fail/blocked/pass_rate 재계산."""
|
|
tcs = (await db.execute(
|
|
select(SiTestCase).where(SiTestCase.plan_id == plan.id)
|
|
)).scalars().all()
|
|
|
|
plan.total_cases = len(tcs)
|
|
plan.pass_count = sum(1 for t in tcs if t.last_result == SiTestResult.PASS)
|
|
plan.fail_count = sum(1 for t in tcs if t.last_result == SiTestResult.FAIL)
|
|
plan.blocked_count = sum(1 for t in tcs if t.last_result == SiTestResult.BLOCKED)
|
|
plan.pass_rate = (plan.pass_count * 100 // plan.total_cases) if plan.total_cases else 0
|