pulp-console/tests/test_views.py

from app import views


def test_gpg_status_pass():
    repo = {"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 1}}
    assert views.gpg_status(repo)["level"] == "pass"


def test_gpg_status_warn_when_partial():
    repo = {"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 0}}
    assert views.gpg_status(repo)["level"] == "warn"


def test_gpg_status_unset_when_no_config():
    assert views.gpg_status({})["level"] == "unset"
    assert views.gpg_status({"repo_config": {}})["level"] == "unset"


def test_version_number_parses_trailing_int():
    assert views.version_number("/pulp/api/v3/repositories/rpm/rpm/x/versions/7/") == 7
    assert views.version_number(None) is None
    assert views.version_number("/no/number/here/") is None


def test_package_count_sums_present_counts():
    version = {
        "content_summary": {
            "present": {
                "rpm.package": {"count": 1500},
                "rpm.advisory": {"count": 42},
            }
        }
    }
    assert views.package_count(version) == 1542
    assert views.package_count(None) is None
    assert views.package_count({}) == 0


def test_build_repo_view_combines_repo_and_version():
    repo = {
        "pulp_href": "/pulp/api/v3/repositories/rpm/rpm/abc-123/",
        "name": "rocky9-baseos",
        "latest_version_href": "/pulp/api/v3/repositories/rpm/rpm/abc-123/versions/2/",
        "repo_config": {"gpgcheck": 1, "repo-gpgcheck": 1},
    }
    version = {
        "pulp_created": "2026-06-01T00:00:00Z",
        "content_summary": {"present": {"rpm.package": {"count": 1500}}},
    }
    view = views.build_repo_view(repo, version)
    assert view["name"] == "rocky9-baseos"
    assert view["uuid"] == "abc-123"
    assert view["latest_version"] == 2
    assert view["package_count"] == 1500
    assert view["last_sync"] == "2026-06-01T00:00:00Z"
    assert view["gpg"]["level"] == "pass"
    assert view["deployed_version"] is None  # 단계1 미구현


def test_summarize_counts():
    repo_views = [
        {"gpg": {"level": "pass"}, "package_count": 100},
        {"gpg": {"level": "warn"}, "package_count": 50},
        {"gpg": {"level": "pass"}, "package_count": None},
    ]
    summary = views.summarize(repo_views)
    assert summary == {
        "total": 3,
        "syncing": 0,
        "gpg_pass": 2,
        "total_packages": 150,
    }


def test_summarize_includes_syncing_count():
    assert views.summarize([], syncing=2)["syncing"] == 2


def test_task_progress_running_sums_reports():
    task = {
        "state": "running",
        "progress_reports": [{"done": 5, "total": 10}, {"done": 3, "total": 10}],
    }
    p = views.task_progress(task)
    assert p["running"] is True
    assert p["completed"] is False
    assert (p["done"], p["total"], p["percent"]) == (8, 20, 40)


def test_task_progress_completed_extracts_version():
    task = {
        "state": "completed",
        "created_resources": ["/pulp/api/v3/repositories/rpm/rpm/x/versions/4/"],
    }
    p = views.task_progress(task)
    assert p["completed"] is True
    assert p["new_version"] == 4


def test_task_progress_failed_surfaces_error():
    p = views.task_progress({"state": "failed", "error": {"description": "boom"}})
    assert p["failed"] is True
    assert p["error"] == "boom"


def test_build_version_view_with_deltas_and_deploy_flag():
    version = {
        "pulp_href": "/pulp/api/v3/repositories/rpm/rpm/x/versions/3/",
        "pulp_created": "2026-06-01T00:00:00Z",
        "content_summary": {
            "present": {"rpm.package": {"count": 4821}},
            "added": {"rpm.package": {"count": 130}},
            "removed": {"rpm.package": {"count": 9}},
        },
    }
    gpg = {"level": "pass", "label": "검증됨"}
    v = views.build_version_view(version, deployed_number=3, gpg=gpg)
    assert v["number"] == 3
    assert v["package_count"] == 4821
    assert (v["added"], v["removed"]) == (130, 9)
    assert v["is_deployed"] is True
    assert v["gpg"]["level"] == "pass"


def test_build_version_view_not_deployed():
    version = {"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/x/versions/2/"}
    v = views.build_version_view(version, deployed_number=3)
    assert v["is_deployed"] is False
    assert v["added"] == 0 and v["removed"] == 0


def test_verification_detail():
    repo = {"repo_config": {"gpgcheck": 1, "repo-gpgcheck": 0}}
    d = views.verification_detail(repo)
    assert d["gpgcheck"] == 1
    assert d["repo_gpgcheck"] == 0
    assert d["status"]["level"] == "warn"


def test_build_version_view_deployable_only_when_verified_and_not_deployed():
    version = {"pulp_href": "/pulp/api/v3/repositories/rpm/rpm/x/versions/3/"}
    verified = {"level": "pass", "label": "검증됨"}
    unverified = {"level": "unset", "label": "미설정"}
    # 검증 통과 + 미배포 → 배포 가능
    assert views.build_version_view(version, 2, verified)["deployable"] is True
    # 검증 통과 but 현재 배포 중 → 불가
    assert views.build_version_view(version, 3, verified)["deployable"] is False
    # 미검증 → 불가
    assert views.build_version_view(version, 2, unverified)["deployable"] is False


def test_deploy_preview_net_change():
    vv = [
        {"number": 3, "package_count": 4821},
        {"number": 2, "package_count": 4700},
    ]
    p = views.deploy_preview(vv, target_number=3, deployed_number=2)
    assert (p["current_version"], p["current_count"]) == (2, 4700)
    assert (p["target_version"], p["target_count"]) == (3, 4821)
    assert p["net"] == 121
    assert p["net_sign"] == "+"


def test_sort_repos_by_name_packages_verified():
    rv = [
        {
            "name": "bravo",
            "gpg": {"level": "unset"},
            "latest_version": 1,
            "package_count": 50,
            "last_sync": "",
        },
        {
            "name": "alpha",
            "gpg": {"level": "pass"},
            "latest_version": 3,
            "package_count": 200,
            "last_sync": "",
        },
    ]
    assert [r["name"] for r in views.sort_repos(rv, "name", "asc")] == [
        "alpha",
        "bravo",
    ]
    assert [r["name"] for r in views.sort_repos(rv, "name", "desc")] == [
        "bravo",
        "alpha",
    ]
    assert [r["package_count"] for r in views.sort_repos(rv, "packages", "desc")] == [
        200,
        50,
    ]
    # verified: pass 가 unset 보다 앞
    assert views.sort_repos(rv, "verified", "asc")[0]["gpg"]["level"] == "pass"


def test_deploy_preview_handles_no_current_deploy():
    vv = [{"number": 1, "package_count": 50}]
    p = views.deploy_preview(vv, target_number=1, deployed_number=None)
    assert p["current_count"] == 0
    assert p["net"] == 50
    assert p["net_sign"] == "+"