File: test_customer_gateways.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (112 lines) | stat: -rw-r--r-- 3,746 bytes parent folder | download
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
import boto3
import pytest
from botocore.exceptions import ClientError

from moto import mock_aws
from tests.test_ec2 import ec2_aws_verified

from .helpers import assert_dryrun_error


@mock_aws
def test_create_customer_gateways():
    ec2 = boto3.client("ec2", region_name="us-east-1")

    customer_gateway = create_customer_gateway(ec2)
    assert customer_gateway["CustomerGatewayId"].startswith("cgw-")
    assert customer_gateway["Type"] == "ipsec.1"
    assert customer_gateway["BgpAsn"] == "65534"
    assert customer_gateway["IpAddress"] == "205.251.242.54"


@mock_aws
def test_create_customer_gateways_using_publicip_argument():
    ec2 = boto3.client("ec2", region_name="us-east-1")

    # The PublicIp-argument is deprecated, but should still be supported by Moto
    # https://github.com/boto/botocore/commit/86202c8698cf77fb6ecccfdbc05bbc218e861d14#diff-c51449716bfc26c1eac92ec403b470827d2dcba1126cf303567074b872d5c592
    customer_gateway = ec2.create_customer_gateway(
        Type="ipsec.1", IpAddress="205.251.242.53", BgpAsn=65534
    )["CustomerGateway"]
    customer_gateway["CustomerGatewayId"].startswith("cgw-")
    assert customer_gateway["Type"] == "ipsec.1"
    assert customer_gateway["BgpAsn"] == "65534"
    assert customer_gateway["IpAddress"] == "205.251.242.53"


@mock_aws
def test_describe_customer_gateways_dryrun():
    client = boto3.client("ec2", region_name="us-east-1")

    with pytest.raises(ClientError) as ex:
        client.describe_customer_gateways(DryRun=True)
    assert_dryrun_error(ex)


@mock_aws
def test_describe_customer_gateways():
    ec2 = boto3.client("ec2", region_name="us-east-1")

    customer_gateway = create_customer_gateway(ec2)
    cg_id = customer_gateway["CustomerGatewayId"]

    cgws = ec2.describe_customer_gateways()["CustomerGateways"]
    cg_ids = [cg["CustomerGatewayId"] for cg in cgws]
    assert cg_id in cg_ids

    cgw = ec2.describe_customer_gateways(CustomerGatewayIds=[cg_id])[
        "CustomerGateways"
    ][0]
    assert "BgpAsn" in cgw
    assert cgw["CustomerGatewayId"] == cg_id
    assert "IpAddress" in cgw
    assert cgw["State"] == "available"
    assert cgw["Type"] == "ipsec.1"

    all_cgws = ec2.describe_customer_gateways()["CustomerGateways"]
    assert len(all_cgws) >= 1, (
        "Should have at least the one CustomerGateway we just created"
    )


@mock_aws
def test_delete_customer_gateways():
    ec2 = boto3.client("ec2", region_name="us-east-1")

    customer_gateway = create_customer_gateway(ec2)
    cg_id = customer_gateway["CustomerGatewayId"]

    cgws = ec2.describe_customer_gateways(CustomerGatewayIds=[cg_id])[
        "CustomerGateways"
    ]
    assert len(cgws) == 1
    assert cgws[0]["State"] == "available"

    ec2.delete_customer_gateway(CustomerGatewayId=customer_gateway["CustomerGatewayId"])

    cgws = ec2.describe_customer_gateways(CustomerGatewayIds=[cg_id])[
        "CustomerGateways"
    ]
    assert len(cgws) == 1
    assert cgws[0]["State"] == "deleted"


@ec2_aws_verified()
@pytest.mark.aws_verified
def test_delete_customer_gateways_bad_id(ec2_client=None):
    with pytest.raises(ClientError) as exc:
        ec2_client.delete_customer_gateway(CustomerGatewayId="cgw-0123abcd")
    response = exc.value.response
    assert response["ResponseMetadata"]["HTTPStatusCode"] == 400
    assert "RequestId" in response["ResponseMetadata"]
    assert response["Error"]["Code"] == "InvalidCustomerGatewayID.NotFound"
    assert (
        response["Error"]["Message"]
        == "The customerGateway ID 'cgw-0123abcd' does not exist"
    )


def create_customer_gateway(ec2):
    return ec2.create_customer_gateway(
        Type="ipsec.1", PublicIp="205.251.242.54", BgpAsn=65534
    )["CustomerGateway"]