Add support for Hook Annotations. (#277)

Author: mrinaudo-aws

.pre-commit-config.yaml

@@ -50,6 +50,7 @@ repos:
   hooks:
   - id: bandit
     files: ^(src|python)/
+    additional_dependencies: [pbr]
 - repo: https://github.com/pre-commit/mirrors-mypy
   rev: v1.7.0
   hooks:

src/cloudformation_cli_python_lib/__init__.py

@@ -7,6 +7,9 @@
     BaseHookHandlerRequest,
     BaseResourceHandlerRequest,
     HandlerErrorCode,
+    HookAnnotation,
+    HookAnnotationSeverityLevel,
+    HookAnnotationStatus,
     HookContext,
     HookInvocationPoint,
     HookProgressEvent,

src/cloudformation_cli_python_lib/hook.py

@@ -283,6 +283,7 @@ def _create_progress_response(
         response.errorCode = progress_event.errorCode
         if request:
             response.clientRequestToken = request.get("clientRequestToken")
+        response.annotations = progress_event.annotations
         return response
 
     @staticmethod

src/cloudformation_cli_python_lib/interface.py

@@ -58,6 +58,20 @@ class HookStatus(str, _AutoName):
     FAILED = auto()
 
 
+class HookAnnotationStatus(str, _AutoName):
+    PASSED = auto()
+    FAILED = auto()
+    SKIPPED = auto()
+
+
+class HookAnnotationSeverityLevel(str, _AutoName):
+    INFORMATIONAL = auto()
+    LOW = auto()
+    MEDIUM = auto()
+    HIGH = auto()
+    CRITICAL = auto()
+
+
 class HandlerErrorCode(str, _AutoName):
     NotUpdatable = auto()
     InvalidRequest = auto()
@@ -105,6 +119,26 @@ def _deserialize(
         raise NotImplementedError()
 
 
+@dataclass
+class HookAnnotation:
+    annotationName: str
+    status: HookAnnotationStatus
+    statusMessage: Optional[str] = None
+    remediationMessage: Optional[str] = None
+    remediationLink: Optional[str] = None
+    severityLevel: Optional[HookAnnotationSeverityLevel] = None
+
+    def _serialize(self) -> Mapping[str, Any]:
+        ser = {k: v for k, v in self.__dict__.items() if v is not None}
+
+        ser["status"] = ser.pop("status").name
+
+        if self.severityLevel:
+            ser["severityLevel"] = self.severityLevel.name
+
+        return ser
+
+
 # pylint: disable=too-many-instance-attributes
 @dataclass
 class ProgressEvent:
@@ -118,6 +152,7 @@ class ProgressEvent:
     resourceModel: Optional[BaseModel] = None
     resourceModels: Optional[List[BaseModel]] = None
     nextToken: Optional[str] = None
+    annotations: Optional[List[HookAnnotation]] = None
 
     def _serialize(self) -> MutableMapping[str, Any]:
         # to match Java serialization, which drops `null` values, and the
@@ -126,6 +161,10 @@ def _serialize(self) -> MutableMapping[str, Any]:
 
         # mutate to what's expected in the response
 
+        # removing hook response only fields
+        ser.pop("result", None)
+        ser.pop("annotations", None)
+
         ser["status"] = ser.pop("status").name
 
         if self.resourceModel:
@@ -184,25 +223,39 @@ class HookProgressEvent:
     callbackDelaySeconds: int = 0
     result: Optional[str] = None
     clientRequestToken: Optional[str] = None
+    annotations: Optional[List[HookAnnotation]] = None
 
     def _serialize(self) -> MutableMapping[str, Any]:
         # to match Java serialization, which drops `null` values, and the
         # contract tests currently expect this also
         ser = {k: v for k, v in self.__dict__.items() if v is not None}
 
         # mutate to what's expected in the response
-
         ser["hookStatus"] = ser.pop("hookStatus").name
 
+        if self.annotations:
+            ser["annotations"] = [
+                annotation._serialize()  # pylint: disable=protected-access
+                for annotation in self.annotations
+            ]
         if self.errorCode:
             ser["errorCode"] = self.errorCode.name
+
         return ser
 
     @classmethod
     def failed(
-        cls: Type["HookProgressEvent"], error_code: HandlerErrorCode, message: str = ""
+        cls: Type["HookProgressEvent"],
+        error_code: HandlerErrorCode,
+        message: str = "",
+        annotations: Optional[List[HookAnnotation]] = None,
     ) -> "HookProgressEvent":
-        return cls(hookStatus=HookStatus.FAILED, errorCode=error_code, message=message)
+        return cls(
+            hookStatus=HookStatus.FAILED,
+            errorCode=error_code,
+            message=message,
+            annotations=annotations,
+        )
 
 
 @dataclass

src/cloudformation_cli_python_lib/resource.py

@@ -235,6 +235,8 @@ def print_or_log(message: str) -> None:
 
         if progress.result:  # pragma: no cover
             progress.result = None
+        if progress.annotations:  # pragma: no cover
+            progress.annotations = None
 
         # use the raw event_data as a last-ditch attempt to call back if the
         # request is invalid

src/setup.py

@@ -3,7 +3,7 @@
 
 setup(
     name="cloudformation-cli-python-lib",
-    version="2.1.19",
+    version="2.1.20",
     description=__doc__,
     author="Amazon Web Services",
     author_email="aws-cloudformation-developers@amazon.com",

tests/lib/interface_test.py

@@ -6,11 +6,15 @@
 from cloudformation_cli_python_lib.interface import (
     BaseModel,
     HandlerErrorCode,
+    HookAnnotation,
+    HookAnnotationSeverityLevel,
+    HookAnnotationStatus,
     HookProgressEvent,
     HookStatus,
     OperationStatus,
     ProgressEvent,
 )
+from cloudformation_cli_python_lib.utils import KitchenSinkEncoder
 
 import hypothesis.strategies as s  # pylint: disable=C0411
 import json
@@ -140,6 +144,80 @@ def test_hook_progress_event_failed_is_json_serializable(error_code, message):
     }
 
 
+@given(
+    s.sampled_from(HandlerErrorCode),
+    s.text(ascii_letters),
+    s.sampled_from(HookAnnotationSeverityLevel),
+)
+def test_hook_progress_event_failed_with_annotations_is_json_serializable(
+    error_code,
+    message,
+    annotation_severity_level,
+):
+    event = HookProgressEvent(
+        hookStatus=OperationStatus.FAILED,
+        message=message,
+        errorCode=error_code,
+        annotations=[
+            HookAnnotation(
+                annotationName="test_annotation_name_1",
+                status=HookAnnotationStatus.FAILED,
+                statusMessage="test_status_message_1",
+                remediationMessage="test_remediation_message",
+                remediationLink="https://localhost/test-1",
+                severityLevel=annotation_severity_level,
+            ),
+            HookAnnotation(
+                annotationName="test_annotation_name_2",
+                status=HookAnnotationStatus.PASSED,
+                statusMessage="test_status_message_2",
+            ),
+        ],
+    )
+
+    assert event.hookStatus == HookStatus.FAILED
+    assert event.errorCode == error_code
+    assert event.message == message
+
+    assert event.annotations[0].annotationName == "test_annotation_name_1"
+    assert event.annotations[0].status == HookAnnotationStatus.FAILED.name
+    assert event.annotations[0].statusMessage == "test_status_message_1"
+    assert event.annotations[0].remediationMessage == "test_remediation_message"
+    assert event.annotations[0].remediationLink == "https://localhost/test-1"
+    assert event.annotations[0].severityLevel == annotation_severity_level
+
+    assert event.annotations[1].annotationName == "test_annotation_name_2"
+    assert event.annotations[1].status == HookAnnotationStatus.PASSED.name
+    assert event.annotations[1].statusMessage == "test_status_message_2"
+
+    assert json.loads(
+        json.dumps(
+            event._serialize(),
+            cls=KitchenSinkEncoder,
+        )
+    ) == {
+        "hookStatus": HookStatus.FAILED.value,
+        "errorCode": error_code.value,
+        "message": message,
+        "callbackDelaySeconds": 0,
+        "annotations": [
+            {
+                "annotationName": "test_annotation_name_1",
+                "status": "FAILED",
+                "statusMessage": "test_status_message_1",
+                "remediationMessage": "test_remediation_message",
+                "remediationLink": "https://localhost/test-1",
+                "severityLevel": annotation_severity_level.name,
+            },
+            {
+                "annotationName": "test_annotation_name_2",
+                "status": "PASSED",
+                "statusMessage": "test_status_message_2",
+            },
+        ],
+    }
+
+
 @given(s.text(ascii_letters))
 def test_hook_progress_event_serialize_to_response_with_context(message):
     event = HookProgressEvent(
@@ -154,6 +232,43 @@ def test_hook_progress_event_serialize_to_response_with_context(message):
     }
 
 
+@given(s.text(ascii_letters))
+def test_hook_progress_event_serialize_to_response_with_context_with_annotation(
+    message,
+):
+    event = HookProgressEvent(
+        hookStatus=HookStatus.SUCCESS,
+        message=message,
+        callbackContext={"a": "b"},
+        annotations=[
+            HookAnnotation(
+                annotationName="test_annotation_name",
+                status=HookAnnotationStatus.PASSED,
+                statusMessage="test_status_message",
+            ),
+        ],
+    )
+
+    assert json.loads(
+        json.dumps(
+            event._serialize(),
+            cls=KitchenSinkEncoder,
+        )
+    ) == {
+        "hookStatus": HookStatus.SUCCESS.name,  # pylint: disable=no-member
+        "message": message,
+        "callbackContext": {"a": "b"},
+        "callbackDelaySeconds": 0,
+        "annotations": [
+            {
+                "annotationName": "test_annotation_name",
+                "status": "PASSED",
+                "statusMessage": "test_status_message",
+            },
+        ],
+    }
+
+
 @given(s.text(ascii_letters))
 def test_hook_progress_event_serialize_to_response_with_data(message):
     result = "My hook data"
@@ -169,6 +284,42 @@ def test_hook_progress_event_serialize_to_response_with_data(message):
     }
 
 
+@given(s.text(ascii_letters))
+def test_hook_progress_event_serialize_to_response_with_data_with_annotation(message):
+    result = "My hook data"
+    event = HookProgressEvent(
+        hookStatus=HookStatus.SUCCESS,
+        message=message,
+        result=result,
+        annotations=[
+            HookAnnotation(
+                annotationName="test_annotation_name",
+                status=HookAnnotationStatus.PASSED,
+                statusMessage="test_status_message",
+            ),
+        ],
+    )
+
+    assert json.loads(
+        json.dumps(
+            event._serialize(),
+            cls=KitchenSinkEncoder,
+        )
+    ) == {
+        "hookStatus": HookStatus.SUCCESS.name,  # pylint: disable=no-member
+        "message": message,
+        "callbackDelaySeconds": 0,
+        "result": result,
+        "annotations": [
+            {
+                "annotationName": "test_annotation_name",
+                "status": "PASSED",
+                "statusMessage": "test_status_message",
+            },
+        ],
+    }
+
+
 @given(s.text(ascii_letters))
 def test_hook_progress_event_serialize_to_response_with_error_code(message):
     event = HookProgressEvent(
@@ -185,6 +336,43 @@ def test_hook_progress_event_serialize_to_response_with_error_code(message):
     }
 
 
+@given(s.text(ascii_letters))
+def test_hook_progress_event_serialize_to_response_with_error_code_with_annotation(
+    message,
+):
+    event = HookProgressEvent(
+        hookStatus=HookStatus.FAILED,
+        message=message,
+        errorCode=HandlerErrorCode.InvalidRequest,
+        annotations=[
+            HookAnnotation(
+                annotationName="test_annotation_name",
+                status=HookAnnotationStatus.FAILED,
+                statusMessage="test_status_message",
+            ),
+        ],
+    )
+
+    assert json.loads(
+        json.dumps(
+            event._serialize(),
+            cls=KitchenSinkEncoder,
+        )
+    ) == {
+        "hookStatus": HookStatus.FAILED.name,  # pylint: disable=no-member
+        "message": message,
+        "errorCode": HandlerErrorCode.InvalidRequest.name,  # pylint: disable=no-member
+        "callbackDelaySeconds": 0,
+        "annotations": [
+            {
+                "annotationName": "test_annotation_name",
+                "status": "FAILED",
+                "statusMessage": "test_status_message",
+            },
+        ],
+    }
+
+
 def test_operation_status_enum_matches_sdk(client):
     sdk = set(client.meta.service_model.shape_for("OperationStatus").enum)
     enum = set(OperationStatus.__members__)