added test_deploy_lambda__to_aws__dev__qa__prod

Author: DinisCruz

osbot_fast_api_serverless/deploy/Deploy__Serverless__Fast_API.py

@@ -1,3 +1,4 @@
+from osbot_utils.utils.Http                                                    import GET_json
 from osbot_utils.utils.Objects                                                 import obj
 from osbot_aws.aws.lambda_.Lambda                                              import DEFAULT__LAMBDA__EPHEMERAL_STORAGE, DEFAULT__LAMBDA__MEMORY_SIZE
 from osbot_aws.aws.lambda_.dependencies.Lambda__Dependency                     import Lambda__Dependency
@@ -13,7 +14,10 @@
 from osbot_fast_api_serverless.deploy.Schema__AWS_Setup__Serverless__Fast_API  import Schema__AWS_Setup__Serverless__Fast_API
 
 BASE__LAMBDA_NAME__FAST_API__SERVERLESS  = 'fast-api__serverless'        # make this a Safe_Str__Lambda_Name
-
+DEFAULT__ERROR_MESSAGE__WHEN_FAST_API_IS_OK = ('The adapter was unable to infer a handler to use for the '
+                                                'event. This is likely related to how the Lambda function was '
+                                                'invoked. (Are you testing locally? Make sure the request '
+                                                'payload is valid for a supported handler.)')
 
 class Deploy__Serverless__Fast_API(Type_Safe):
     aws_config       : AWS_Config
@@ -45,9 +49,14 @@ def deploy_lambda(self):
             _.set_env_variable(ENV_VAR__FAST_API__AUTH__API_KEY__VALUE, self.api_key__value())
             return _
 
-    def delete_function(self):
+    def delete(self):
         return self.lambda_function().delete()
 
+    def create(self):
+        if self.create_or_update__lambda_function():
+            self.create__lambda_function__url()
+            return True
+        return False
 
     def create_or_update__lambda_function(self):
         with self.deploy_lambda() as _:
@@ -69,6 +78,15 @@ def create__lambda_function__url(self):
                 function_url = result.get('function_url_create').get('FunctionUrl')
             return function_url
 
+    def invoke(self, payload=None):
+        return self.lambda_function().invoke(payload)
+
+    def invoke__function_url(self, path=''):
+        with self.lambda_function() as _:
+            url     = _.function_url() +  path
+            headers = { self.api_key__name(): self.api_key__value()}
+            return GET_json(url, headers=headers)
+
     def lambda_configuration(self):
         return obj(self.lambda_function().info().get('Configuration'))
 

poetry.lock

@@ -6,7 +6,7 @@
 from osbot_fast_api_serverless.deploy.Deploy__Serverless__Fast_API import Deploy__Serverless__Fast_API
 
 
-class test_deploy_lambda__to_aws(TestCase):
+class test_deploy_lambda__to_aws__dev(TestCase):
 
     @classmethod
     def setUpClass(cls):
@@ -59,4 +59,4 @@ def test_3__lambda_configuration(self):
 
     def test_4__delete_function(self):
         with self.deploy_fast_api as _:
-            assert _.delete_function() is True
+            assert _.delete() is True

…deploy_aws/test_deploy_lambda__to_aws.py …y_aws/test_deploy_lambda__to_aws__dev.pytests/deploy_aws/test_deploy_lambda__to_aws.py renamed to tests/deploy_aws/test_deploy_lambda__to_aws__dev.py tests/deploy_aws/test_deploy_lambda__to_aws.py renamed to tests/deploy_aws/test_deploy_lambda__to_aws__dev.py

@@ -0,0 +1,42 @@
+import pytest
+from unittest                                                       import TestCase
+from osbot_fast_api_serverless.utils.Version                        import version__osbot_fast_api_serverless
+from osbot_fast_api_serverless.deploy.Deploy__Serverless__Fast_API  import Deploy__Serverless__Fast_API, DEFAULT__ERROR_MESSAGE__WHEN_FAST_API_IS_OK
+
+
+class test_deploy_lambda__to_aws__dev__qa__prod(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.deploy_fast_api__dev  = Deploy__Serverless__Fast_API(stage = 'dev')
+        cls.deploy_fast_api__qa   = Deploy__Serverless__Fast_API(stage = 'qa')
+        cls.deploy_fast_api__prod = Deploy__Serverless__Fast_API(stage = 'prod')
+
+        with cls.deploy_fast_api__dev as _:
+            if _.aws_config.aws_configured() is False:
+                pytest.skip("this test needs valid AWS credentials")
+
+    def test_1__check_stages(self):
+        assert self.deploy_fast_api__dev .stage == 'dev'
+        assert self.deploy_fast_api__qa  .stage == 'qa'
+        assert self.deploy_fast_api__prod.stage == 'prod'
+
+    def test_2__create(self):
+        assert self.deploy_fast_api__dev .create() is True
+        assert self.deploy_fast_api__qa  .create() is True
+        assert self.deploy_fast_api__prod.create() is True
+
+    def test_3__invoke(self):
+        assert self.deploy_fast_api__dev .invoke().get('errorMessage') == DEFAULT__ERROR_MESSAGE__WHEN_FAST_API_IS_OK
+        assert self.deploy_fast_api__qa  .invoke().get('errorMessage') == DEFAULT__ERROR_MESSAGE__WHEN_FAST_API_IS_OK
+        assert self.deploy_fast_api__prod.invoke().get('errorMessage') == DEFAULT__ERROR_MESSAGE__WHEN_FAST_API_IS_OK
+
+    def test_3__invoke__function_url(self):
+        version = {'version': version__osbot_fast_api_serverless}
+        assert self.deploy_fast_api__dev .invoke__function_url('/info/version') == version
+        assert self.deploy_fast_api__qa  .invoke__function_url('/info/version') == version
+        assert self.deploy_fast_api__prod.invoke__function_url('/info/version') == version
+
+    def test_4__delete(self):
+        assert self.deploy_fast_api__dev .delete() is True
+        assert self.deploy_fast_api__qa  .delete() is True
+        assert self.deploy_fast_api__prod.delete() is True