pulp-console/tests/test_deploy.py

import json

from fastapi.testclient import TestClient

from app.config import get_settings
from app.main import app, get_pulp_client

client = TestClient(app)

VH = "/pulp/api/v3/repositories/rpm/rpm/abc/versions/3/"
REPO = {"name": "rocky9-baseos", "repo_config": {"gpgcheck": 1, "repo-gpgcheck": 1}}
VERSIONS = [
    {
        "pulp_href": "/pulp/api/v3/repositories/rpm/rpm/abc/versions/2/",
        "content_summary": {"present": {"rpm.package": {"count": 4700}}},
    },
    {
        "pulp_href": VH,
        "content_summary": {"present": {"rpm.package": {"count": 4821}}},
    },
]
DISTRIBUTIONS = [
    {
        "pulp_href": "/pulp/api/v3/distributions/rpm/rpm/dabc/",
        "publication": "/pulp/api/v3/publications/rpm/rpm/abc-2/",
    }
]
PUBLICATIONS = {
    "/pulp/api/v3/publications/rpm/rpm/abc-2/": {
        "repository_version": "/pulp/api/v3/repositories/rpm/rpm/abc/versions/2/"
    }
}


class _FakePulp:
    def __init__(
        self,
        repo=None,
        distributions=None,
        pub_state="completed",
        dist_state="completed",
        pub_created=True,
    ):
        self._repo = REPO if repo is None else repo
        self._distributions = DISTRIBUTIONS if distributions is None else distributions
        self._pub_state = pub_state
        self._dist_state = dist_state
        self._pub_created = pub_created
        self.created_with = None
        self.dist_args = None

    def get_repo(self, uuid):
        return self._repo

    def list_versions(self, uuid):
        return VERSIONS

    def list_distributions(self):
        return self._distributions

    def get_publication(self, href):
        return PUBLICATIONS.get(href, {})

    def create_publication(self, version_href):
        self.created_with = version_href
        return "/pulp/api/v3/tasks/pub/"

    def update_distribution(self, dist_href, publication_href):
        self.dist_args = (dist_href, publication_href)
        return "/pulp/api/v3/tasks/dist/"

    def wait_for_task(self, task_href, attempts=120, delay=1.0):
        if "pub" in task_href:
            created = (
                ["/pulp/api/v3/publications/rpm/rpm/new/"] if self._pub_created else []
            )
            return {"state": self._pub_state, "created_resources": created}
        return {"state": self._dist_state, "created_resources": []}


def _use(fake):
    app.dependency_overrides[get_pulp_client] = lambda: fake


def teardown_function():
    app.dependency_overrides.clear()


def _post(confirm_name):
    return client.post(
        "/repos/abc/deploy",
        data={"version_href": VH, "confirm_name": confirm_name},
    )


def test_confirm_modal_shows_preview_and_type_to_confirm():
    _use(_FakePulp())
    resp = client.get("/repos/abc/deploy/confirm", params={"version_href": VH})
    assert resp.status_code == 200
    body = resp.text
    assert "운영망에 배포" in body
    assert "rocky9-baseos" in body  # type-to-confirm 대상 이름
    assert "배포 확정" in body
    assert "+121" in body  # 4821 - 4700 미리보기


def test_deploy_blocked_when_not_verified():
    _use(_FakePulp(repo={"name": "r", "repo_config": {"gpgcheck": 0}}))
    resp = client.post(
        "/repos/abc/deploy", data={"version_href": VH, "confirm_name": "r"}
    )
    assert resp.status_code == 403
    assert "검증" in resp.text


def test_deploy_rejects_name_mismatch():
    _use(_FakePulp())
    resp = _post("wrong-name")
    assert resp.status_code == 400
    assert "이름이 일치" in resp.text


def test_deploy_requires_existing_distribution():
    _use(_FakePulp(distributions=[]))
    resp = _post("rocky9-baseos")
    assert resp.status_code == 404
    assert "distribution" in resp.text


def test_deploy_publication_failure_keeps_prod_unchanged():
    _use(_FakePulp(pub_state="failed"))
    resp = _post("rocky9-baseos")
    assert resp.status_code == 502
    assert "운영망은 변경되지 않았습니다" in resp.text


def test_deploy_distribution_failure_reported():
    _use(_FakePulp(dist_state="failed"))
    resp = _post("rocky9-baseos")
    assert resp.status_code == 502
    assert "운영망 변경에 실패" in resp.text


def test_deploy_success_writes_audit_and_refreshes_list(tmp_path, monkeypatch):
    monkeypatch.setenv("AUDIT_LOG_PATH", str(tmp_path / "audit.jsonl"))
    get_settings.cache_clear()

    fake = _FakePulp()
    _use(fake)
    resp = _post("rocky9-baseos")
    assert resp.status_code == 200
    assert "배포 완료" in resp.text
    assert 'hx-swap-oob="true"' in resp.text  # 버전 목록 OOB 갱신

    # publication 은 대상 버전으로, distribution 은 새 publication 으로 교체
    assert fake.created_with == VH
    assert fake.dist_args == (
        "/pulp/api/v3/distributions/rpm/rpm/dabc/",
        "/pulp/api/v3/publications/rpm/rpm/new/",
    )

    # 감사 로그: from=2, to=3, repo 이름 기록
    entry = json.loads((tmp_path / "audit.jsonl").read_text(encoding="utf-8").strip())
    assert entry["action"] == "deploy"
    assert entry["from_version"] == 2
    assert entry["to_version"] == 3
    assert entry["repo_name"] == "rocky9-baseos"