feat: FastAPI console with dashboard, Pulp client, healthz (stages 0-1)

- config: PULP_* env + 사내 CA(verify=) 대응
- pulp_client: Basic Auth get/post/patch, status/list_repos/get_version
- BFF 라우트: GET / (대시보드), GET /repos (조각), GET /healthz
- views: Pulp JSON → 표시 모델 (GPG 검증 배지 3단계, 요약 집계)
- HTMX + Jinja 템플릿, vendored pico.css/htmx.js (CDN 의존 0)
- 브라우저는 Pulp 직접 호출 안 함 — 모두 BFF 경유 (스펙 §2)
- 테스트: pulp_client(respx), 라우트(dependency_overrides), views 순수함수

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-17 10:55:01 +09:00
parent a91b9d0245
commit 3622b4f20a
21 changed files with 671 additions and 0 deletions

80
app/views.py Normal file
View File

@@ -0,0 +1,80 @@
"""Pulp JSON → 화면용 표시 모델 변환 (순수 함수, httpx 의존 없음).
GPG 검증 배지 로직(✅/⚠️/⚪)은 화면 4 에서도 재사용하므로 여기 함수로 둔다(스펙 §6 화면4).
"""
from __future__ import annotations
from typing import Any
def gpg_status(repo: dict[str, Any]) -> dict[str, str]:
"""repo_config 의 gpgcheck / repo-gpgcheck 로 3단계 검증 배지 산출.
- pass(✅): 패키지 서명 + repo 메타데이터 서명 모두 검증
- warn(⚠️): 둘 중 하나만 검증
- unset(⚪): 검증 미설정
"""
cfg = repo.get("repo_config") or {}
pkg = bool(cfg.get("gpgcheck", 0))
meta = bool(cfg.get("repo-gpgcheck", 0))
# level 은 색상 칩(badge-{level})으로 렌더 — 장식 이모지는 쓰지 않는다(design-ref §7).
if pkg and meta:
return {"level": "pass", "label": "검증됨"}
if pkg or meta:
return {"level": "warn", "label": "부분 검증"}
return {"level": "unset", "label": "미설정"}
def version_number(version_href: str | None) -> int | None:
"""latest_version_href 끝의 정수(vN) 추출. 예: .../versions/3/ → 3"""
if not version_href:
return None
parts = [p for p in version_href.split("/") if p]
try:
return int(parts[-1])
except (ValueError, IndexError):
return None
def package_count(version: dict[str, Any] | None) -> int | None:
"""RepositoryVersion 의 content_summary.present 카운트 합계."""
if version is None:
return None
present = (version.get("content_summary") or {}).get("present") or {}
return sum((info or {}).get("count", 0) for info in present.values())
def _uuid_from_href(href: str) -> str:
parts = [p for p in href.split("/") if p]
return parts[-1] if parts else ""
def build_repo_view(
repo: dict[str, Any], version: dict[str, Any] | None = None
) -> dict[str, Any]:
"""저장소 1건 + (선택) 최신 버전 상세 → 대시보드 행 모델."""
return {
"name": repo.get("name", ""),
"uuid": _uuid_from_href(repo.get("pulp_href", "")),
"href": repo.get("pulp_href", ""),
"latest_version": version_number(repo.get("latest_version_href")),
"package_count": package_count(version),
# 최신 스냅샷 생성 시각 ≈ 마지막 동기화 시각
"last_sync": (version or {}).get("pulp_created"),
"gpg": gpg_status(repo),
# TODO(단계3/4): 현재 운영 배포 버전은 Distribution→publication→version 역추적 필요.
# 단계1에선 최신 버전만 표시한다.
"deployed_version": None,
}
def summarize(repo_views: list[dict[str, Any]]) -> dict[str, int]:
"""상단 요약 카드 수치."""
return {
"total": len(repo_views),
# TODO(단계2): 진행 중 sync task 수를 메모리 task 맵에서 집계
"syncing": 0,
"gpg_pass": sum(1 for r in repo_views if r["gpg"]["level"] == "pass"),
"total_packages": sum(r["package_count"] or 0 for r in repo_views),
}