feat(demo): reflect deploys in demo state so the badge moves

- _DEPLOYED 가변 맵 + create_publication/update_distribution/wait_for_task 가
  배포한 버전을 기록 → 데모에서도 '운영 배포 중' 배지가 방금 버전으로 이동

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-17 16:52:43 +09:00
parent 84fb568e96
commit 59134927d8

View File

@@ -65,6 +65,11 @@ _REPOS: list[dict[str, Any]] = [
for i, name in enumerate(_NAMES, start=1) for i, name in enumerate(_NAMES, start=1)
] ]
# 저장소별 현재 배포 버전 (가변) — 배포 시 갱신해 데모에서도 배지가 이동하게.
_DEPLOYED: dict[str, int] = {
_uuid(i): max(1, _latest(i) - 1) for i in range(1, len(_NAMES) + 1)
}
def _synth_version(version_href: str) -> dict[str, Any]: def _synth_version(version_href: str) -> dict[str, Any]:
parts = version_href.rstrip("/").split("/") parts = version_href.rstrip("/").split("/")
@@ -114,18 +119,14 @@ class DemoPulpClient:
return out return out
def list_distributions(self) -> list[dict[str, Any]]: def list_distributions(self) -> list[dict[str, Any]]:
# 저장소별 현재 배포 버전 (보통 최신-1). publication href 에 uuid-버전 인코딩 # publication href 에 uuid-버전 인코딩. _DEPLOYED(가변) 기준 → 배포 시 배지 이동.
dists = [] return [
for i in range(1, len(_NAMES) + 1):
deployed = max(1, _latest(i) - 1)
u = _uuid(i)
dists.append(
{ {
"pulp_href": f"/pulp/api/v3/distributions/rpm/rpm/{u}/", "pulp_href": f"/pulp/api/v3/distributions/rpm/rpm/{u}/",
"publication": f"/pulp/api/v3/publications/rpm/rpm/{u}-{deployed}/", "publication": f"/pulp/api/v3/publications/rpm/rpm/{u}-{v}/",
} }
) for u, v in _DEPLOYED.items()
return dists ]
def get_publication(self, publication_href: str) -> dict[str, Any]: def get_publication(self, publication_href: str) -> dict[str, Any]:
tail = publication_href.rstrip("/").split("/")[-1] # "0001-2" tail = publication_href.rstrip("/").split("/")[-1] # "0001-2"
@@ -147,15 +148,22 @@ class DemoPulpClient:
} }
def create_publication(self, version_href: str) -> str: def create_publication(self, version_href: str) -> str:
return "/pulp/api/v3/tasks/demo-pub/" parts = version_href.rstrip("/").split("/")
uuid, number = parts[-3], parts[-1] # .../{uuid}/versions/{n}/
return f"/pulp/api/v3/tasks/demo-pub-{uuid}-{number}/"
def update_distribution(self, dist_href: str, publication_href: str) -> str: def update_distribution(self, dist_href: str, publication_href: str) -> str:
tail = publication_href.rstrip("/").split("/")[-1] # "{uuid}-{n}"
try:
uuid, number = tail.split("-")
_DEPLOYED[uuid] = int(number) # 데모 상태 갱신 → 배지 이동
except ValueError:
pass
return "/pulp/api/v3/tasks/demo-dist/" return "/pulp/api/v3/tasks/demo-dist/"
def wait_for_task(self, task_href: str, attempts: int = 120, delay: float = 1.0): def wait_for_task(self, task_href: str, attempts: int = 120, delay: float = 1.0):
created = ( created = []
["/pulp/api/v3/publications/rpm/rpm/demo-new/"] if "demo-pub-" in task_href:
if "pub" in task_href tail = task_href.rstrip("/").split("demo-pub-")[-1] # "{uuid}-{n}"
else [] created = [f"/pulp/api/v3/publications/rpm/rpm/{tail}/"]
)
return {"state": "completed", "created_resources": created} return {"state": "completed", "created_resources": created}