wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

perf(cd): skip redundant mirror and bootstrap work

Browse Source
This commit is contained in:
Your Name
2026-07-18 19:34:31 +08:00
parent e7e3bcf8a5
commit 5c43b3da2e
5 changed files with 269 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
scripts/security/tests/test_runtime_image_mirror_controller.py
View File

@@ -46,6 +46,10 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
return replace(policy, images=(policy.images[0],))
def _single_image_policy(self):
policy = controller.load_policy(POLICY_PATH)
return replace(policy, images=(policy.images[0],))
def _image_with_availability_guard(self):
image = controller.load_policy(POLICY_PATH).images[0]
return replace(
@@ -904,12 +908,12 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
controller.RuntimeCache,
"verify_source",
return_value="quay.io/argoproj/argocd:v3.3.6",
),
) as verify_source,
patch.object(
controller.RuntimeCache,
"source_config_digest",
return_value=image.source_config_digest,
),
) as source_config_digest,
patch.object(controller.RuntimeCache, "export") as export,
patch.object(
controller, "_registry_descriptor", return_value=descriptor
@@ -928,11 +932,85 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
self.assertEqual(receipt["state"], "verified")
self.assertEqual(receipt["missing_before_count"], 0)
self.assertFalse(receipt["source_cache_accessed"])
self.assertEqual(receipt["images"][0]["execution"], "already_present")
self.assertTrue(receipt["images"][0]["provenance_config_digest_match"])
self.assertFalse(receipt["images"][0]["source_cache_verified"])
self.assertEqual(
receipt["images"][0]["source_cache_verification_skipped_reason"],
"internal_registry_provenance_already_verified",
)
verify_source.assert_not_called()
source_config_digest.assert_not_called()
export.assert_not_called()
run.assert_not_called()
def test_mirror_reuses_verified_target_digest_without_runtime_cache(self) -> None:
policy = self._single_image_policy()
with tempfile.TemporaryDirectory() as temp:
receipt_path = Path(temp) / "receipt.json"
with (
patch.object(
controller.RuntimeCache,
"verify_source",
return_value="quay.io/example/image:v1",
) as verify_source,
patch.object(controller.RuntimeCache, "export") as export,
patch.object(controller, "_target_digest_present", return_value=True),
patch.object(controller, "_run") as run,
):
receipt = controller.mirror_images(
policy,
trace_id="aia-p0-006-02a-existing",
receipt_path=receipt_path,
ssh_key=Path("/unused/key"),
known_hosts=Path("/unused/known-hosts"),
apply=True,
)
self.assertEqual(receipt["state"], "verified")
self.assertEqual(receipt["missing_before_count"], 0)
self.assertFalse(receipt["source_cache_accessed"])
self.assertEqual(receipt["images"][0]["execution"], "already_present")
self.assertFalse(receipt["images"][0]["source_cache_verified"])
self.assertEqual(
receipt["images"][0]["source_cache_verification_skipped_reason"],
"target_digest_already_verified",
)
verify_source.assert_not_called()
export.assert_not_called()
run.assert_not_called()
def test_mirror_missing_target_still_verifies_runtime_cache(self) -> None:
policy = self._single_image_policy()
with tempfile.TemporaryDirectory() as temp:
receipt_path = Path(temp) / "receipt.json"
with (
patch.object(
controller.RuntimeCache,
"verify_source",
return_value="quay.io/example/image:v1",
) as verify_source,
patch.object(controller.RuntimeCache, "export") as export,
patch.object(controller, "_target_digest_present", return_value=False),
):
receipt = controller.mirror_images(
policy,
trace_id="aia-p0-006-02a-missing-check",
receipt_path=receipt_path,
ssh_key=Path("/unused/key"),
known_hosts=Path("/unused/known-hosts"),
apply=False,
)
self.assertEqual(receipt["state"], "blocked_with_safe_next_action")
self.assertEqual(receipt["missing_before_count"], 1)
self.assertTrue(receipt["source_cache_accessed"])
self.assertEqual(receipt["images"][0]["execution"], "check_only_missing")
self.assertTrue(receipt["images"][0]["source_cache_verified"])
verify_source.assert_called_once_with(policy.images[0])
export.assert_not_called()
def test_staging_exports_cached_source_and_verifies_remote_provenance(
self,
) -> None:
@@ -985,6 +1063,7 @@ class RuntimeImageMirrorControllerTest(unittest.TestCase):
self.assertEqual(receipt["state"], "verified")
self.assertEqual(receipt["missing_before_count"], 1)
self.assertTrue(receipt["source_cache_accessed"])
self.assertEqual(receipt["images"][0]["execution"], "staged_from_runtime_cache")
self.assertEqual(
receipt["images"][0]["target_registry_digest"], descriptor.digest