5a593fba5e
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m28s
CD Pipeline / build-and-deploy (push) Successful in 11m29s
CD Pipeline / post-deploy-checks (push) Successful in 1m54s
628 lines
25 KiB
Python
628 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
CONTROLLER_PATH = ROOT / "scripts/security/runtime_image_mirror_controller.py"
|
|
POLICY_PATH = ROOT / "config/security/runtime-image-mirror-policy.json"
|
|
STAGING_POLICY_PATH = ROOT / "config/security/runtime-image-mirror-staging-policy.json"
|
|
|
|
|
|
def _load_controller():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"runtime_image_mirror_controller", CONTROLLER_PATH
|
|
)
|
|
assert spec is not None and spec.loader is not None
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
controller = _load_controller()
|
|
|
|
|
|
class RuntimeImageMirrorControllerTest(unittest.TestCase):
|
|
def _image_with_container_kind(self, container_kind: str):
|
|
payload = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
|
payload["images"] = [payload["images"][0]]
|
|
payload["images"][0]["workload"]["container_kind"] = container_kind
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "policy.json"
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
return controller.load_policy(path).images[0]
|
|
|
|
def test_policy_is_internal_digest_pinned_and_runtime_cache_only(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
raw = POLICY_PATH.read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(policy.work_item_id, "AIA-P0-006-02A")
|
|
self.assertEqual(policy.risk_level, "high")
|
|
self.assertEqual(len(policy.images), 10)
|
|
self.assertNotIn("github.com", raw)
|
|
self.assertNotIn("ghcr.io", raw)
|
|
self.assertIn('"external_pull_allowed": false', raw)
|
|
self.assertTrue(
|
|
all(
|
|
image.runtime_ref.startswith(
|
|
"192.168.0.110:5000/awoooi/runtime-mirror/"
|
|
)
|
|
for image in policy.images
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
image.runtime_ref.endswith("@" + image.target_registry_digest)
|
|
for image in policy.images
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
image.platform_digest != image.target_registry_digest
|
|
for image in policy.images
|
|
)
|
|
)
|
|
container_kinds = [image.workload.container_kind for image in policy.images]
|
|
self.assertEqual(container_kinds.count("container"), 8)
|
|
self.assertEqual(container_kinds.count("init_container"), 2)
|
|
|
|
def test_policy_loads_and_validates_init_container_kind(self) -> None:
|
|
image = self._image_with_container_kind("init_container")
|
|
|
|
self.assertEqual(image.workload.container_kind, "init_container")
|
|
|
|
with self.assertRaisesRegex(
|
|
controller.ControllerError, "container_kind_not_allowed"
|
|
):
|
|
self._image_with_container_kind("ephemeral_container")
|
|
|
|
def test_staging_policy_is_runtime_cache_only_and_init_canary_scoped(
|
|
self,
|
|
) -> None:
|
|
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
|
|
raw = STAGING_POLICY_PATH.read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(policy.work_item_id, "AIA-P0-006-02B")
|
|
self.assertEqual(policy.risk_level, "high")
|
|
self.assertEqual(len(policy.images), 1)
|
|
image = policy.images[0]
|
|
self.assertEqual(image.runtime_repository, "argocd")
|
|
self.assertEqual(image.workload.name, "argocd-dex-server")
|
|
self.assertEqual(image.workload.container, "copyutil")
|
|
self.assertEqual(image.workload.container_kind, "init_container")
|
|
self.assertNotIn("github.com", raw)
|
|
self.assertNotIn("ghcr.io", raw)
|
|
self.assertIn('"external_pull_allowed": false', raw)
|
|
|
|
def test_argocd_copyutil_canary_matches_verified_staging_contract(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0]
|
|
canary = next(
|
|
image
|
|
for image in policy.images
|
|
if image.asset_id == "runtime-image:argocd-copyutil"
|
|
)
|
|
|
|
self.assertEqual(canary.source_index_digest, staging.source_index_digest)
|
|
self.assertEqual(canary.platform_digest, staging.platform_digest)
|
|
self.assertEqual(canary.push_ref, staging.push_ref)
|
|
self.assertEqual(canary.workload, staging.workload)
|
|
self.assertEqual(
|
|
canary.target_registry_digest,
|
|
"sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc",
|
|
)
|
|
self.assertTrue(
|
|
canary.runtime_ref.endswith("@" + canary.target_registry_digest)
|
|
)
|
|
|
|
def test_argocd_controller_batch_reuses_verified_staging_artifact(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0]
|
|
batch = {
|
|
image.asset_id: image
|
|
for image in policy.images
|
|
if image.asset_id
|
|
in {
|
|
"runtime-image:argocd-applicationset-controller",
|
|
"runtime-image:argocd-notifications-controller",
|
|
}
|
|
}
|
|
|
|
self.assertEqual(
|
|
set(batch),
|
|
{
|
|
"runtime-image:argocd-applicationset-controller",
|
|
"runtime-image:argocd-notifications-controller",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
(
|
|
image.workload.kind,
|
|
image.workload.namespace,
|
|
image.workload.name,
|
|
image.workload.container,
|
|
image.workload.container_kind,
|
|
)
|
|
for image in batch.values()
|
|
},
|
|
{
|
|
(
|
|
"deployment",
|
|
"argocd",
|
|
"argocd-applicationset-controller",
|
|
"argocd-applicationset-controller",
|
|
"container",
|
|
),
|
|
(
|
|
"deployment",
|
|
"argocd",
|
|
"argocd-notifications-controller",
|
|
"argocd-notifications-controller",
|
|
"container",
|
|
),
|
|
},
|
|
)
|
|
for image in batch.values():
|
|
self.assertEqual(image.source_index_digest, staging.source_index_digest)
|
|
self.assertEqual(image.platform_digest, staging.platform_digest)
|
|
self.assertEqual(image.push_ref, staging.push_ref)
|
|
self.assertEqual(
|
|
image.target_registry_digest,
|
|
"sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc",
|
|
)
|
|
self.assertTrue(
|
|
image.runtime_ref.endswith("@" + image.target_registry_digest)
|
|
)
|
|
|
|
def test_argocd_repo_server_batch_reuses_verified_staging_artifact(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
staging = controller.load_staging_policy(STAGING_POLICY_PATH).images[0]
|
|
batch = {
|
|
image.asset_id: image
|
|
for image in policy.images
|
|
if image.asset_id
|
|
in {
|
|
"runtime-image:argocd-repo-server",
|
|
"runtime-image:argocd-repo-server-copyutil",
|
|
}
|
|
}
|
|
|
|
self.assertEqual(
|
|
set(batch),
|
|
{
|
|
"runtime-image:argocd-repo-server",
|
|
"runtime-image:argocd-repo-server-copyutil",
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
{
|
|
(
|
|
image.workload.kind,
|
|
image.workload.namespace,
|
|
image.workload.name,
|
|
image.workload.container,
|
|
image.workload.container_kind,
|
|
)
|
|
for image in batch.values()
|
|
},
|
|
{
|
|
(
|
|
"deployment",
|
|
"argocd",
|
|
"argocd-repo-server",
|
|
"argocd-repo-server",
|
|
"container",
|
|
),
|
|
(
|
|
"deployment",
|
|
"argocd",
|
|
"argocd-repo-server",
|
|
"copyutil",
|
|
"init_container",
|
|
),
|
|
},
|
|
)
|
|
for image in batch.values():
|
|
self.assertEqual(image.source_index_digest, staging.source_index_digest)
|
|
self.assertEqual(image.platform_digest, staging.platform_digest)
|
|
self.assertEqual(image.push_ref, staging.push_ref)
|
|
self.assertEqual(
|
|
image.target_registry_digest,
|
|
"sha256:8b7e25d7e6036d6750c5bda8920cbd80d8902c25771666f6767dc97bd7fab8fc",
|
|
)
|
|
self.assertTrue(
|
|
image.runtime_ref.endswith("@" + image.target_registry_digest)
|
|
)
|
|
|
|
def test_policy_rejects_external_pull_and_mutable_runtime_ref(self) -> None:
|
|
payload = json.loads(POLICY_PATH.read_text(encoding="utf-8"))
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "policy.json"
|
|
payload["source_contract"]["external_pull_allowed"] = True
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
with self.assertRaisesRegex(
|
|
controller.ControllerError, "external_pull_must_be_disabled"
|
|
):
|
|
controller.load_policy(path)
|
|
|
|
payload["source_contract"]["external_pull_allowed"] = False
|
|
payload["images"][0][
|
|
"runtime_ref"
|
|
] = "192.168.0.110:5000/awoooi/runtime-mirror/dex:latest"
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
with self.assertRaisesRegex(
|
|
controller.ControllerError, "runtime_ref_digest_mismatch"
|
|
):
|
|
controller.load_policy(path)
|
|
|
|
def test_archive_verifier_requires_source_platform_manifest(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
image = policy.images[0]
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
archive_path = Path(temp) / "image.tar"
|
|
member = "blobs/sha256/" + image.platform_digest.removeprefix("sha256:")
|
|
with tarfile.open(archive_path, "w") as archive:
|
|
info = tarfile.TarInfo(member)
|
|
info.size = 2
|
|
archive.addfile(info, BytesIO(b"{}"))
|
|
|
|
self.assertTrue(
|
|
controller._archive_contains_digest(archive_path, image.platform_digest)
|
|
)
|
|
self.assertFalse(
|
|
controller._archive_contains_digest(
|
|
archive_path, image.target_registry_digest
|
|
)
|
|
)
|
|
|
|
def test_target_digest_verifier_uses_internal_registry_transport(self) -> None:
|
|
image = controller.load_policy(POLICY_PATH).images[0]
|
|
with patch.object(controller.subprocess, "run") as run:
|
|
run.return_value.returncode = 0
|
|
|
|
self.assertTrue(controller._target_digest_present(image))
|
|
|
|
command = run.call_args.args[0]
|
|
self.assertEqual(command[:3], ["docker", "manifest", "inspect"])
|
|
self.assertIn("--insecure", command)
|
|
self.assertEqual(command[-1], controller._target_digest_ref(image))
|
|
|
|
def test_source_manifests_match_internal_policy(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
by_asset = {image.asset_id: image for image in policy.images}
|
|
kured = (ROOT / "k8s/kured/kured.yaml").read_text(encoding="utf-8")
|
|
event = (ROOT / "k8s/observability/event-exporter.yaml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
kube_state_metrics = (
|
|
ROOT / "k8s/kube-state-metrics/kube-state-metrics.yaml"
|
|
).read_text(encoding="utf-8")
|
|
node_problem_detector = (ROOT / "k8s/npd/node-problem-detector.yaml").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
self.assertIn(by_asset["runtime-image:kured"].runtime_ref, kured)
|
|
self.assertIn(by_asset["runtime-image:event-exporter"].runtime_ref, event)
|
|
self.assertIn(
|
|
by_asset["runtime-image:kube-state-metrics"].runtime_ref,
|
|
kube_state_metrics,
|
|
)
|
|
self.assertIn(
|
|
by_asset["runtime-image:node-problem-detector"].runtime_ref,
|
|
node_problem_detector,
|
|
)
|
|
self.assertNotIn("ghcr.io", kured)
|
|
self.assertNotIn("ghcr.io", event)
|
|
self.assertNotIn("registry.k8s.io", kube_state_metrics)
|
|
self.assertNotIn("registry.k8s.io", node_problem_detector)
|
|
|
|
def test_controller_and_cd_never_pull_external_images(self) -> None:
|
|
controller_source = CONTROLLER_PATH.read_text(encoding="utf-8")
|
|
workflow = (ROOT / ".gitea/workflows/cd.yaml").read_text(encoding="utf-8")
|
|
|
|
self.assertNotIn("docker pull", controller_source)
|
|
self.assertNotIn('"images", "pull"', controller_source)
|
|
self.assertIn("Mirror Runtime Cache Images to Harbor", workflow)
|
|
self.assertIn("Apply Runtime Image Mirror Policy", workflow)
|
|
self.assertIn("runtime-image-mirror-staging-policy.json", workflow)
|
|
self.assertIn("runtime_image_staging_stage=verified", workflow)
|
|
self.assertIn("runtime_image_mirror_controller.py", workflow)
|
|
mirror_block = workflow[
|
|
workflow.index(
|
|
"- name: Mirror Runtime Cache Images to Harbor"
|
|
) : workflow.index("# ── API 鏡像建置")
|
|
]
|
|
self.assertNotIn("HARBOR_PASSWORD", mirror_block)
|
|
|
|
def test_staging_reuses_verified_internal_artifact_without_export(self) -> None:
|
|
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
|
|
image = policy.images[0]
|
|
descriptor = controller.RegistryDescriptor(
|
|
digest="sha256:" + "1" * 64,
|
|
config_digest=image.source_config_digest,
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
receipt_path = Path(temp) / "receipt.json"
|
|
with (
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"verify_source",
|
|
return_value="quay.io/argoproj/argocd:v3.3.6",
|
|
),
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"source_config_digest",
|
|
return_value=image.source_config_digest,
|
|
),
|
|
patch.object(controller.RuntimeCache, "export") as export,
|
|
patch.object(
|
|
controller, "_registry_descriptor", return_value=descriptor
|
|
),
|
|
patch.object(controller, "_registry_digest_present", return_value=True),
|
|
patch.object(controller, "_run") as run,
|
|
):
|
|
receipt = controller.stage_images(
|
|
policy,
|
|
trace_id="aia-p0-006-02b-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.assertEqual(receipt["images"][0]["execution"], "already_present")
|
|
self.assertTrue(receipt["images"][0]["provenance_config_digest_match"])
|
|
export.assert_not_called()
|
|
run.assert_not_called()
|
|
|
|
def test_staging_exports_cached_source_and_verifies_remote_provenance(
|
|
self,
|
|
) -> None:
|
|
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
|
|
image = policy.images[0]
|
|
descriptor = controller.RegistryDescriptor(
|
|
digest="sha256:" + "2" * 64,
|
|
config_digest=image.source_config_digest,
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
receipt_path = Path(temp) / "receipt.json"
|
|
with (
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"verify_source",
|
|
return_value="quay.io/argoproj/argocd:v3.3.6",
|
|
),
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"source_config_digest",
|
|
return_value=image.source_config_digest,
|
|
),
|
|
patch.object(controller.RuntimeCache, "export") as export,
|
|
patch.object(
|
|
controller,
|
|
"_registry_descriptor",
|
|
side_effect=[None, descriptor],
|
|
),
|
|
patch.object(controller, "_local_image_present", return_value=False),
|
|
patch.object(controller, "_archive_contains_digest", return_value=True),
|
|
patch.object(
|
|
controller,
|
|
"_local_config_digest",
|
|
return_value=image.source_config_digest,
|
|
),
|
|
patch.object(controller, "_registry_digest_present", return_value=True),
|
|
patch.object(controller, "_remove_local_image") as remove,
|
|
patch.object(controller, "_run") as run,
|
|
):
|
|
receipt = controller.stage_images(
|
|
policy,
|
|
trace_id="aia-p0-006-02b-stage",
|
|
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"], 1)
|
|
self.assertEqual(receipt["images"][0]["execution"], "staged_from_runtime_cache")
|
|
self.assertEqual(
|
|
receipt["images"][0]["target_registry_digest"], descriptor.digest
|
|
)
|
|
export.assert_called_once()
|
|
commands = [call.args[0] for call in run.call_args_list]
|
|
self.assertEqual([command[1] for command in commands], ["load", "tag", "push"])
|
|
self.assertTrue(all("pull" not in command for command in commands))
|
|
self.assertEqual(remove.call_count, 2)
|
|
|
|
def test_staging_rejects_internal_artifact_with_wrong_provenance(self) -> None:
|
|
policy = controller.load_staging_policy(STAGING_POLICY_PATH)
|
|
image = policy.images[0]
|
|
descriptor = controller.RegistryDescriptor(
|
|
digest="sha256:" + "3" * 64,
|
|
config_digest="sha256:" + "4" * 64,
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
with (
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"verify_source",
|
|
return_value="quay.io/argoproj/argocd:v3.3.6",
|
|
),
|
|
patch.object(
|
|
controller.RuntimeCache,
|
|
"source_config_digest",
|
|
return_value=image.source_config_digest,
|
|
),
|
|
patch.object(
|
|
controller, "_registry_descriptor", return_value=descriptor
|
|
),
|
|
self.assertRaisesRegex(
|
|
controller.ControllerError,
|
|
"internal_registry_provenance_mismatch",
|
|
),
|
|
):
|
|
controller.stage_images(
|
|
policy,
|
|
trace_id="aia-p0-006-02b-wrong-provenance",
|
|
receipt_path=Path(temp) / "receipt.json",
|
|
ssh_key=Path("/unused/key"),
|
|
known_hosts=Path("/unused/known-hosts"),
|
|
apply=True,
|
|
)
|
|
|
|
def test_verified_mirror_receipt_is_bound_to_trace_and_policy(self) -> None:
|
|
policy = controller.load_policy(POLICY_PATH)
|
|
receipt = {
|
|
"schema_version": controller.RECEIPT_SCHEMA_VERSION,
|
|
"trace_id": "aia-p0-006-02a-test",
|
|
"work_item_id": policy.work_item_id,
|
|
"policy_checksum": policy.checksum,
|
|
"state": "verified",
|
|
"verified_count": len(policy.images),
|
|
}
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
path = Path(temp) / "receipt.json"
|
|
path.write_text(json.dumps(receipt), encoding="utf-8")
|
|
self.assertEqual(
|
|
controller._read_mirror_receipt(path, policy, "aia-p0-006-02a-test"),
|
|
receipt,
|
|
)
|
|
with self.assertRaisesRegex(
|
|
controller.ControllerError, "mirror_receipt_not_verified"
|
|
):
|
|
controller._read_mirror_receipt(path, policy, "different-trace")
|
|
|
|
def test_init_container_read_and_patch_use_init_container_spec(self) -> None:
|
|
image = self._image_with_container_kind("init_container")
|
|
kubernetes = controller.Kubernetes(
|
|
use_sudo=False,
|
|
kubeconfig=None,
|
|
server=None,
|
|
)
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"containers": [{"name": "other", "image": "other:image"}],
|
|
"initContainers": [
|
|
{
|
|
"name": image.workload.container,
|
|
"image": "source:init",
|
|
}
|
|
],
|
|
}
|
|
}
|
|
}
|
|
}
|
|
with patch.object(kubernetes, "workload", return_value=workload):
|
|
self.assertEqual(kubernetes.current_image(image), "source:init")
|
|
|
|
with patch.object(kubernetes, "run") as run:
|
|
kubernetes.patch_image(image, image.runtime_ref, dry_run=True)
|
|
|
|
args = run.call_args.args[0]
|
|
patch_payload = json.loads(args[args.index("--patch") + 1])
|
|
pod_spec = patch_payload["spec"]["template"]["spec"]
|
|
self.assertNotIn("containers", pod_spec)
|
|
self.assertEqual(
|
|
pod_spec["initContainers"],
|
|
[{"name": image.workload.container, "image": image.runtime_ref}],
|
|
)
|
|
self.assertIn("--dry-run=server", args)
|
|
|
|
def test_init_container_verifier_accepts_completed_digest_pinned_status(
|
|
self,
|
|
) -> None:
|
|
image = self._image_with_container_kind("init_container")
|
|
kubernetes = controller.Kubernetes(
|
|
use_sudo=False,
|
|
kubeconfig=None,
|
|
server=None,
|
|
)
|
|
workload = {
|
|
"spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}}
|
|
}
|
|
pods = {
|
|
"items": [
|
|
{
|
|
"metadata": {"name": "runtime-mirror-test-1"},
|
|
"status": {
|
|
"initContainerStatuses": [
|
|
{
|
|
"name": image.workload.container,
|
|
"ready": False,
|
|
"state": {"terminated": {"exitCode": 0}},
|
|
"imageID": "registry.local/repo@"
|
|
+ image.target_registry_digest,
|
|
}
|
|
]
|
|
},
|
|
}
|
|
]
|
|
}
|
|
with (
|
|
patch.object(kubernetes, "workload", return_value=workload),
|
|
patch.object(kubernetes, "run") as run,
|
|
):
|
|
run.return_value.stdout = json.dumps(pods)
|
|
self.assertEqual(kubernetes.verify_pods(image), 1)
|
|
|
|
def test_init_container_verifier_rejects_incomplete_or_wrong_digest(self) -> None:
|
|
image = self._image_with_container_kind("init_container")
|
|
workload = {
|
|
"spec": {"selector": {"matchLabels": {"app": "runtime-mirror-test"}}}
|
|
}
|
|
cases = (
|
|
(
|
|
{
|
|
"state": {"running": {}},
|
|
"imageID": "repo@" + image.target_registry_digest,
|
|
},
|
|
"pod_init_container_not_completed",
|
|
),
|
|
(
|
|
{
|
|
"state": {"terminated": {"exitCode": 0}},
|
|
"imageID": "repo@sha256:" + "0" * 64,
|
|
},
|
|
"pod_image_digest_mismatch",
|
|
),
|
|
)
|
|
for status, error_code in cases:
|
|
with self.subTest(error_code=error_code):
|
|
status["name"] = image.workload.container
|
|
pods = {
|
|
"items": [
|
|
{"metadata": {}, "status": {"initContainerStatuses": [status]}}
|
|
]
|
|
}
|
|
kubernetes = controller.Kubernetes(
|
|
use_sudo=False,
|
|
kubeconfig=None,
|
|
server=None,
|
|
)
|
|
with (
|
|
patch.object(kubernetes, "workload", return_value=workload),
|
|
patch.object(kubernetes, "run") as run,
|
|
self.assertRaisesRegex(controller.ControllerError, error_code),
|
|
):
|
|
run.return_value.stdout = json.dumps(pods)
|
|
kubernetes.verify_pods(image)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|