pulp-console/app/demo.py

"""임시 데모용 가짜 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]:
        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 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 list_versions(self, uuid: str) -> list[dict[str, Any]]:
        base = f"/pulp/api/v3/repositories/rpm/rpm/{uuid}/versions/"
        return [
            {
                "pulp_href": base + "1/",
                "pulp_created": "2026-05-02T01:00:00Z",
                "content_summary": {
                    "present": {"rpm.package": {"count": 4500}},
                    "added": {"rpm.package": {"count": 4500}},
                    "removed": {},
                },
            },
            {
                "pulp_href": base + "2/",
                "pulp_created": "2026-05-28T04:30:00Z",
                "content_summary": {
                    "present": {"rpm.package": {"count": 4700}},
                    "added": {"rpm.package": {"count": 250}},
                    "removed": {"rpm.package": {"count": 50}},
                },
            },
            {
                "pulp_href": base + "3/",
                "pulp_created": "2026-06-15T02:11:00Z",
                "content_summary": {
                    "present": {"rpm.package": {"count": 4821}},
                    "added": {"rpm.package": {"count": 130}},
                    "removed": {"rpm.package": {"count": 9}},
                },
            },
        ]

    def list_distributions(self) -> list[dict[str, Any]]:
        # 저장소별 현재 배포 버전 (publication href 에 uuid-버전 인코딩)
        deployed = {"0001": 2, "0002": 1, "0003": 3, "0004": 1}
        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 create_publication(self, version_href: str) -> str:
        return "/pulp/api/v3/tasks/demo-pub/"

    def update_distribution(self, dist_href: str, publication_href: str) -> str:
        return "/pulp/api/v3/tasks/demo-dist/"

    def wait_for_task(self, task_href: str, attempts: int = 120, delay: float = 1.0):
        # 데모: 즉시 완료. publication task 면 새 publication 리소스를 돌려준다.
        created = (
            ["/pulp/api/v3/publications/rpm/rpm/demo-new/"]
            if "pub" in task_href
            else []
        )
        return {"state": "completed", "created_resources": created}