69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
"""CI Monitor — Gitea commit status 기반 CI 결과 대기.
|
|
|
|
Woodpecker CI의 결과를 Gitea commit status API로 폴링합니다.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
from integrations.gitea_client import GiteaClient
|
|
|
|
logger = logging.getLogger("variet.ci")
|
|
|
|
|
|
class CIMonitor:
|
|
"""CI 결과 모니터링."""
|
|
|
|
def __init__(self, gitea: GiteaClient = None):
|
|
self.gitea = gitea or GiteaClient()
|
|
|
|
async def wait_for_ci(
|
|
self,
|
|
sha: str,
|
|
timeout: int = 300,
|
|
poll_interval: int = 10,
|
|
) -> dict:
|
|
"""CI 결과 대기.
|
|
|
|
Args:
|
|
sha: 커밋 SHA
|
|
timeout: 최대 대기 시간 (초)
|
|
poll_interval: 폴링 간격 (초)
|
|
|
|
Returns:
|
|
{"status": "success|failure|pending", "details": [...]}
|
|
"""
|
|
elapsed = 0
|
|
logger.info(f"CI 결과 대기 중... (SHA: {sha[:7]}, timeout: {timeout}s)")
|
|
|
|
while elapsed < timeout:
|
|
try:
|
|
combined = await self.gitea.get_combined_status(sha)
|
|
state = combined.get("state", "pending")
|
|
statuses = combined.get("statuses", []) or []
|
|
|
|
if state in ("success", "failure", "error"):
|
|
logger.info(f"CI 완료: {state} ({len(statuses)}개 체크)")
|
|
return {
|
|
"status": state,
|
|
"details": [
|
|
{
|
|
"context": s.get("context", ""),
|
|
"status": s.get("status", ""),
|
|
"description": s.get("description", ""),
|
|
}
|
|
for s in statuses
|
|
],
|
|
}
|
|
|
|
logger.debug(f"CI 진행 중... ({elapsed}s/{timeout}s)")
|
|
|
|
except Exception as e:
|
|
logger.warning(f"CI 상태 조회 실패: {e}")
|
|
|
|
await asyncio.sleep(poll_interval)
|
|
elapsed += poll_interval
|
|
|
|
logger.warning(f"CI 타임아웃 ({timeout}s)")
|
|
return {"status": "timeout", "details": []}
|