97 lines
3.7 KiB
Python
97 lines
3.7 KiB
Python
|
import json
|
||
|
|
import os
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
from types import SimpleNamespace
|
||
|
|
from unittest import mock
|
||
|
|
|
||
|
|
import create_image
|
||
|
|
|
||
|
|
|
||
|
|
class CreateImageTest(unittest.TestCase):
|
||
|
|
def test_writes_available_manifest_and_removes_pending_manifest(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
||
|
|
manifest_path = Path(temporary_directory) / "image-manifest.json"
|
||
|
|
environment = {
|
||
|
|
"OCI_IMAGE_ADMISSION_SHA256": "admission",
|
||
|
|
"OCI_IMAGE_BASE_IMAGE_OCID": "base",
|
||
|
|
"OCI_IMAGE_COMPARTMENT_ID": "compartment",
|
||
|
|
"OCI_IMAGE_GUEST_SHA256": "guest",
|
||
|
|
"OCI_IMAGE_INSTANCE_ID": "instance",
|
||
|
|
"OCI_IMAGE_MANIFEST_PATH": str(manifest_path),
|
||
|
|
"OCI_IMAGE_NAME": "relay",
|
||
|
|
"OCI_IMAGE_REVISION_SHA256": "revision",
|
||
|
|
"OCI_IMAGE_SMTP_DENY_SHA256": "smtp-deny",
|
||
|
|
"OCI_IMAGE_SMTP_EVIDENCE_SHA256": "smtp-evidence",
|
||
|
|
"OCI_IMAGE_VLAGENT_SHA256": "vlagent",
|
||
|
|
"OCI_OPERATOR_API_VAULT_PATH": "secret/oci/operator-api",
|
||
|
|
}
|
||
|
|
secret = {
|
||
|
|
"data": {
|
||
|
|
"data": {
|
||
|
|
"fingerprint": "fingerprint",
|
||
|
|
"private_key": "private-key",
|
||
|
|
"region": "region",
|
||
|
|
"tenancy_ocid": "tenancy",
|
||
|
|
"user_ocid": "user",
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
process = SimpleNamespace(stdout=json.dumps(secret))
|
||
|
|
image = SimpleNamespace(
|
||
|
|
display_name="relay",
|
||
|
|
id="image",
|
||
|
|
lifecycle_state="AVAILABLE",
|
||
|
|
)
|
||
|
|
response = SimpleNamespace(data=image)
|
||
|
|
client = mock.Mock()
|
||
|
|
client.create_image.return_value = response
|
||
|
|
client.get_image.return_value = response
|
||
|
|
|
||
|
|
with (
|
||
|
|
mock.patch.dict(os.environ, environment, clear=True),
|
||
|
|
mock.patch.object(
|
||
|
|
create_image.subprocess,
|
||
|
|
"run",
|
||
|
|
return_value=process,
|
||
|
|
),
|
||
|
|
mock.patch.object(create_image.oci.signer, "Signer"),
|
||
|
|
mock.patch.object(
|
||
|
|
create_image.oci.core,
|
||
|
|
"ComputeClient",
|
||
|
|
return_value=client,
|
||
|
|
),
|
||
|
|
mock.patch.object(
|
||
|
|
create_image.oci,
|
||
|
|
"wait_until",
|
||
|
|
return_value=response,
|
||
|
|
),
|
||
|
|
):
|
||
|
|
create_image.main()
|
||
|
|
|
||
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
|
|
build = manifest["builds"][0]
|
||
|
|
self.assertEqual("region:image", build["artifact_id"])
|
||
|
|
self.assertEqual("AVAILABLE", manifest["lifecycle_state"])
|
||
|
|
self.assertEqual("smtp-deny", build["custom_data"]["smtp_deny_sha256"])
|
||
|
|
self.assertEqual("smtp-evidence", build["custom_data"]["smtp_evidence_sha256"])
|
||
|
|
self.assertFalse(manifest_path.with_suffix(".json.pending").exists())
|
||
|
|
|
||
|
|
def test_refuses_existing_pending_manifest(self) -> None:
|
||
|
|
with tempfile.TemporaryDirectory() as temporary_directory:
|
||
|
|
manifest_path = Path(temporary_directory) / "image-manifest.json"
|
||
|
|
manifest_path.with_suffix(".json.pending").touch()
|
||
|
|
with (
|
||
|
|
mock.patch.dict(
|
||
|
|
os.environ,
|
||
|
|
{"OCI_IMAGE_MANIFEST_PATH": str(manifest_path)},
|
||
|
|
clear=True,
|
||
|
|
),
|
||
|
|
self.assertRaises(FileExistsError),
|
||
|
|
):
|
||
|
|
create_image.main()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|