Hyemin Lee
abf8654b52
- list_repos: Pulp next 페이지 끝까지 순회 (첫-페이지-only 누락 수정) - BFF 메모리 검색(이름)·정렬(이름/검증/버전/패키지/동기화)·페이지네이션(20/페이지) - GET /repos/table 조각: 검색창은 표 밖(포커스 유지), 정렬헤더/페이저는 표 안, hx-include 로 검색어·정렬상태 상호 전송 - 폰트 재조정: 데이터 셀 14px, 헤더 13px, 배지/액션버튼 14px (동기화 버튼 대비 균형) - 버전 페이지 제목↔검증상세 마진, 데모 저장소 26개로 확장 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from app.config import get_settings, tls_verify
|
|
from app.pulp_client import PulpClient
|
|
from tests.conftest import PULP_TEST_URL
|
|
|
|
|
|
def test_tls_verify_prefers_ca_file(monkeypatch):
|
|
monkeypatch.setenv("PULP_CA_FILE", "/etc/ssl/boknet-ca.pem")
|
|
monkeypatch.setenv("PULP_VERIFY_TLS", "false")
|
|
get_settings.cache_clear()
|
|
assert tls_verify(get_settings()) == "/etc/ssl/boknet-ca.pem"
|
|
|
|
|
|
def test_tls_verify_falls_back_to_bool():
|
|
# conftest: CA 비어있음, VERIFY_TLS=false
|
|
assert tls_verify(get_settings()) is False
|
|
|
|
|
|
@respx.mock
|
|
def test_status_returns_json():
|
|
route = respx.get(f"{PULP_TEST_URL}/pulp/api/v3/status/").mock(
|
|
return_value=httpx.Response(200, json={"online": True, "versions": []})
|
|
)
|
|
with PulpClient() as pulp:
|
|
data = pulp.status()
|
|
assert route.called
|
|
assert data == {"online": True, "versions": []}
|
|
|
|
|
|
@respx.mock
|
|
def test_get_raises_on_http_error():
|
|
respx.get(f"{PULP_TEST_URL}/pulp/api/v3/status/").mock(
|
|
return_value=httpx.Response(503)
|
|
)
|
|
with PulpClient() as pulp, pytest.raises(httpx.HTTPStatusError):
|
|
pulp.status()
|
|
|
|
|
|
@respx.mock
|
|
def test_sync_repo_posts_remote_and_returns_task():
|
|
import json
|
|
|
|
route = respx.post(
|
|
f"{PULP_TEST_URL}/pulp/api/v3/repositories/rpm/rpm/abc/sync/"
|
|
).mock(return_value=httpx.Response(202, json={"task": "/pulp/api/v3/tasks/t1/"}))
|
|
with PulpClient() as pulp:
|
|
href = pulp.sync_repo("abc", "/pulp/api/v3/remotes/rpm/rpm/r1/")
|
|
assert href == "/pulp/api/v3/tasks/t1/"
|
|
body = json.loads(route.calls.last.request.content)
|
|
assert body == {"remote": "/pulp/api/v3/remotes/rpm/rpm/r1/"}
|
|
|
|
|
|
@respx.mock
|
|
def test_get_task_returns_state():
|
|
respx.get(f"{PULP_TEST_URL}/pulp/api/v3/tasks/t1/").mock(
|
|
return_value=httpx.Response(200, json={"state": "running"})
|
|
)
|
|
with PulpClient() as pulp:
|
|
task = pulp.get_task("/pulp/api/v3/tasks/t1/")
|
|
assert task["state"] == "running"
|
|
|
|
|
|
@respx.mock
|
|
def test_list_repos_follows_pagination():
|
|
first = f"{PULP_TEST_URL}/pulp/api/v3/repositories/rpm/rpm/?limit=100"
|
|
second = f"{PULP_TEST_URL}/pulp/api/v3/repositories/rpm/rpm/?limit=100&offset=100"
|
|
respx.get(first).mock(
|
|
return_value=httpx.Response(
|
|
200, json={"results": [{"name": "a"}], "next": second}
|
|
)
|
|
)
|
|
respx.get(second).mock(
|
|
return_value=httpx.Response(
|
|
200, json={"results": [{"name": "b"}], "next": None}
|
|
)
|
|
)
|
|
with PulpClient() as pulp:
|
|
repos = pulp.list_repos()
|
|
assert [r["name"] for r in repos] == ["a", "b"]
|
|
|
|
|
|
@respx.mock
|
|
def test_create_publication_posts_version_and_returns_task():
|
|
import json
|
|
|
|
route = respx.post(f"{PULP_TEST_URL}/pulp/api/v3/publications/rpm/rpm/").mock(
|
|
return_value=httpx.Response(202, json={"task": "/pulp/api/v3/tasks/p1/"})
|
|
)
|
|
vh = "/pulp/api/v3/repositories/rpm/rpm/abc/versions/3/"
|
|
with PulpClient() as pulp:
|
|
href = pulp.create_publication(vh)
|
|
assert href == "/pulp/api/v3/tasks/p1/"
|
|
assert json.loads(route.calls.last.request.content) == {"repository_version": vh}
|
|
|
|
|
|
@respx.mock
|
|
def test_update_distribution_patches_publication_and_returns_task():
|
|
import json
|
|
|
|
dist = "/pulp/api/v3/distributions/rpm/rpm/d1/"
|
|
route = respx.patch(f"{PULP_TEST_URL}{dist}").mock(
|
|
return_value=httpx.Response(202, json={"task": "/pulp/api/v3/tasks/d1/"})
|
|
)
|
|
with PulpClient() as pulp:
|
|
href = pulp.update_distribution(dist, "/pulp/api/v3/publications/rpm/rpm/p1/")
|
|
assert href == "/pulp/api/v3/tasks/d1/"
|
|
assert json.loads(route.calls.last.request.content) == {
|
|
"publication": "/pulp/api/v3/publications/rpm/rpm/p1/"
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_wait_for_task_returns_on_terminal_state():
|
|
respx.get(f"{PULP_TEST_URL}/pulp/api/v3/tasks/t1/").mock(
|
|
return_value=httpx.Response(200, json={"state": "completed"})
|
|
)
|
|
with PulpClient() as pulp:
|
|
task = pulp.wait_for_task("/pulp/api/v3/tasks/t1/", attempts=1, delay=0)
|
|
assert task["state"] == "completed"
|