powertools-lambda-python/tests/unit/parser/test_vpc_latticev2.py at develop · ryanpudd/powertools-lambda-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import pytest

from aws_lambda_powertools.utilities.parser import ValidationError, envelopes, parse
from aws_lambda_powertools.utilities.parser.models import VpcLatticeV2Model
from tests.functional.utils import load_event
from tests.unit.parser.schemas import MyVpcLatticeBusiness


def test_vpc_lattice_v2_event_with_envelope():
    raw_event = load_event("vpcLatticeV2Event.json")
    raw_event["body"] = '{"username": "Stephen", "name": "Bawks"}'
    parsed_event: MyVpcLatticeBusiness = parse(
        event=raw_event,
        model=MyVpcLatticeBusiness,
        envelope=envelopes.VpcLatticeV2Envelope,
    )

    assert parsed_event.username == "Stephen"
    assert parsed_event.name == "Bawks"


def test_vpc_lattice_v2_event():
    raw_event = load_event("vpcLatticeV2Event.json")
    model = VpcLatticeV2Model(**raw_event)

    assert model.body == raw_event["body"]
    assert model.method == raw_event["method"]
    assert model.path == raw_event["path"]
    assert model.is_base64_encoded == raw_event["isBase64Encoded"]
    assert model.headers == raw_event["headers"]
    assert model.query_string_parameters == raw_event["queryStringParameters"]
    assert model.request_context.region == raw_event["requestContext"]["region"]
    assert model.request_context.service_network_arn == raw_event["requestContext"]["serviceNetworkArn"]
    assert model.request_context.service_arn == raw_event["requestContext"]["serviceArn"]
    assert model.request_context.target_group_arn == raw_event["requestContext"]["targetGroupArn"]
    assert model.request_context.time_epoch == float(raw_event["requestContext"]["timeEpoch"])
    convert_time = int((model.request_context.time_epoch_as_datetime.timestamp() * 1000))
    event_converted_time = round(int(raw_event["requestContext"]["timeEpoch"]) / 1000)
    assert convert_time == event_converted_time
    assert model.request_context.identity.source_vpc_arn == raw_event["requestContext"]["identity"]["sourceVpcArn"]
    assert model.request_context.identity.get_type == raw_event["requestContext"]["identity"]["type"]
    assert model.request_context.identity.principal == raw_event["requestContext"]["identity"]["principal"]
    assert model.request_context.identity.session_name == raw_event["requestContext"]["identity"]["sessionName"]
    assert model.request_context.identity.x509_san_dns == raw_event["requestContext"]["identity"]["x509SanDns"]
    assert model.request_context.identity.x509_issuer_ou is None
    assert model.request_context.identity.x509_san_name_cn is None
    assert model.request_context.identity.x509_san_uri is None
    assert model.request_context.identity.x509_subject_cn is None
    assert model.request_context.identity.principal_org_id is None


def test_vpc_lattice_v2_event_custom_model():
    class MyCustomResource(VpcLatticeV2Model):
        body: str

    raw_event = load_event("vpcLatticeV2Event.json")
    model = MyCustomResource(**raw_event)

    assert model.body == raw_event["body"]


def test_vpc_lattice_v2_event_invalid():
    raw_event = load_event("vpcLatticeV2Event.json")
    raw_event["body"] = ["some_more_data"]

    with pytest.raises(ValidationError):
        VpcLatticeV2Model(**raw_event)