ci(infra): test and publish one candidate artifact
tracker-policy-pr / check-pr (pull_request) Successful in 20s
test / integration-docker (pull_request) Successful in 22s
lint / lint (push) Failing after 46s
test / build-infra (pull_request) Failing after 49s
test / integration-firecracker (pull_request) Has been skipped
test / coverage (pull_request) Has been skipped
test / unit (pull_request) Successful in 1m28s
test / publish-infra (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 20s
test / integration-docker (pull_request) Successful in 22s
lint / lint (push) Failing after 46s
test / build-infra (pull_request) Failing after 49s
test / integration-firecracker (pull_request) Has been skipped
test / coverage (pull_request) Has been skipped
test / unit (pull_request) Successful in 1m28s
test / publish-infra (pull_request) Has been skipped
This commit is contained in:
@@ -6,8 +6,10 @@ read it into memory. Network is mocked; no Docker, no real build.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import tempfile
|
||||
import unittest
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
@@ -58,5 +60,45 @@ class TestPut(unittest.TestCase):
|
||||
self.assertEqual(b"abc123 rootfs\n", captured[0].data)
|
||||
|
||||
|
||||
class TestPublishBundle(unittest.TestCase):
|
||||
def _bundle(self, root: Path, version: str) -> None:
|
||||
payload = b"candidate"
|
||||
(root / "version.txt").write_text(version + "\n")
|
||||
(root / "rootfs.ext4.gz").write_bytes(payload)
|
||||
digest = hashlib.sha256(payload).hexdigest()
|
||||
(root / "rootfs.ext4.gz.sha256").write_text(
|
||||
f"{digest} rootfs.ext4.gz\n")
|
||||
|
||||
def test_existing_identical_artifact_is_success(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
root = Path(d)
|
||||
self._bundle(root, "v1")
|
||||
sha = (root / "rootfs.ext4.gz.sha256").read_bytes()
|
||||
response = mock.MagicMock()
|
||||
response.__enter__.return_value.read.return_value = sha
|
||||
with mock.patch.object(
|
||||
pub.infra_artifact, "infra_artifact_version", return_value="v1"
|
||||
), mock.patch.object(
|
||||
pub.urllib.request, "urlopen", return_value=response
|
||||
), mock.patch.object(pub, "_put") as put:
|
||||
self.assertEqual("v1", pub._publish_bundle(root, "token"))
|
||||
put.assert_not_called()
|
||||
|
||||
def test_partial_artifact_is_replaced(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
root = Path(d)
|
||||
self._bundle(root, "v1")
|
||||
missing = urllib.error.HTTPError("u", 404, "missing", {}, None)
|
||||
with mock.patch.object(
|
||||
pub.infra_artifact, "infra_artifact_version", return_value="v1"
|
||||
), mock.patch.object(
|
||||
pub.urllib.request, "urlopen", side_effect=missing
|
||||
), mock.patch.object(pub, "_delete") as delete, \
|
||||
mock.patch.object(pub, "_put") as put:
|
||||
pub._publish_bundle(root, "token")
|
||||
self.assertEqual(3, delete.call_count)
|
||||
self.assertEqual(3, put.call_count)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user