test: add missing tests

Author: igorlg

unicorn_properties/tests/events/lambda/content_integrity_validator_function_success.json

@@ -0,0 +1,10 @@
+{
+    "imageModerations": [
+        {
+            "ModerationLabels": []
+        }
+    ],
+    "contentSentiment": {
+        "Sentiment": "POSITIVE"
+    }
+}

unicorn_properties/tests/events/lambda/wait_for_contract_approval_function.json

@@ -0,0 +1,6 @@
+{
+    "TaskToken": "xxx",
+    "Input": {
+        "property_id": "usa/anytown/main-street/123"
+    }
+}

unicorn_properties/tests/unit/test_content_integrity_validator_function.py

@@ -1,3 +1,94 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: MIT-0
 
+import os
+
+import pytest
+from unittest import mock
+
+from .lambda_context import LambdaContext
+from .helper import load_event, return_env_vars_dict
+
+
+@pytest.fixture
+def stepfunctions_event():
+    return load_event('tests/events/lambda/content_integrity_validator_function_success.json')
+    
+
+@pytest.fixture
+def invalid_image_moderation(stepfunctions_event):
+    e = {
+        'imageModerations': [
+            {
+                'ModerationLabels': [
+                    {
+                        "Confidence": 99.24723052978516,
+                        "ParentName": "",
+                        "Name": "Explicit Nudity"
+                    },
+                    {
+                        "Confidence": 99.24723052978516,
+                        "ParentName": "Explicit Nudity",
+                        "Name": "Graphic Male Nudity"
+                    },
+                ]
+            }
+        ]
+    }
+    return {**stepfunctions_event, **e}
+
+
+@pytest.fixture
+def invalid_content_sentiment(stepfunctions_event):
+    e = {
+        'contentSentiment': {
+            'Sentiment': 'NEGATIVE'
+        }
+    }
+    return {**stepfunctions_event, **e}
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_valid_image_and_valid_sentiment(stepfunctions_event):
+    from properties_service import content_integrity_validator_function
+    ret = content_integrity_validator_function.lambda_handler(stepfunctions_event, LambdaContext())
+
+    assert ret['validation_result'] == "PASS"
+    assert ret['imageModerations'] == stepfunctions_event['imageModerations']
+    assert ret['contentSentiment'] == stepfunctions_event['contentSentiment']
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_valid_image_and_invalid_sentiment(invalid_content_sentiment):
+    event = invalid_content_sentiment
+
+    from properties_service import content_integrity_validator_function
+    ret = content_integrity_validator_function.lambda_handler(event, LambdaContext())
+
+    assert ret['validation_result'] == "FAIL"
+    assert ret['imageModerations'] == event['imageModerations']
+    assert ret['contentSentiment'] == event['contentSentiment']
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_invalid_image_and_valid_sentiment(invalid_image_moderation):
+    event = invalid_image_moderation
+
+    from properties_service import content_integrity_validator_function
+    ret = content_integrity_validator_function.lambda_handler(event, LambdaContext())
+
+    assert ret['validation_result'] == "FAIL"
+    assert ret['imageModerations'] == event['imageModerations']
+    assert ret['contentSentiment'] == event['contentSentiment']
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_invalid_image_and_invalid_sentiment(invalid_image_moderation, invalid_content_sentiment):
+    event = {**invalid_image_moderation, **invalid_content_sentiment}
+
+    from properties_service import content_integrity_validator_function
+    ret = content_integrity_validator_function.lambda_handler(event, LambdaContext())
+
+    assert ret['validation_result'] == "FAIL"
+    assert ret['imageModerations'] == event['imageModerations']
+    assert ret['contentSentiment'] == event['contentSentiment']

unicorn_properties/tests/unit/test_properties_approval_sync_function.py

@@ -1,3 +1,37 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: MIT-0
 
+import os
+from importlib import reload
+
+from unittest import mock
+
+from .lambda_context import LambdaContext
+from .helper import load_event, return_env_vars_dict
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_handle_status_changed_draft(stepfunction, mocker):
+    ddbstream_event = load_event('tests/events/dbb_stream_events/contract_status_changed_draft.json')
+
+    from properties_service import properties_approval_sync_function
+    reload(properties_approval_sync_function)
+
+    ret = properties_approval_sync_function.lambda_handler(ddbstream_event, LambdaContext())
+
+    assert ret is None
+
+
+# NOTE: This test cannot be implemented at this time because `moto`` does not yet support mocking `stepfunctions.send_task_success`
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_handle_status_changed_approved(caplog, stepfunction, mocker):
+    pass
+    # ddbstream_event = load_event('tests/events/dbb_stream_events/status_approved_waiting_for_approval.json')
+
+    # from properties_service import properties_approval_sync_function
+    # reload(properties_approval_sync_function)
+
+    # ret = properties_approval_sync_function.lambda_handler(ddbstream_event, LambdaContext())
+
+    # assert ret is None
+    # assert 'Contract status for property is APPROVED' in caplog.text

unicorn_properties/tests/unit/test_wait_for_contract_approval_function.py

@@ -1,3 +1,29 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: MIT-0
 
+import os
+from importlib import reload
+
+from unittest import mock
+
+from .lambda_context import LambdaContext
+from .helper import load_event, return_env_vars_dict, create_ddb_table_contracts_with_entry
+
+
+@mock.patch.dict(os.environ, return_env_vars_dict(), clear=True)
+def test_handle_wait_for_contract_approval_function(dynamodb, mocker):
+    stepfunctions_event = load_event('tests/events/lambda/wait_for_contract_approval_function.json')
+
+    from properties_service import wait_for_contract_approval_function
+    reload(wait_for_contract_approval_function)
+
+    create_ddb_table_contracts_with_entry(dynamodb)
+
+    ddbitem_before = dynamodb.Table('table1').get_item(Key={'property_id': stepfunctions_event['Input']['property_id']})
+    assert 'sfn_wait_approved_task_token' not in ddbitem_before['Item']
+
+    ret = wait_for_contract_approval_function.lambda_handler(stepfunctions_event, LambdaContext())
+    ddbitem_after = dynamodb.Table('table1').get_item(Key={'property_id': stepfunctions_event['Input']['property_id']})
+
+    assert ret['property_id'] == stepfunctions_event['Input']['property_id']    
+    assert ddbitem_after['Item']['sfn_wait_approved_task_token'] == stepfunctions_event['TaskToken']