Files
pulp-console/app/demo.py
Hyemin Lee 6c41b89c53 feat: sync execution with 3s progress polling (stage 2)
- pulp_client: get_repo / sync_repo / get_task (비동기 task href 반환)
- POST /repos/{uuid}/sync: remote 확인 후 동기화 시작, 진행률 조각 반환
- GET /repos/{uuid}/progress: progress_reports 합산 → 바/퍼센트,
  완료/실패 시 hx-trigger 제거로 폴링 중단
- task_store: uuid→task_href 메모리 추적 (단일 인스턴스), 대시보드 '동기화 중' 집계
- views.task_progress: state/done/total/percent + created_resources에서 새 vN
- 동기화는 안전 동작이라 확인 모달 없이 실행 (스펙 §5-1)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-17 11:09:37 +09:00

84 lines
3.0 KiB
Python

"""임시 데모용 가짜 Pulp 클라이언트 — 성공 대시보드 미리보기.
PULP_DEMO=true 일 때만 사용된다. 실제 Pulp 서버 없이 화면을 확인하기 위한 것.
정식 기능과 무관하므로 단계 5(또는 실제 Pulp 연동 시) 제거 가능.
"""
from __future__ import annotations
from typing import Any
_REPOS: list[dict[str, Any]] = [
{
"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/0001/",
"name": "rocky9-baseos",
"latest_version_href": "/pulp/api/v3/repositories/rpm/rpm/0001/versions/3/",
"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 1},
},
{
"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/0002/",
"name": "rocky9-appstream",
"latest_version_href": "/pulp/api/v3/repositories/rpm/rpm/0002/versions/2/",
"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 1},
},
{
"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/0003/",
"name": "ubuntu2204-main",
"latest_version_href": "/pulp/api/v3/repositories/rpm/rpm/0003/versions/5/",
"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 0}, # 부분 검증
},
{
"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/0004/",
"name": "internal-tools",
"latest_version_href": "/pulp/api/v3/repositories/rpm/rpm/0004/versions/1/",
"repo_config": {}, # 미설정
},
]
_VERSIONS: dict[str, dict[str, Any]] = {
"/pulp/api/v3/repositories/rpm/rpm/0001/versions/3/": {
"pulp_created": "2026-06-15T02:11:00Z",
"content_summary": {"present": {"rpm.package": {"count": 4821}}},
},
"/pulp/api/v3/repositories/rpm/rpm/0002/versions/2/": {
"pulp_created": "2026-06-14T23:40:00Z",
"content_summary": {"present": {"rpm.package": {"count": 8934}}},
},
"/pulp/api/v3/repositories/rpm/rpm/0003/versions/5/": {
"pulp_created": "2026-06-16T08:05:00Z",
"content_summary": {"present": {"deb.package": {"count": 12500}}},
},
"/pulp/api/v3/repositories/rpm/rpm/0004/versions/1/": {
"pulp_created": "2026-05-30T11:20:00Z",
"content_summary": {"present": {"rpm.package": {"count": 87}}},
},
}
class DemoPulpClient:
def close(self) -> None: # 인터페이스 호환
pass
def status(self) -> dict[str, Any]:
return {"online": True, "demo": True}
def list_repos(self) -> list[dict[str, Any]]:
return _REPOS
def get_version(self, version_href: str) -> dict[str, Any]:
return _VERSIONS.get(version_href, {})
def get_repo(self, uuid: str) -> dict[str, Any]:
return {"remote": "/pulp/api/v3/remotes/rpm/rpm/demo/"}
def sync_repo(self, uuid: str, remote_href: str) -> str:
return f"/pulp/api/v3/tasks/demo-{uuid}/"
def get_task(self, task_href: str) -> dict[str, Any]:
# 데모: 첫 폴링에서 바로 완료 처리
return {
"state": "completed",
"progress_reports": [{"done": 120, "total": 120}],
"created_resources": ["/pulp/api/v3/repositories/rpm/rpm/demo/versions/9/"],
}