pulp-console/app/demo.py

"""임시 데모용 가짜 Pulp 클라이언트 — 실제 Pulp 없이 화면 확인.

PULP_DEMO=true 일 때만 사용. 검색·정렬·페이지네이션을 눈으로 보도록 저장소를 여러 개 생성한다.
정식 기능과 무관하므로 실제 Pulp 연동 시 제거 가능.
"""

from __future__ import annotations

from typing import Any

_NAMES = [
    "rocky9-baseos",
    "rocky9-appstream",
    "rocky9-extras",
    "rocky8-baseos",
    "rocky8-appstream",
    "almalinux9-baseos",
    "almalinux9-appstream",
    "ubuntu2204-main",
    "ubuntu2204-updates",
    "ubuntu2204-security",
    "ubuntu2004-main",
    "debian12-main",
    "debian12-updates",
    "epel9",
    "epel8",
    "centos-stream9",
    "docker-ce",
    "kubernetes",
    "postgresql16",
    "nginx-stable",
    "grafana",
    "elastic-8",
    "hashicorp",
    "internal-tools",
    "internal-agents",
    "monitoring-exporters",
]

# 검증 상태를 섞는다: pass / warn(부분) / unset
_CFGS = [
    {"gpgcheck": 1, "repo-gpgcheck": 1},
    {"gpgcheck": 1, "repo-gpgcheck": 0},
    {},
]


def _uuid(i: int) -> str:
    return f"{i:04d}"


def _latest(i: int) -> int:
    return (i % 5) + 1  # 1~5


_REPOS: list[dict[str, Any]] = [
    {
        "pulp_href": f"/pulp/api/v3/repositories/rpm/rpm/{_uuid(i)}/",
        "name": name,
        "latest_version_href": (
            f"/pulp/api/v3/repositories/rpm/rpm/{_uuid(i)}/versions/{_latest(i)}/"
        ),
        "repo_config": _CFGS[i % 3],
    }
    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]:
    parts = version_href.rstrip("/").split("/")
    n = int(parts[-1]) if parts[-1].isdigit() else 1
    base = 800 + (int(parts[-3]) if parts[-3].isdigit() else 1) * 53
    present = base + n * 137
    return {
        "pulp_created": f"2026-06-{(n * 5) % 27 + 1:02d}T0{n}:11:00Z",
        "content_summary": {
            "present": {"rpm.package": {"count": present}},
            "added": {"rpm.package": {"count": n * 137}},
            "removed": {"rpm.package": {"count": n * 7}},
        },
    }


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 _synth_version(version_href)

    def get_repo(self, uuid: str) -> dict[str, Any]:
        for r in _REPOS:
            if r["pulp_href"].rstrip("/").endswith(uuid):
                return {**r, "remote": "/pulp/api/v3/remotes/rpm/rpm/demo/"}
        return {"remote": "/pulp/api/v3/remotes/rpm/rpm/demo/"}

    def list_versions(self, uuid: str) -> list[dict[str, Any]]:
        base = f"/pulp/api/v3/repositories/rpm/rpm/{uuid}/versions/"
        latest = 1
        for i, r in enumerate(_NAMES, start=1):
            if _uuid(i) == uuid:
                latest = _latest(i)
                break
        out = []
        for n in range(1, latest + 1):
            href = f"{base}{n}/"
            out.append({"pulp_href": href, **_synth_version(href)})
        return out

    def list_distributions(self) -> list[dict[str, Any]]:
        # publication href 에 uuid-버전 인코딩. _DEPLOYED(가변) 기준 → 배포 시 배지 이동.
        return [
            {
                "pulp_href": f"/pulp/api/v3/distributions/rpm/rpm/{u}/",
                "publication": f"/pulp/api/v3/publications/rpm/rpm/{u}-{v}/",
            }
            for u, v in _DEPLOYED.items()
        ]

    def get_publication(self, publication_href: str) -> dict[str, Any]:
        tail = publication_href.rstrip("/").split("/")[-1]  # "0001-2"
        uuid, number = tail.split("-")
        return {
            "repository_version": (
                f"/pulp/api/v3/repositories/rpm/rpm/{uuid}/versions/{number}/"
            )
        }

    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/"],
        }

    def create_publication(self, version_href: str) -> str:
        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:
        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/"

    def wait_for_task(self, task_href: str, attempts: int = 120, delay: float = 1.0):
        created = []
        if "demo-pub-" in task_href:
            tail = task_href.rstrip("/").split("demo-pub-")[-1]  # "{uuid}-{n}"
            created = [f"/pulp/api/v3/publications/rpm/rpm/{tail}/"]
        return {"state": "completed", "created_resources": created}