chore: scaffold tests of lambdas

terraform/python/rekognition_api/conf.py

@@ -32,6 +32,7 @@
 class SettingsDefaults:
     """Default values for Settings"""
 
+    AWS_PROFILE = None
     AWS_REGION = "us-east-1"
     DEBUG_MODE = False
     TABLE_ID = "rekognition"
@@ -67,12 +68,18 @@ def empty_str_to_int_default(v: str, default: int) -> int:
 class Settings(BaseSettings):
     """Settings for Lambda functions"""
 
+    _aws_session: boto3.Session = None
+
     debug_mode: Optional[bool] = Field(
         SettingsDefaults.DEBUG_MODE,
         env="DEBUG_MODE",
         pre=True,
         getter=lambda v: empty_str_to_bool_default(v, SettingsDefaults.DEBUG_MODE),
     )
+    aws_profile: Optional[str] = Field(
+        SettingsDefaults.AWS_PROFILE,
+        env="AWS_PROFILE",
+    )
     aws_regions: Optional[List[str]] = Field(AWS_REGIONS, description="The list of AWS regions")
     aws_region: Optional[str] = Field(
         SettingsDefaults.AWS_REGION,
@@ -110,20 +117,30 @@ class Settings(BaseSettings):
         getter=lambda v: empty_str_to_int_default(v, SettingsDefaults.FACE_DETECT_THRESHOLD),
     )
 
+    @property
+    def aws_session(self):
+        """AWS session"""
+        if not self._aws_session:
+            if self.aws_profile:
+                self._aws_session = boto3.Session(profile_name=self.aws_profile, region_name=self.aws_region)
+            else:
+                self._aws_session = boto3.Session(region_name=self.aws_region)
+        return self._aws_session
+
     @property
     def s3_client(self):
         """S3 client"""
-        return boto3.resource("s3")
+        return self.aws_session.client("s3")
 
     @property
     def dynamodb_client(self):
         """DynamoDB client"""
-        return boto3.resource("dynamodb")
+        return self.aws_session.client("dynamodb")
 
     @property
     def rekognition_client(self):
         """Rekognition client"""
-        return boto3.client("rekognition")
+        return self.aws_session.client("rekognition")
 
     @property
     def dynamodb_table(self):
@@ -154,6 +171,14 @@ class Config:
 
         frozen = True
 
+    @validator("aws_profile", pre=True)
+    # pylint: disable=no-self-argument,unused-argument
+    def validate_aws_profile(cls, v, values, **kwargs):
+        """Validate aws_profile"""
+        if v in [None, ""]:
+            return SettingsDefaults.AWS_PROFILE
+        return v
+
     @validator("aws_region", pre=True)
     # pylint: disable=no-self-argument,unused-argument
     def validate_aws_region(cls, v, values, **kwargs):

terraform/python/rekognition_api/lambda_index.py

@@ -106,7 +106,7 @@ def validate_event(event):
             msg = f"lambda_index() is intended to be called for ObjectCreated:Put event, but was invoked by {event}"
             raise RekognitionIlligalInvocationError(msg)
 
-    except (TypeError, RekognitionIlligalInvocationError) as e:
+    except (KeyError, TypeError, RekognitionIlligalInvocationError) as e:
         return http_response_factory(status_code=500, body=exception_response_factory(e))
     return True
 

terraform/python/rekognition_api/tests/mock_data/json/apigateway_index_lambda_event_bad_event.json

@@ -0,0 +1,40 @@
+{
+  "event": {
+    "Records": [
+      {
+        "eventVersion": "2.1",
+        "eventSource": "aws:s3",
+        "awsRegion": "us-east-1",
+        "eventTime": "2023-12-13T20:52:38.891Z",
+        "eventName": "BAD-DATA",
+        "userIdentity": {
+          "principalId": "AWS:AROARKEXDU3E64TIYPAYH:BackplaneAssumeRoleSession"
+        },
+        "requestParameters": {
+          "sourceIPAddress": "44.210.64.80"
+        },
+        "responseElements": {
+          "x-amz-request-id": "16JS14Q4XNDTHWK4",
+          "x-amz-id-2": "bXS5u99NXOxDsV5FouyQWv1QKWbYI2rb3pMyDTnXDYla00IT8jxPK7+VDKDu1bkbxC+XoW4kbMMjAPIxQmMnmR37+Tvh3D/9"
+        },
+        "s3": {
+          "s3SchemaVersion": "1.0",
+          "configurationId": "tf-s3-lambda-20231213140006617000000009",
+          "bucket": {
+            "name": "090511222473-rekognition-f799c26b853b1d12e9092e66413f4492",
+            "ownerIdentity": {
+              "principalId": "A3NTRY8BU6N8RZ"
+            },
+            "arn": "arn:aws:s3:::090511222473-rekognition-f799c26b853b1d12e9092e66413f4492"
+          },
+          "object": {
+            "key": "Keanu-Reeves.jpg",
+            "size": 879589,
+            "eTag": "c9230f7b3889b61da34a4e0057ccb6d8",
+            "sequencer": "00657A1996C279C087"
+          }
+        }
+      }
+    ]
+  }
+}

terraform/python/rekognition_api/tests/mock_data/json/apigateway_index_lambda_event_bad_source.json

@@ -0,0 +1,40 @@
+{
+  "event": {
+    "Records": [
+      {
+        "eventVersion": "2.1",
+        "eventSource": "WRONG-SOURCE",
+        "awsRegion": "us-east-1",
+        "eventTime": "2023-12-13T20:52:38.891Z",
+        "eventName": "ObjectCreated:Put",
+        "userIdentity": {
+          "principalId": "AWS:AROARKEXDU3E64TIYPAYH:BackplaneAssumeRoleSession"
+        },
+        "requestParameters": {
+          "sourceIPAddress": "44.210.64.80"
+        },
+        "responseElements": {
+          "x-amz-request-id": "16JS14Q4XNDTHWK4",
+          "x-amz-id-2": "bXS5u99NXOxDsV5FouyQWv1QKWbYI2rb3pMyDTnXDYla00IT8jxPK7+VDKDu1bkbxC+XoW4kbMMjAPIxQmMnmR37+Tvh3D/9"
+        },
+        "s3": {
+          "s3SchemaVersion": "1.0",
+          "configurationId": "tf-s3-lambda-20231213140006617000000009",
+          "bucket": {
+            "name": "090511222473-rekognition-f799c26b853b1d12e9092e66413f4492",
+            "ownerIdentity": {
+              "principalId": "A3NTRY8BU6N8RZ"
+            },
+            "arn": "arn:aws:s3:::090511222473-rekognition-f799c26b853b1d12e9092e66413f4492"
+          },
+          "object": {
+            "key": "Keanu-Reeves.jpg",
+            "size": 879589,
+            "eTag": "c9230f7b3889b61da34a4e0057ccb6d8",
+            "sequencer": "00657A1996C279C087"
+          }
+        }
+      }
+    ]
+  }
+}

terraform/python/rekognition_api/tests/mock_data/json/apigateway_index_lambda_event_no_records.json

@@ -0,0 +1,3 @@
+{
+  "event": {}
+}

terraform/python/rekognition_api/tests/test_aws_infrastructure.py

@@ -26,6 +26,7 @@
 
 # our stuff
 sys.path.append(PYTHON_ROOT)  # noqa: E402
+from rekognition_api.conf import settings  # noqa: E402
 from rekognition_api.tests.test_setup import get_test_image  # noqa: E402
 
 
@@ -184,6 +185,21 @@ def get_api_keys(self) -> str:
                 return item["value"]
         return False
 
+    def rekognition_collection_exists(self):
+        """Test that the Rekognition collection exists."""
+        rekognition_client = self.aws_session.client("rekognition")
+        response = rekognition_client.list_collections()
+        for collection in response["CollectionIds"]:
+            if collection == settings.collection_id:
+                return True
+        return False
+
+    def test_rekognition_collection_exists(self):
+        """Test that the Rekognition collection exists."""
+        self.assertTrue(
+            self.rekognition_collection_exists(), f"Rekognition collection {settings.collection_id} does not exist."
+        )
+
     def test_aws_connection_works(self):
         """Test that the AWS connection works."""
         self.assertTrue(self.aws_connection_works(), "AWS connection failed.")

terraform/python/rekognition_api/tests/test_lambda_index.py

@@ -13,22 +13,55 @@
 PYTHON_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(os.path.dirname(__file__))))
 sys.path.append(PYTHON_ROOT)  # noqa: E402
 
+from rekognition_api.exceptions import RekognitionIlligalInvocationError  # noqa: E402
+from rekognition_api.lambda_index import (  # noqa: E402
+    get_bucket_name,
+    get_records,
+    validate_event,
+)
+
 # our stuff
 from rekognition_api.tests.test_setup import get_test_file  # noqa: E402
 
 
-# from rekognition_api.lambda_index import lambda_handler  # noqa: E402
-
-
 class TestLambdaIndex(unittest.TestCase):
     """Test Index Lambda function."""
 
     # load a mock lambda_index event
     event = get_test_file("json/apigateway_index_lambda_event.json")
+    event = event["event"]
     response = get_test_file("json/apigateway_index_lambda_response.json")
 
     def setUp(self):
         """Set up test fixtures."""
 
-    def test_noop(self):
-        """Test to ensure that test suite setup works and that lambda_handler is importable."""
+    def get_event(self, event):
+        """Get the event json from the mock file."""
+        return event["event"]
+
+    def test_get_records(self):
+        """Test get_records."""
+        records = get_records(self.event)
+        self.assertEqual(records, self.event["Records"])
+
+    def test_get_bucket_name(self):
+        """Test get_bucket_name."""
+        bucket_name = get_bucket_name(self.event)
+        self.assertEqual(bucket_name, self.event["Records"][0]["s3"]["bucket"]["name"])
+
+    def test_validate_event(self):
+        """Test validate_event."""
+        event = get_test_file("json/apigateway_index_lambda_event_no_records.json")
+        event = self.get_event(event)
+        retval = validate_event(event)
+        self.assertEqual(retval["statusCode"], 500)
+
+        event = get_test_file("json/apigateway_index_lambda_event_bad_source.json")
+        event = self.get_event(event)
+        retval = validate_event(event)
+        self.assertEqual(retval["statusCode"], 500)
+
+        event = get_test_file("json/apigateway_index_lambda_event_bad_event.json")
+        event = self.get_event(event)
+        retval = validate_event(event)
+        self.assertEqual(retval["statusCode"], 500)

terraform/python/rekognition_api/tests/test_lambda_search.py

@@ -13,23 +13,49 @@
 sys.path.append(PYTHON_ROOT)  # noqa: E402
 
 # our stuff
-from rekognition_api.tests.test_setup import get_test_file  # noqa: E402
-
-
-# from rekognition_api.lambda_search import lambda_handler  # noqa: E402
+from rekognition_api.conf import settings  # noqa: E402
+from rekognition_api.lambda_search import get_faces, get_image_from_event  # noqa: E402
+from rekognition_api.tests.test_setup import (  # noqa: E402
+    get_test_file,
+    get_test_image,
+    pack_image_data,
+)
 
 
 class TestLambdaIndex(unittest.TestCase):
     """Test Search Lambda function."""
 
+    image_filename = "Keanu-Reeves.jpg"
+    image = get_test_image(filename=image_filename)
+    image_packed = pack_image_data(filename=image_filename)
+
     # load a mock lambda_index event
-    event = get_test_file("json/apigateway_search_lambda_event.json")
+    search_event = get_test_file("json/apigateway_search_lambda_event.json")
+    index_event = get_test_file("json/apigateway_index_lambda_event.json")
     response = get_test_file("json/apigateway_search_lambda_response.json")
     dynamodb_records = get_test_file("json/dynamodb-sample-records.json")
     rekognition_search_output = get_test_file("json/rekognition_search_output.json")
 
     def setUp(self):
         """Set up test fixtures."""
 
-    def test_noop(self):
-        """Test to ensure that test suite setup works and that lambda_handler is importable."""
+    def test_get_image_from_event(self):
+        """Test get_image_from_event."""
+        #     image_from_event = get_image_from_event(self.search_event)
+
+        #     self.assertEqual(image_from_event, self.image)
+        print("Not implemented")
+        assert True
+
+    def test_get_faces(self):
+        """Test get_faces."""
+        # faces = settings.rekognition_client.search_faces_by_image(
+        #     Image=self.image_packed,
+        #     CollectionId=settings.collection_id,
+        #     MaxFaces=settings.face_detect_max_faces_count,
+        #     FaceMatchThreshold=settings.face_detect_threshold,
+        #     QualityFilter=settings.face_detect_quality_filter,
+        # )
+        # faces = get_faces(self.image_packed)
+        print("Not implemented")
+        assert True

terraform/python/rekognition_api/tests/test_setup.py

@@ -2,6 +2,8 @@
 # pylint: disable=wrong-import-position
 """Test Search Lambda function."""
 
+import base64
+
 # python stuff
 import json
 import os
@@ -29,3 +31,21 @@ def get_test_image(filename: str):
     path = os.path.join(HERE, "mock_data", "img", filename)
     with open(path, "rb") as file:
         return file.read()
+
+
+def pack_image_data(filename: str):
+    """extract and decode the raw image data from the event"""
+    image_raw = get_test_image(filename)
+    image_decoded = base64.b64decode(image_raw)
+
+    # https://stackoverflow.com/questions/6269765/what-does-the-b-character-do-in-front-of-a-string-literal
+    # Image: base64-encoded bytes or an S3 object.
+    # Image={
+    #     'Bytes': b'bytes',
+    #     'S3Object': {
+    #         'Bucket': 'string',
+    #         'Name': 'string',
+    #         'Version': 'string'
+    #     }
+    # },
+    return {"Bytes": image_decoded}