Started adding lambdas support

cbr_athena/_extra_methods/for_osbot_aws/Lambda.py

@@ -0,0 +1,11 @@
+from osbot_utils.utils.Misc import wait_for
+
+# this could be done with a waiter, but it takes much more time (1.5 secs vs 5 seconds)
+#    self.lambda_function.client().get_waiter('function_updated').wait(FunctionName=self.lambda_function.name)
+def wait_for_function_update_to_complete(lambda_function, max_attempts=20, wait_time=0.1):
+    for i in range(max_attempts):
+        status = lambda_function.configuration().get('LastUpdateStatus')
+        if status == 'Successful':
+            break
+        wait_for(wait_time)
+    return status

cbr_athena/lambdas/dev/hello_world.py

@@ -0,0 +1,3 @@
+def run(event, context):
+    name = event.get('name') or 'world'
+    return f'hello {name}!'

cbr_athena/lambdas/dev/permissions_tests.py

@@ -0,0 +1,8 @@
+from osbot_aws.apis.S3 import S3
+
+
+def run(event, context):
+    name = event.get('name') or 'world'
+    s3 = S3()
+    return s3.buckets()
+    return f'testing permissions {name}!'

requirements.txt

@@ -3,4 +3,9 @@
 git+https://github.com/owasp-sbot/OSBot-Utils.git
 gradio
 openai
-python-dotenv
+python-dotenv
+
+# for AWS support
+git+https://github.com/owasp-sbot/OSBot-AWS.git
+boto3
+#botocore

tests/aws/test_AWS_Setup.py

@@ -0,0 +1,13 @@
+from unittest import TestCase
+
+from osbot_aws.apis.STS import STS
+
+
+class test_AWS_Setup(TestCase):
+
+    def setUp(self) -> None:
+        pass
+
+    def test_setup(self):
+        STS().check_current_session_credentials()
+        assert True is True

tests/lambdas/dev/test_hello_world.py

@@ -0,0 +1,35 @@
+#@pytest.mark.skip('needs AWS credentials to run')
+from unittest import TestCase
+
+from osbot_aws.apis.Session import Session
+from osbot_aws.apis.test_helpers.Temp_Aws_Roles import Temp_Aws_Roles
+from osbot_aws.deploy.Deploy_Lambda import Deploy_Lambda
+from osbot_utils.utils.Dev import pprint
+
+from cbr_athena._extra_methods.for_osbot_aws.Lambda import wait_for_function_update_to_complete
+from cbr_athena.lambdas.dev.hello_world import run
+
+
+class test_hello_world(TestCase):
+
+    def setUp(self) -> None:
+        self.handler         = run
+        self.deploy_lambda   = Deploy_Lambda(self.handler)
+        self.lambda_function = self.deploy_lambda.lambda_function()
+
+    def test_invoke_directly(self):
+        result = self.handler({'name': 'me'}, None)
+        assert result == 'hello me!'
+
+    def test_deploying_lambda_function(self):
+        assert self.deploy_lambda.update().get('status') == 'ok'
+        wait_for_function_update_to_complete(self.lambda_function)
+
+        assert self.lambda_function.invoke({'name': 'me'}) == 'hello me!'
+
+    def test_lambda_setup(self):
+        temp_aws_roles = Temp_Aws_Roles()
+        if  temp_aws_roles.for_lambda_invocation__not_exists():
+            result = temp_aws_roles.for_lambda_invocation__create()
+            pprint(result)
+        assert temp_aws_roles.for_lambda_invocation_exists()

tests/lambdas/dev/test_permissions_tests.py

@@ -0,0 +1,55 @@
+#@pytest.mark.skip('needs AWS credentials to run')
+from unittest import TestCase
+
+from osbot_aws.apis.STS import STS
+from osbot_aws.apis.Session import Session
+from osbot_aws.apis.test_helpers.Temp_Aws_Roles import Temp_Aws_Roles
+from osbot_aws.deploy.Deploy_Lambda import Deploy_Lambda
+from osbot_aws.helpers.IAM_Role import IAM_Role
+from osbot_utils.utils.Dev import pprint
+
+from cbr_athena._extra_methods.for_osbot_aws.Lambda import wait_for_function_update_to_complete
+from cbr_athena.lambdas.dev.permissions_tests import run
+
+
+class test_hello_world(TestCase):
+
+    def setUp(self) -> None:
+        self.handler         = run
+        self.deploy_lambda   = Deploy_Lambda(self.handler)
+        self.lambda_function = self.deploy_lambda.lambda_function()
+
+    def test_invoke_directly(self):
+        self.test_check_credentials()
+        result = self.handler({'name': 'me'}, None)
+        pprint(result)
+        #assert result == 'hello me!'
+
+    def test_check_credentials(self):
+
+        role_arn  = self.deploy_lambda.role_arn
+        role_name = self.deploy_lambda.role_arn.split('/')[-1]
+        #role_arn = "temp_role_for_lambda_invocation"
+        pprint(role_arn)
+        iam_role = IAM_Role(role_name)
+
+        pprint(iam_role.policies_statements())
+        #
+        #STS()
+
+    def test_deploying_lambda_function(self):
+        self.test_check_credentials()
+        self.deploy_lambda.add_osbot_aws()
+        assert self.deploy_lambda.update().get('status') == 'ok'
+        wait_for_function_update_to_complete(self.lambda_function)
+
+        result = self.lambda_function.invoke({'name': 'me'})
+        pprint(result)
+        #assert self.lambda_function.invoke({'name': 'me'}) == 'testing permissions me!'
+
+    def test_lambda_setup(self):
+        temp_aws_roles = Temp_Aws_Roles()
+        if  temp_aws_roles.for_lambda_invocation__not_exists():
+            result = temp_aws_roles.for_lambda_invocation__create()
+            pprint(result)
+        assert temp_aws_roles.for_lambda_invocation_exists()