test: assert digest rollback fails before side effects
CI / tests-and-images (pull_request) Successful in 2m6s
Runner smoke test / runner-smoke (pull_request) Successful in 3s

This commit is contained in:
codex committed 2026-10-20 00:13:55 +03:00
1 parent fc0473e7e2
commit 802bfd117e
1 file changed
+44
+44
View File
@@ -1,7 +1,11 @@
import json
import subprocess
import sys
import unittest
from types import SimpleNamespace
from unittest.mock import patch
import rollback
from rollback import image_restore_plan, require_tag_rollback, tag_previous_images, verify_previous_images
@@ -72,6 +76,46 @@ class RollbackPlanTests(unittest.TestCase):
tag_previous_images(plan)
run_mock.assert_not_called()
def test_digest_rollback_fails_before_stopping_or_changing_references(self):
backend_id = "sha256:" + "a" * 64
opensearch_id = "sha256:" + "b" * 64
digest = "registry/backend@sha256:" + "c" * 64
manifest = {
"images": {
"backend": {"configured": digest, "image_id": backend_id},
"opensearch": self.manifest["images"]["opensearch"],
},
"deployment_files": {
"deploy/production/compose.yaml": "compose-hash",
"deploy/production/Caddyfile": "caddy-hash",
},
}
rendered = {
"services": {
"backend": {"image": digest},
"opensearch": {"image": self.configured["opensearch"]},
}
}
fake_archive = SimpleNamespace(
digest=lambda path: "compose-hash" if path.name == "compose.yaml" else "caddy-hash",
verify_archive=lambda _backup: manifest,
)
backup = rollback.BACKUP_DIR / "synthetic.tar.gz"
with (
patch("rollback.Path.is_file", return_value=True),
patch("rollback.Path.is_symlink", return_value=False),
patch("rollback.Path.stat", return_value=SimpleNamespace(st_uid=0, st_mode=0o100600)),
patch("rollback.compose", return_value=json.dumps(rendered)) as compose_mock,
patch("rollback.run", side_effect=[backend_id, backend_id, opensearch_id]) as run_mock,
patch.dict(sys.modules, {"archive": fake_archive}),
):
with self.assertRaisesRegex(ValueError, "not supported safely"):
rollback.rollback(backup, data_compatible=True)
compose_mock.assert_called_once_with("config", "--format", "json", capture=True)
self.assertEqual(run_mock.call_count, 3)
self.assertTrue(all(call.args[0][:3] == ["docker", "image", "inspect"]
for call in run_mock.call_args_list))
def test_tag_rollback_keeps_existing_image_tag_operation(self):
plan = image_restore_plan(self.manifest, self.configured)
with patch("rollback.run") as run_mock: