@@ -26,9 +26,14 @@
| {% if v.gpg %}{{ v.gpg.label }}{% endif %} |
{% if v.is_deployed %}운영 배포 중{% endif %} |
- {% if not v.is_deployed %}
-
-
+ {% if v.is_deployed %}
+ {# 배포 상태 칸에 배지 표시 #}
+ {% elif v.deployable %}
+
+ {% else %}
+
{% endif %}
|
diff --git a/app/templates/versions.html b/app/templates/versions.html
index 81c18e1..3911627 100644
--- a/app/templates/versions.html
+++ b/app/templates/versions.html
@@ -5,6 +5,7 @@
{{ repo_name }} · 버전
+
{% if error %}
버전 목록을 불러오지 못했습니다: {{ error }}
@@ -21,4 +22,7 @@
{% include "partials/version_list.html" %}
{% endif %}
+
+
+
{% endblock %}
diff --git a/app/views.py b/app/views.py
index 706e80c..e6ed643 100644
--- a/app/views.py
+++ b/app/views.py
@@ -85,14 +85,44 @@ def build_version_view(
호출부에서 repo 의 gpg_status() 를 주입한다.
"""
number = version_number(version.get("pulp_href"))
+ is_deployed = number is not None and number == deployed_number
+ # 배포 가능: 검증 통과(pass) + 현재 배포 중이 아님 (스펙 §5-2 게이트)
+ deployable = (gpg or {}).get("level") == "pass" and not is_deployed
return {
"number": number,
+ "href": version.get("pulp_href"),
"created": version.get("pulp_created"),
"package_count": _content_count(version, "present"),
"added": _content_count(version, "added"),
"removed": _content_count(version, "removed"),
"gpg": gpg,
- "is_deployed": number is not None and number == deployed_number,
+ "is_deployed": is_deployed,
+ "deployable": deployable,
+ }
+
+
+def deploy_preview(
+ version_views: list[dict[str, Any]],
+ target_number: int | None,
+ deployed_number: int | None,
+) -> dict[str, Any]:
+ """배포 확인 모달용 미리보기: 현재 운영 버전 → 대상 버전, 순 패키지 변화."""
+
+ def find(n: int | None) -> dict[str, Any] | None:
+ return next((v for v in version_views if v["number"] == n), None)
+
+ current = find(deployed_number)
+ target = find(target_number)
+ current_count = current["package_count"] if current else 0
+ target_count = target["package_count"] if target else 0
+ net = target_count - current_count
+ return {
+ "current_version": deployed_number,
+ "current_count": current_count,
+ "target_version": target_number,
+ "target_count": target_count,
+ "net": abs(net),
+ "net_sign": "+" if net >= 0 else "−",
}
diff --git a/tests/test_deploy.py b/tests/test_deploy.py
new file mode 100644
index 0000000..cbe082b
--- /dev/null
+++ b/tests/test_deploy.py
@@ -0,0 +1,167 @@
+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"
diff --git a/tests/test_pulp_client.py b/tests/test_pulp_client.py
index b04650e..05596de 100644
--- a/tests/test_pulp_client.py
+++ b/tests/test_pulp_client.py
@@ -61,3 +61,43 @@ def test_get_task_returns_state():
with PulpClient() as pulp:
task = pulp.get_task("/pulp/api/v3/tasks/t1/")
assert task["state"] == "running"
+
+
+@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"
diff --git a/tests/test_views.py b/tests/test_views.py
index a522eed..a41dfca 100644
--- a/tests/test_views.py
+++ b/tests/test_views.py
@@ -135,3 +135,35 @@ def test_verification_detail():
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_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"] == "+"