Merge pull request #6 from hypoport/dev

adds support for lambda function configuration

Author: bweigel

aws_dataclasses/lambda_function_configuration.py

@@ -0,0 +1,111 @@
+from datetime import datetime
+from typing import Dict, Any, List
+
+import arrow
+from dataclasses import dataclass, InitVar, field
+
+from aws_dataclasses.util import get_logger
+from aws_dataclasses.base import GenericDataClass
+
+LOG = get_logger(__name__)
+
+
+@dataclass
+class LambdaTracingConfig(GenericDataClass):
+    Mode: InitVar[str] = field(repr=False, default=None)
+    mode: str = field(init=False)
+
+    def __post_init__(self, Mode: str):
+        self.mode = Mode
+
+
+@dataclass
+class LambdaVpcConfig(GenericDataClass):
+    SubnetIds: InitVar[List[str]] = field(repr=False, default=None)
+    SecurityGroupIds: InitVar[List[str]] = field(repr=False, default=None)
+    VpcId: InitVar[str] = field(repr=False, default=None)
+    subnet_ids: List[str] = field(init=False)
+    security_group_ids: List[str] = field(init=False)
+    vpc_id: str = field(init=False)
+
+    def __post_init__(self, SubnetIds: List[str], SecurityGroupIds: List[str], VpcId: str):
+        self.subnet_ids = SubnetIds
+        self.security_group_ids = SecurityGroupIds
+        self.vpc_id = VpcId
+
+
+@dataclass
+class LambdaDlqConfig(GenericDataClass):
+    TargetArn: InitVar[str] = field(repr=False, default=None)
+    target_arn: str = field(init=False)
+
+    def __post_init__(self, TargetArn: str):
+        self.target_arn = TargetArn
+
+
+@dataclass
+class LambdaEnvironment(GenericDataClass):
+    Variables: InitVar[Dict] = field(repr=False, default=None)
+    variables: Dict = field(init=False)
+
+    def __post_init__(self, Variables: Dict):
+        self.variables = Variables
+
+
+@dataclass
+class LambdaFunctionConfiguration(GenericDataClass):
+    FunctionName: InitVar[str] = field(repr=False, default=None)
+    FunctionArn: InitVar[str] = field(repr=False, default=None)
+    Runtime: InitVar[str] = field(repr=False, default=None)
+    Role: InitVar[str] = field(repr=False, default=None)
+    Handler: InitVar[str] = field(repr=False, default=None)
+    CodeSize: InitVar[int] = field(repr=False, default=None)
+    Description: InitVar[str] = field(repr=False, default=None)
+    Timeout: InitVar[int] = field(repr=False, default=None)
+    MemorySize: InitVar[int] = field(repr=False, default=None)
+    LastModified: InitVar[str] = field(repr=False, default=None)
+    CodeSha256: InitVar[str] = field(repr=False, default=None)
+    Version: InitVar[str] = field(repr=False, default=None)
+    VpcConfig: InitVar[Any] = field(repr=False, default=None)
+    DeadLetterConfig: InitVar[Any] = field(repr=False, default=None)
+    Environment: InitVar[Any] = field(repr=False, default=None)
+    TracingConfig: InitVar[Any] = field(repr=False, default=None)
+    RevisionId: InitVar[str] = field(repr=False, default=None)
+    function_name: str = field(init=False)
+    function_arn: str = field(init=False)
+    runtime: str = field(init=False)
+    role: str = field(init=False)
+    handler: str = field(init=False)
+    code_size: int = field(init=False)
+    description: str = field(init=False)
+    timeout: int = field(init=False)
+    memory_size: int = field(init=False)
+    last_modified: datetime = field(init=False)
+    code_sha_256: str = field(init=False)
+    version: str = field(init=False)
+    vpc_config: VpcConfig = field(init=False)
+    dead_letter_config: LambdaDlqConfig = field(init=False)
+    environment: LambdaEnvironment = field(init=False)
+    tracing_config: TracingConfig = field(init=False)
+    revision_id: str = field(init=False)
+
+    def __post_init__(self, FunctionName: str, FunctionArn: str, Runtime: str, Role: str, Handler: str, CodeSize: int,
+                      Description: str, Timeout: int, MemorySize: int, LastModified: str, CodeSha256: str, Version: str,
+                      VpcConfig: Any, DeadLetterConfig: Any, Environment: Any, TracingConfig: Any, RevisionId: str):
+        self.last_modified = arrow.get(LastModified).datetime
+        self.function_arn = FunctionArn
+        self.function_name = FunctionName
+        self.runtime = Runtime
+        self.role = Role
+        self.handler = Handler
+        self.code_size = CodeSize
+        self.description = Description
+        self.timeout = Timeout
+        self.memory_size = MemorySize
+        self.code_sha_256 = CodeSha256
+        self.version = Version
+        self.vpc_config = LambdaVpcConfig.from_json(VpcConfig) if VpcConfig is not None else None
+        self.dead_letter_config = LambdaDlqConfig.from_json(DeadLetterConfig) if DeadLetterConfig is not None else None
+        self.environment = LambdaEnvironment.from_json(Environment) if Environment is not None else None
+        self.tracing_config = LambdaTracingConfig.from_json(TracingConfig) if TracingConfig is not None else None
+        self.revision_id = RevisionId

tests/resources/function_configuration.json

@@ -0,0 +1,29 @@
+{
+  "FunctionName": "test",
+  "FunctionArn": "arn:aws:lambda:eu-central-1:0123456789:function:test",
+  "Runtime": "python3.6",
+  "Role": "arn:aws:iam::0123456789:role/test-role",
+  "Handler": "main.lambda_handler",
+  "CodeSize": 30280933,
+  "Description": "I am a description.",
+  "Timeout": 300,
+  "MemorySize": 3008,
+  "LastModified": "2018-09-11T07:45:58.641+0000",
+  "CodeSha256": "ucRYwmrtivyiEicBTcxbPvsmPWw0Sy6LAWuTPvjWX1g=",
+  "Version": "$LATEST",
+  "Environment": {
+    "Variables": {
+      "S3_BUCKET": "testbucket",
+      "STAGE": "prod"
+    }
+  },
+  "VpcConfig": {
+    "SubnetIds": ["a", "b"],
+    "SecurityGroupIds": [],
+    "VpcId": "1234"
+  },
+  "TracingConfig": {
+    "Mode": "PassThrough"
+  },
+  "RevisionId": "beee94f5-9d0a-4d5f-872c-e317c3fb12e9"
+}

tests/test_lambda_list_functions.py

@@ -0,0 +1,44 @@
+import datetime
+
+import pytest
+from dateutil.tz import tzutc
+
+from aws_dataclasses.lambda_function_configuration import LambdaFunctionConfiguration
+
+from .util import get_event_dict
+
+
+@pytest.fixture(scope="module")
+def lambda_functionconfig_raw():
+    return get_event_dict("function_configuration.json")
+
+
+@pytest.fixture(scope="module")
+def lambda_functionconfig(lambda_functionconfig_raw):
+    return LambdaFunctionConfiguration.from_json(lambda_functionconfig_raw)
+
+
+def test_toplevel_items(lambda_functionconfig):
+    cfg = lambda_functionconfig
+    assert cfg.function_name == "test"
+    assert cfg.function_arn == "arn:aws:lambda:eu-central-1:0123456789:function:test"
+    assert cfg.runtime == "python3.6"
+    assert cfg.role == "arn:aws:iam::0123456789:role/test-role"
+    assert cfg.handler == "main.lambda_handler"
+    assert cfg.code_sha_256 == "ucRYwmrtivyiEicBTcxbPvsmPWw0Sy6LAWuTPvjWX1g="
+    assert cfg.timeout == 300
+    assert cfg.code_size == 30280933
+    assert cfg.memory_size == 3008
+    assert cfg.last_modified == datetime.datetime(2018, 9, 11, 7, 45, 58, 641000, tzinfo=tzutc())
+
+
+def test_environment(lambda_functionconfig):
+    env = lambda_functionconfig.environment
+    assert "STAGE" in env.variables
+    assert env.variables["S3_BUCKET"] == "testbucket"
+
+
+def test_vpcconfig(lambda_functionconfig):
+    vpc = lambda_functionconfig.vpc_config
+    assert vpc.vpc_id == "1234"
+    assert len(vpc.subnet_ids) is 2