- 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>
81 lines
3.0 KiB
Python
81 lines
3.0 KiB
Python
"""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),
|
|
}
|