From 6f997bf118225f4c946af9336839beae58e9a16b Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 27 Jul 2026 17:29:06 +0000 Subject: [PATCH] test(release): cover publication failure paths --- tests/unit/test_release_publish.py | 69 ++++++++- tests/unit/test_release_qualification.py | 171 ++++++++++++++++++++++- 2 files changed, 235 insertions(+), 5 deletions(-) diff --git a/tests/unit/test_release_publish.py b/tests/unit/test_release_publish.py index ece53bcb..1f8ff969 100644 --- a/tests/unit/test_release_publish.py +++ b/tests/unit/test_release_publish.py @@ -2,12 +2,20 @@ from __future__ import annotations import tempfile import unittest +import urllib.error from pathlib import Path -from unittest.mock import patch +from unittest.mock import MagicMock, patch from bot_bottle.release_bundle import build_bundle_index, canonical_bytes from bot_bottle.release_manifest import ReleaseManifestError -from bot_bottle.release_publish import bundle_url, publish_bundle +from bot_bottle.release_publish import ( + bundle_url, + remote_bytes, + remote_sha256, + request, + upload as publish_artifact, + publish_bundle, +) from tests.unit.test_release_manifest import manifest @@ -96,3 +104,60 @@ class TestReleasePublish(unittest.TestCase): ReleaseManifestError, "already differs", ): publish_bundle(index, wheel) + + def test_request_applies_auth_and_length(self) -> None: + result = request( + "https://packages.example/file", + method="PUT", + data=b"body", + length=4, + ) + self.assertEqual(result.get_method(), "PUT") + self.assertEqual(result.get_header("Authorization"), "token token") + self.assertEqual(result.get_header("Content-length"), "4") + + @patch("bot_bottle.release_publish.urllib.request.urlopen") + def test_remote_helpers_read_response(self, urlopen: MagicMock) -> None: + response = urlopen.return_value.__enter__.return_value + response.read.side_effect = [b"body"] + self.assertEqual( + remote_bytes("https://packages.example/file"), b"body") + response.read.side_effect = [b"body", b""] + self.assertEqual( + remote_sha256("https://packages.example/file"), + "230d8358dc8e8890b4c58deeb62912ee2" + "f20357ae92a5cc861b98e68fe31acb5", + ) + + @patch("bot_bottle.release_publish.urllib.request.urlopen") + def test_remote_helpers_handle_not_found(self, urlopen: MagicMock) -> None: + urlopen.side_effect = urllib.error.HTTPError( + "url", 404, "missing", MagicMock(), None) + self.assertIsNone(remote_bytes("https://packages.example/file")) + self.assertIsNone(remote_sha256("https://packages.example/file")) + + @patch("bot_bottle.release_publish.urllib.request.urlopen") + def test_remote_helpers_report_registry_errors( + self, urlopen: MagicMock, + ) -> None: + urlopen.side_effect = urllib.error.URLError("offline") + with self.assertRaisesRegex(ReleaseManifestError, "offline"): + remote_bytes("https://packages.example/file") + with self.assertRaisesRegex(ReleaseManifestError, "offline"): + remote_sha256("https://packages.example/file") + + @patch("bot_bottle.release_publish.urllib.request.urlopen") + def test_upload_supports_bytes_and_files(self, urlopen: MagicMock) -> None: + urlopen.return_value.__enter__.return_value = MagicMock() + publish_artifact("https://packages.example/bytes", b"body") + with tempfile.TemporaryDirectory() as tmp: + source = Path(tmp) / "artifact" + source.write_bytes(b"artifact") + publish_artifact("https://packages.example/file", source) + self.assertEqual(urlopen.call_count, 2) + + @patch("bot_bottle.release_publish.urllib.request.urlopen") + def test_upload_reports_failure(self, urlopen: MagicMock) -> None: + urlopen.side_effect = urllib.error.URLError("offline") + with self.assertRaisesRegex(ReleaseManifestError, "publishing"): + publish_artifact("https://packages.example/file", b"body") diff --git a/tests/unit/test_release_qualification.py b/tests/unit/test_release_qualification.py index 898cbc99..12d2b015 100644 --- a/tests/unit/test_release_qualification.py +++ b/tests/unit/test_release_qualification.py @@ -4,27 +4,33 @@ from __future__ import annotations import json import unittest +import urllib.error +from email.message import Message from unittest import mock from bot_bottle.release_manifest import ReleaseManifestError from bot_bottle.release_qualification import ( build_release_pointer, + canonical_pointer, + delete, + parse_release_pointer, publish_qualification, release_channel, + release_version, ) COMMIT = "a" * 40 -def bundle() -> bytes: +def bundle(source_commit: str = COMMIT) -> bytes: return json.dumps({ "schema": 1, - "source_commit": COMMIT, + "source_commit": source_commit, "wheel": { "filename": "bot_bottle.whl", "url": ( "https://gitea.dideric.is/api/packages/didericis/generic/" - f"bot-bottle-builds/{COMMIT}/bot_bottle.whl" + f"bot-bottle-builds/{source_commit}/bot_bottle.whl" ), "sha256": "b" * 64, }, @@ -54,6 +60,14 @@ class ReleaseQualificationTest(unittest.TestCase): with self.assertRaises(ReleaseManifestError): release_channel("latest") + def test_release_versions_are_monotonic(self) -> None: + self.assertLess( + release_version("v1.2.3-rc.4"), release_version("v1.2.3")) + self.assertLess( + release_version("v1.2.3"), release_version("v2.0.0")) + with self.assertRaises(ReleaseManifestError): + release_version("v1") + def test_builds_qualified_pointer(self) -> None: pointer = build_release_pointer( tag="v1.2.3", @@ -63,6 +77,42 @@ class ReleaseQualificationTest(unittest.TestCase): ) self.assertEqual(pointer["channel"], "production") self.assertTrue(pointer["qualified"]) + self.assertTrue(canonical_pointer(pointer).endswith(b"\n")) + + def test_rejects_invalid_pointer_fields(self) -> None: + pointer = build_release_pointer( + tag="v1.2.3", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + changes = ( + ("schema", 2), + ("qualified", False), + ("channel", "staging"), + ("source_commit", "short"), + ("bundle_index_url", "https://wrong.example/index.json"), + ("qualification", {}), + ) + for key, value in changes: + with self.subTest(key=key), self.assertRaises(ReleaseManifestError): + parse_release_pointer({**pointer, key: value}) + + def test_rejects_invalid_build_inputs(self) -> None: + with self.assertRaises(ReleaseManifestError): + build_release_pointer( + tag="v1.2.3", + source_commit="short", + workflow_run="run", + qualified_at="now", + ) + with self.assertRaises(ReleaseManifestError): + build_release_pointer( + tag="v1.2.3", + source_commit=COMMIT, + workflow_run="", + qualified_at="now", + ) @mock.patch("bot_bottle.release_qualification.upload") @mock.patch("bot_bottle.release_qualification.remote_bytes") @@ -79,6 +129,45 @@ class ReleaseQualificationTest(unittest.TestCase): publish_qualification(pointer) self.assertEqual(upload.call_count, 2) + @mock.patch("bot_bottle.release_qualification.upload") + @mock.patch("bot_bottle.release_qualification.remote_bytes") + def test_exact_existing_publication_is_noop( + self, remote: mock.Mock, upload: mock.Mock, + ) -> None: + pointer = build_release_pointer( + tag="v1.2.3", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + wanted = canonical_pointer(pointer) + remote.side_effect = [bundle(), wanted, wanted] + publish_qualification(pointer) + upload.assert_not_called() + + @mock.patch("bot_bottle.release_qualification.upload") + @mock.patch("bot_bottle.release_qualification.delete") + @mock.patch("bot_bottle.release_qualification.remote_bytes") + def test_advances_to_newer_channel( + self, remote: mock.Mock, remove: mock.Mock, upload: mock.Mock, + ) -> None: + old = build_release_pointer( + tag="v1.0.0", + source_commit="b" * 40, + workflow_run="old", + qualified_at="then", + ) + pointer = build_release_pointer( + tag="v2.0.0", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + remote.side_effect = [bundle(), None, canonical_pointer(old)] + publish_qualification(pointer) + remove.assert_called_once() + self.assertEqual(upload.call_count, 2) + @mock.patch("bot_bottle.release_qualification.upload") @mock.patch("bot_bottle.release_qualification.remote_bytes") def test_rejects_non_monotonic_channel( @@ -100,6 +189,82 @@ class ReleaseQualificationTest(unittest.TestCase): with self.assertRaisesRegex(ReleaseManifestError, "not monotonic"): publish_qualification(pointer) + @mock.patch("bot_bottle.release_qualification.upload") + @mock.patch("bot_bottle.release_qualification.delete") + @mock.patch("bot_bottle.release_qualification.remote_bytes") + def test_explicit_rollback_allows_older_channel( + self, remote: mock.Mock, remove: mock.Mock, _upload: mock.Mock, + ) -> None: + old = build_release_pointer( + tag="v2.0.0", + source_commit="b" * 40, + workflow_run="old", + qualified_at="then", + ) + pointer = build_release_pointer( + tag="v1.0.0", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + remote.side_effect = [bundle(), None, canonical_pointer(old)] + publish_qualification(pointer, allow_rollback=True) + remove.assert_called_once() + + @mock.patch("bot_bottle.release_qualification.remote_bytes") + def test_rejects_missing_or_malformed_bundle(self, remote: mock.Mock) -> None: + pointer = build_release_pointer( + tag="v1.2.3", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + remote.return_value = None + with self.assertRaisesRegex(ReleaseManifestError, "does not exist"): + publish_qualification(pointer) + remote.return_value = b"not-json" + with self.assertRaisesRegex(ReleaseManifestError, "malformed"): + publish_qualification(pointer) + remote.return_value = bundle("b" * 40) + with self.assertRaisesRegex(ReleaseManifestError, "does not match"): + publish_qualification(pointer) + + @mock.patch("bot_bottle.release_qualification.upload") + @mock.patch("bot_bottle.release_qualification.remote_bytes") + def test_rejects_changed_release_or_malformed_channel( + self, remote: mock.Mock, _upload: mock.Mock, + ) -> None: + pointer = build_release_pointer( + tag="v1.2.3", + source_commit=COMMIT, + workflow_run="run", + qualified_at="now", + ) + remote.side_effect = [bundle(), b"different"] + with self.assertRaisesRegex(ReleaseManifestError, "already differs"): + publish_qualification(pointer) + remote.side_effect = [bundle(), None, b"not-json"] + with self.assertRaisesRegex(ReleaseManifestError, "malformed"): + publish_qualification(pointer) + + @mock.patch("bot_bottle.release_qualification.urllib.request.urlopen") + def test_delete_handles_success_and_not_found(self, urlopen: mock.Mock) -> None: + urlopen.return_value.__enter__.return_value = mock.Mock() + delete("https://example.test/channel") + urlopen.side_effect = urllib.error.HTTPError( + "url", 404, "missing", Message(), None) + delete("https://example.test/channel") + + @mock.patch("bot_bottle.release_qualification.urllib.request.urlopen") + def test_delete_reports_registry_errors(self, urlopen: mock.Mock) -> None: + urlopen.side_effect = urllib.error.HTTPError( + "url", 500, "broken", Message(), None) + with self.assertRaisesRegex(ReleaseManifestError, "HTTP 500"): + delete("https://example.test/channel") + urlopen.side_effect = urllib.error.URLError("offline") + with self.assertRaisesRegex(ReleaseManifestError, "offline"): + delete("https://example.test/channel") + if __name__ == "__main__": unittest.main()