Coverage for tests / scripts / test_ghcr_prune_untagged.py: 99%
117 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-04-03 18:53 +0000
1"""Tests for the GHCR prune untagged utility script."""
3from __future__ import annotations
5from collections.abc import Mapping
6from typing import TYPE_CHECKING, Any, cast
8import pytest
9from assertpy import assert_that
11from scripts.ci.maintenance.ghcr_prune_untagged import (
12 GhcrClient,
13 GhcrVersion,
14 delete_version,
15 list_container_versions,
16 main,
17)
19if TYPE_CHECKING:
20 from types import TracebackType
23# =============================================================================
24# Shared Mock Classes
25# =============================================================================
28class MockOwnerResponse:
29 """Mock response for owner type lookup."""
31 def __init__(self) -> None:
32 """Initialize mock response with default status code 200."""
33 self.status_code = 200
34 self.headers: dict[str, str] = {}
36 def raise_for_status(self) -> None:
37 """No-op since this is a successful response."""
38 return
40 def json(self) -> dict[str, str]:
41 """Return mock user type data."""
42 return {"type": "User"}
45class MockDeleteResponse:
46 """Mock response for delete operations."""
48 def __init__(self, status_code: int = 204) -> None:
49 """Initialize mock response with configurable status code."""
50 self.status_code = status_code
52 def raise_for_status(self) -> None: # pragma: no cover
53 """Raise error for unexpected status codes."""
54 if self.status_code not in (204, 404):
55 raise RuntimeError("boom")
58def make_versions_response(
59 versions_data: list[dict[str, Any]],
60 status_code: int = 200,
61) -> type:
62 """Factory for mock GET response with version data.
64 Args:
65 versions_data: List of version dictionaries to return.
66 status_code: HTTP status code for the response.
68 Returns:
69 Mock response class.
70 """
71 import httpx
73 class MockVersionsResponse:
74 def __init__(self) -> None:
75 self.status_code = status_code
76 self.headers: dict[str, str] = {}
78 def raise_for_status(self) -> None:
79 if self.status_code == 404:
80 raise httpx.HTTPStatusError(
81 message="Not Found",
82 request=httpx.Request("GET", "http://test"),
83 response=httpx.Response(404),
84 )
86 def json(self) -> list[dict[str, Any]]:
87 return versions_data
89 return MockVersionsResponse
92def make_mock_client(
93 versions_data: list[dict[str, Any]],
94 deleted: list[int],
95 missing_packages: list[str] | None = None,
96) -> type:
97 """Factory for mock httpx.Client with configurable behavior.
99 Args:
100 versions_data: Version data to return from GET requests.
101 deleted: List to append deleted version IDs to.
102 missing_packages: Package names that should return 404.
104 Returns:
105 Mock client class.
106 """
107 missing = missing_packages or []
108 versions_resp_cls = make_versions_response(versions_data)
109 not_found_resp_cls = make_versions_response([], status_code=404)
111 class _MockClient: # noqa: N801 - intentional class name for factory pattern
112 def __init__(
113 self,
114 headers: dict[str, str],
115 timeout: int,
116 ) -> None: # noqa: ARG002
117 pass
119 def __enter__(self) -> _MockClient:
120 return self
122 def __exit__(
123 self,
124 exc_type: type[BaseException] | None,
125 exc: BaseException | None,
126 tb: TracebackType | None,
127 ) -> None:
128 return None
130 def get(
131 self,
132 url: str,
133 headers: dict[str, str],
134 ) -> MockOwnerResponse | Any: # noqa: ARG002
135 # Owner type lookup returns a dict with "type" field
136 if "/users/" in url and "/packages/" not in url:
137 return MockOwnerResponse()
138 # Check for missing packages
139 for pkg in missing:
140 if pkg in url:
141 return not_found_resp_cls()
142 return versions_resp_cls()
144 def delete(
145 self,
146 url: str,
147 headers: dict[str, str],
148 ) -> MockDeleteResponse: # noqa: ARG002
149 deleted.append(int(url.rstrip("/").split("/")[-1]))
150 return MockDeleteResponse()
152 return _MockClient
155# =============================================================================
156# Tests
157# =============================================================================
160def test_version_dataclass() -> None:
161 """Construct ``GhcrVersion`` and validate fields are populated."""
162 v = GhcrVersion(id=123, tags=["latest"])
163 assert_that(v.id).is_equal_to(123)
164 assert_that(v.tags).is_equal_to(["latest"])
167def test_list_container_versions_parses_minimal_structure(
168 monkeypatch: pytest.MonkeyPatch,
169) -> None:
170 """Parse a minimal response structure into version objects.
172 Args:
173 monkeypatch: Pytest monkeypatch fixture (not used).
174 """
176 class DummyResp:
177 def __init__(self, data: list[dict[str, Any]]) -> None:
178 self._data = data
179 self.headers: dict[str, str] = {}
181 def raise_for_status(self) -> None: # pragma: no cover
182 return
184 def json(self) -> list[dict[str, Any]]:
185 return self._data
187 class DummyClient:
188 def get(
189 self,
190 url: str,
191 *,
192 headers: Mapping[str, str] | None = None,
193 ) -> DummyResp | MockOwnerResponse: # noqa: ARG002
194 # Owner type lookup returns a dict
195 if "/users/" in url and "/packages/" not in url:
196 return MockOwnerResponse()
197 # Package versions returns a list
198 return DummyResp(
199 data=[
200 {"id": 1, "metadata": {"container": {"tags": ["latest"]}}},
201 {"id": 2, "metadata": {"container": {"tags": []}}},
202 ],
203 )
205 # DummyClient is a test mock implementing only the methods needed for this test
206 # Cast to GhcrClient - the mock only implements get() which is sufficient here
207 versions = list_container_versions(
208 client=cast(GhcrClient, DummyClient()),
209 owner="owner",
210 )
211 assert_that([v.id for v in versions]).is_equal_to([1, 2])
212 assert_that(versions[0].tags).is_equal_to(["latest"])
213 assert_that(versions[1].tags).is_equal_to([])
216def test_delete_version_calls_delete(monkeypatch: pytest.MonkeyPatch) -> None:
217 """Call delete and ensure correct endpoint is used.
219 Args:
220 monkeypatch: Pytest monkeypatch fixture (not used).
221 """
222 calls: list[tuple[str, Mapping[str, str]]] = []
224 class DummyClient:
225 def delete(
226 self,
227 url: str,
228 *,
229 headers: Mapping[str, str] | None = None,
230 ) -> MockDeleteResponse: # noqa: ARG002
231 calls.append((url, headers or {}))
232 return MockDeleteResponse()
234 # DummyClient is a test mock that only implements delete().
235 # Pass base_path to avoid owner type lookup (DummyClient doesn't implement get()).
236 # Cast to GhcrClient - the mock only implements delete() which is sufficient here
237 delete_version(
238 client=cast(GhcrClient, DummyClient()),
239 owner="owner",
240 version_id=42,
241 base_path="https://api.github.com/users/owner/packages/container",
242 )
243 assert_that(calls).is_not_empty()
244 assert_that(calls[0][0]).contains("versions/42")
247def test_delete_version_raises_on_non_204_non_404() -> None:
248 """Raise when the delete operation returns an unexpected status code.
250 Raises:
251 AssertionError: If the expected RuntimeError is not raised.
252 """
254 class DummyClient:
255 def delete(
256 self,
257 url: str,
258 *,
259 headers: Mapping[str, str] | None = None,
260 ) -> MockDeleteResponse: # noqa: ARG002
261 return MockDeleteResponse(status_code=500)
263 try:
264 # DummyClient is a test mock that only implements delete().
265 # Pass base_path to avoid owner type lookup (DummyClient doesn't implement get()).
266 # Cast to GhcrClient - the mock only implements delete() which is sufficient here
267 delete_version(
268 client=cast(GhcrClient, DummyClient()),
269 owner="owner",
270 version_id=1,
271 base_path="https://api.github.com/users/owner/packages/container",
272 )
273 except RuntimeError:
274 return
275 raise AssertionError("Expected RuntimeError on non-204/404 response")
278def test_main_deletes_only_untagged(monkeypatch: pytest.MonkeyPatch) -> None:
279 """Delete only untagged versions using the main entry point.
281 Args:
282 monkeypatch: Pytest monkeypatch fixture for environment and client.
283 """
284 import httpx
286 import scripts.ci.maintenance.ghcr_prune_untagged as mod
288 deleted: list[int] = []
289 versions_data = [
290 {
291 "id": 11,
292 "created_at": "2025-08-24T10:00:00Z",
293 "metadata": {"container": {"tags": ["latest"]}},
294 },
295 {
296 "id": 22,
297 "created_at": "2025-08-24T09:00:00Z",
298 "metadata": {"container": {"tags": []}},
299 },
300 {
301 "id": 33,
302 "created_at": "2025-08-24T08:00:00Z",
303 "metadata": {"container": {"tags": ["0.4.1"]}},
304 },
305 {
306 "id": 44,
307 "created_at": "2025-08-24T07:00:00Z",
308 "metadata": {"container": {"tags": []}},
309 },
310 ]
312 mock_client = make_mock_client(
313 versions_data=versions_data,
314 deleted=deleted,
315 missing_packages=["lintro-tools"],
316 )
318 mock_httpx = type(
319 "MockHttpx",
320 (),
321 {"Client": mock_client, "HTTPStatusError": httpx.HTTPStatusError},
322 )
324 monkeypatch.setenv("GITHUB_TOKEN", "x")
325 monkeypatch.setenv("GITHUB_REPOSITORY", "owner/name")
326 monkeypatch.setattr(mod, "httpx", mock_httpx)
328 rc = main()
329 assert_that(rc).is_equal_to(0)
330 # Only untagged IDs 22 and 44 should be deleted (from py-lintro package only)
331 assert_that(sorted(deleted)).is_equal_to([22, 44])
334def test_main_respects_keep_n_and_dry_run(monkeypatch: pytest.MonkeyPatch) -> None:
335 """Respect keep-N and dry-run options when pruning.
337 Args:
338 monkeypatch: Pytest monkeypatch fixture for environment and client.
339 """
340 import httpx
342 import scripts.ci.maintenance.ghcr_prune_untagged as mod
344 deleted: list[int] = []
345 # 3 untagged versions
346 versions_data = [
347 {
348 "id": 100,
349 "created_at": "2025-08-24T12:00:00Z",
350 "metadata": {"container": {"tags": []}},
351 },
352 {
353 "id": 200,
354 "created_at": "2025-08-24T11:00:00Z",
355 "metadata": {"container": {"tags": []}},
356 },
357 {
358 "id": 300,
359 "created_at": "2025-08-24T10:00:00Z",
360 "metadata": {"container": {"tags": []}},
361 },
362 ]
364 mock_client = make_mock_client(
365 versions_data=versions_data,
366 deleted=deleted,
367 missing_packages=["lintro-tools"],
368 )
370 mock_httpx = type(
371 "MockHttpx",
372 (),
373 {"Client": mock_client, "HTTPStatusError": httpx.HTTPStatusError},
374 )
376 # Dry-run with keep 2 -> no deletions performed
377 monkeypatch.setenv("GITHUB_TOKEN", "x")
378 monkeypatch.setenv("GITHUB_REPOSITORY", "owner/name")
379 monkeypatch.setenv("GHCR_PRUNE_DRY_RUN", "1")
380 monkeypatch.setenv("GHCR_PRUNE_KEEP_UNTAGGED_N", "2")
381 monkeypatch.setattr(mod, "httpx", mock_httpx)
383 rc = main()
384 assert_that(rc).is_equal_to(0)
385 # Keep 2 newest untagged (100, 200). Would delete only 300; dry-run prevents it
386 assert_that(deleted).is_equal_to([])