Hyemin Lee
7ad6d65a11
- pulp_client: create_publication / update_distribution / wait_for_task
- GET /repos/{uuid}/deploy/confirm: 미리보기(현재→대상, 순 변화) + type-to-confirm 모달
- POST /repos/{uuid}/deploy: 검증 게이트(서버측 재확인) → publication 생성 →
distribution 교체 → 감사 로그. 2단계 실패 시 '운영망 변경 여부' 명확화
- audit.record_deploy: 누가/언제/repo/이전버전→대상버전 JSONL (롤백 추적)
- 배포 버튼은 검증 통과 + 미배포 버전만 활성, 성공 시 버전목록 OOB 갱신
- 인증은 TODO (operator placeholder, 배포 비밀번호 재확인 예정)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""버전(스냅샷) 목록 + 검증 배지 (화면 3 전반).
|
|
|
|
GET /repos/{uuid}/versions → 해당 저장소의 RepositoryVersion 목록 페이지(최신순).
|
|
|
|
현재 운영 배포 버전은 Distribution → publication → repository_version 역추적으로 판별한다
|
|
(스펙 §4). 배포 버튼 활성화/확인은 routes/deploy.py(단계 4).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from .. import views
|
|
from ..deps import get_pulp_client, templates
|
|
from ..pulp_client import PulpClient
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _deployed_version_number(pulp: PulpClient, uuid: str) -> int | None:
|
|
"""이 저장소가 현재 운영 배포 중인 버전 번호. 없거나 조회 실패 시 None.
|
|
|
|
Distribution.publication → Publication.repository_version 이 이 저장소의
|
|
버전을 가리키면 그 번호를 반환.
|
|
"""
|
|
needle = f"/repositories/rpm/rpm/{uuid}/versions/"
|
|
try:
|
|
distributions = pulp.list_distributions()
|
|
except httpx.HTTPError:
|
|
return None
|
|
|
|
for dist in distributions:
|
|
pub_href = dist.get("publication")
|
|
if not pub_href:
|
|
continue # repository 직접 배포(최신 추종)는 단계4 TODO
|
|
try:
|
|
pub = pulp.get_publication(pub_href)
|
|
except httpx.HTTPError:
|
|
continue
|
|
rv = pub.get("repository_version") or ""
|
|
if needle in rv:
|
|
return views.version_number(rv)
|
|
return None
|
|
|
|
|
|
def load_version_context(pulp: PulpClient, uuid: str) -> dict[str, Any]:
|
|
"""버전 페이지/조각 렌더용 컨텍스트. get_repo/list_versions 실패는 httpx 예외로 전파."""
|
|
repo = pulp.get_repo(uuid)
|
|
raw_versions = pulp.list_versions(uuid)
|
|
gpg = views.gpg_status(repo)
|
|
deployed = _deployed_version_number(pulp, uuid)
|
|
version_views = [views.build_version_view(v, deployed, gpg) for v in raw_versions]
|
|
version_views.sort(key=lambda v: v["number"] or 0, reverse=True) # 최신순
|
|
return {
|
|
"repo_name": repo.get("name", uuid),
|
|
"uuid": uuid,
|
|
"versions": version_views,
|
|
"verification": views.verification_detail(repo),
|
|
"deployed": deployed,
|
|
"error": None,
|
|
}
|
|
|
|
|
|
@router.get("/repos/{uuid}/versions", response_class=HTMLResponse)
|
|
def versions(
|
|
request: Request, uuid: str, pulp: PulpClient = Depends(get_pulp_client)
|
|
) -> HTMLResponse:
|
|
try:
|
|
ctx = load_version_context(pulp, uuid)
|
|
except httpx.HTTPError as exc:
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"versions.html",
|
|
{"repo_name": uuid, "uuid": uuid, "error": str(exc)},
|
|
)
|
|
return templates.TemplateResponse(request, "versions.html", ctx)
|