1import boto
2import boto3
3import pytest
4import sure  # noqa # pylint: disable=unused-import
5
6from moto import mock_ec2_deprecated, mock_ec2
7from botocore.exceptions import ClientError
8from .test_tags import retrieve_all_tagged
9
10
11@mock_ec2
12def test_attach_unknown_vpn_gateway():
13    """describe_vpn_gateways attachment.vpc-id filter"""
14
15    ec2 = boto3.client("ec2", region_name="us-east-1")
16
17    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
18
19    with pytest.raises(ClientError) as ex:
20        ec2.attach_vpn_gateway(VpcId=vpc["VpcId"], VpnGatewayId="?")
21    err = ex.value.response["Error"]
22    err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
23    err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
24
25
26@mock_ec2
27def test_delete_unknown_vpn_gateway():
28    """describe_vpn_gateways attachment.vpc-id filter"""
29
30    ec2 = boto3.client("ec2", region_name="us-east-1")
31
32    with pytest.raises(ClientError) as ex:
33        ec2.delete_vpn_gateway(VpnGatewayId="?")
34    err = ex.value.response["Error"]
35    err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
36    err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
37
38
39@mock_ec2
40def test_detach_unknown_vpn_gateway():
41    """describe_vpn_gateways attachment.vpc-id filter"""
42
43    ec2 = boto3.client("ec2", region_name="us-east-1")
44
45    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
46
47    with pytest.raises(ClientError) as ex:
48        ec2.detach_vpn_gateway(VpcId=vpc["VpcId"], VpnGatewayId="?")
49    err = ex.value.response["Error"]
50    err["Message"].should.equal("The virtual private gateway ID '?' does not exist")
51    err["Code"].should.equal("InvalidVpnGatewayID.NotFound")
52
53
54@mock_ec2
55def test_describe_vpn_connections_attachment_vpc_id_filter():
56    """describe_vpn_gateways attachment.vpc-id filter"""
57
58    ec2 = boto3.client("ec2", region_name="us-east-1")
59
60    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
61    vpc_id = vpc["Vpc"]["VpcId"]
62    gateway = ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
63    gateway_id = gateway["VpnGateway"]["VpnGatewayId"]
64
65    ec2.attach_vpn_gateway(VpcId=vpc_id, VpnGatewayId=gateway_id)
66
67    gateways = ec2.describe_vpn_gateways(
68        Filters=[{"Name": "attachment.vpc-id", "Values": [vpc_id]}]
69    )
70
71    gateways["VpnGateways"].should.have.length_of(1)
72    gateways["VpnGateways"][0]["VpnGatewayId"].should.equal(gateway_id)
73    gateways["VpnGateways"][0]["VpcAttachments"].should.contain(
74        {"State": "attached", "VpcId": vpc_id}
75    )
76
77
78@mock_ec2
79def test_describe_vpn_connections_state_filter_attached():
80    """describe_vpn_gateways attachment.state filter - match attached"""
81
82    ec2 = boto3.client("ec2", region_name="us-east-1")
83
84    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
85    vpc_id = vpc["Vpc"]["VpcId"]
86    gateway = ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
87    gateway_id = gateway["VpnGateway"]["VpnGatewayId"]
88
89    ec2.attach_vpn_gateway(VpcId=vpc_id, VpnGatewayId=gateway_id)
90
91    all_gateways = retrieve_all(
92        ec2, [{"Name": "attachment.state", "Values": ["attached"]}]
93    )
94
95    [gw["VpnGatewayId"] for gw in all_gateways].should.contain(gateway_id)
96    my_gateway = [gw for gw in all_gateways if gw["VpnGatewayId"] == gateway_id][0]
97    my_gateway["VpcAttachments"].should.contain({"State": "attached", "VpcId": vpc_id})
98
99
100@mock_ec2
101def test_describe_vpn_connections_state_filter_deatched():
102    """describe_vpn_gateways attachment.state filter - don't match detatched"""
103
104    ec2 = boto3.client("ec2", region_name="us-east-1")
105
106    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
107    vpc_id = vpc["Vpc"]["VpcId"]
108    gateway = ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
109    gateway_id = gateway["VpnGateway"]["VpnGatewayId"]
110
111    ec2.attach_vpn_gateway(VpcId=vpc_id, VpnGatewayId=gateway_id)
112
113    gateways = ec2.describe_vpn_gateways(
114        Filters=[{"Name": "attachment.state", "Values": ["detached"]}]
115    )
116
117    gateways["VpnGateways"].should.have.length_of(0)
118
119
120@mock_ec2
121def test_describe_vpn_connections_id_filter_match():
122    """describe_vpn_gateways vpn-gateway-id filter - match correct id"""
123
124    ec2 = boto3.client("ec2", region_name="us-east-1")
125
126    gateway = ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
127    gateway_id = gateway["VpnGateway"]["VpnGatewayId"]
128
129    gateways = ec2.describe_vpn_gateways(
130        Filters=[{"Name": "vpn-gateway-id", "Values": [gateway_id]}]
131    )
132
133    gateways["VpnGateways"].should.have.length_of(1)
134    gateways["VpnGateways"][0]["VpnGatewayId"].should.equal(gateway_id)
135
136
137@mock_ec2
138def test_describe_vpn_connections_id_filter_miss():
139    """describe_vpn_gateways vpn-gateway-id filter - don't match"""
140
141    ec2 = boto3.client("ec2", region_name="us-east-1")
142
143    ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
144
145    gateways = ec2.describe_vpn_gateways(
146        Filters=[{"Name": "vpn-gateway-id", "Values": ["unknown_gateway_id"]}]
147    )
148
149    gateways["VpnGateways"].should.have.length_of(0)
150
151
152@mock_ec2
153def test_describe_vpn_connections_type_filter_match():
154    """describe_vpn_gateways type filter - match"""
155
156    ec2 = boto3.client("ec2", region_name="us-east-1")
157
158    gateway = ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
159    gateway_id = gateway["VpnGateway"]["VpnGatewayId"]
160
161    my_gateways = retrieve_all(ec2, [{"Name": "type", "Values": ["ipsec.1"]}])
162
163    [gw["VpnGatewayId"] for gw in my_gateways].should.contain(gateway_id)
164
165
166@mock_ec2
167def test_describe_vpn_connections_type_filter_miss():
168    """describe_vpn_gateways type filter - don't match"""
169
170    ec2 = boto3.client("ec2", region_name="us-east-1")
171
172    ec2.create_vpn_gateway(AvailabilityZone="us-east-1a", Type="ipsec.1")
173
174    gateways = ec2.describe_vpn_gateways(
175        Filters=[{"Name": "type", "Values": ["unknown_type"]}]
176    )
177
178    gateways["VpnGateways"].should.have.length_of(0)
179
180
181# Has boto3 equivalent
182@mock_ec2_deprecated
183def test_virtual_private_gateways():
184    conn = boto.connect_vpc("the_key", "the_secret")
185
186    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
187    vpn_gateway.should_not.be.none
188    vpn_gateway.id.should.match(r"vgw-\w+")
189    vpn_gateway.type.should.equal("ipsec.1")
190    vpn_gateway.state.should.equal("available")
191    vpn_gateway.availability_zone.should.equal("us-east-1a")
192
193
194@mock_ec2
195def test_virtual_private_gateways_boto3():
196    client = boto3.client("ec2", region_name="us-west-1")
197
198    vpn_gateway = client.create_vpn_gateway(
199        Type="ipsec.1", AvailabilityZone="us-east-1a"
200    )["VpnGateway"]
201
202    vpn_gateway["VpnGatewayId"].should.match(r"vgw-\w+")
203    vpn_gateway["Type"].should.equal("ipsec.1")
204    vpn_gateway["State"].should.equal("available")
205    vpn_gateway["AvailabilityZone"].should.equal("us-east-1a")
206
207
208# Has boto3 equivalent
209@mock_ec2_deprecated
210def test_describe_vpn_gateway():
211    conn = boto.connect_vpc("the_key", "the_secret")
212    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
213
214    vgws = conn.get_all_vpn_gateways()
215    vgws.should.have.length_of(1)
216
217    gateway = vgws[0]
218    gateway.id.should.match(r"vgw-\w+")
219    gateway.id.should.equal(vpn_gateway.id)
220    vpn_gateway.type.should.equal("ipsec.1")
221    vpn_gateway.state.should.equal("available")
222    vpn_gateway.availability_zone.should.equal("us-east-1a")
223
224
225@mock_ec2
226def test_describe_vpn_gateway_boto3():
227    client = boto3.client("ec2", region_name="us-west-1")
228    vpn_gateway = client.create_vpn_gateway(
229        Type="ipsec.1", AvailabilityZone="us-east-1a"
230    )["VpnGateway"]
231
232    vgws = client.describe_vpn_gateways(VpnGatewayIds=[vpn_gateway["VpnGatewayId"]])[
233        "VpnGateways"
234    ]
235    vgws.should.have.length_of(1)
236
237    gateway = vgws[0]
238    gateway["VpnGatewayId"].should.match(r"vgw-\w+")
239    gateway["VpnGatewayId"].should.equal(vpn_gateway["VpnGatewayId"])
240    # TODO: fixme. This currently returns the ID
241    # gateway["Type"].should.equal("ipsec.1")
242    gateway["State"].should.equal("available")
243    gateway["AvailabilityZone"].should.equal("us-east-1a")
244
245
246# Has boto3 equivalent
247@mock_ec2_deprecated
248def test_vpn_gateway_vpc_attachment():
249    conn = boto.connect_vpc("the_key", "the_secret")
250    vpc = conn.create_vpc("10.0.0.0/16")
251    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
252
253    conn.attach_vpn_gateway(vpn_gateway_id=vpn_gateway.id, vpc_id=vpc.id)
254
255    gateway = conn.get_all_vpn_gateways()[0]
256    attachments = gateway.attachments
257    attachments.should.have.length_of(1)
258    attachments[0].vpc_id.should.equal(vpc.id)
259    attachments[0].state.should.equal("attached")
260
261
262@mock_ec2
263def test_vpn_gateway_vpc_attachment_boto3():
264    ec2 = boto3.resource("ec2", region_name="us-west-1")
265    client = boto3.client("ec2", region_name="us-west-1")
266    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
267    vpn_gateway = client.create_vpn_gateway(
268        Type="ipsec.1", AvailabilityZone="us-east-1a"
269    )["VpnGateway"]
270    vpng_id = vpn_gateway["VpnGatewayId"]
271
272    client.attach_vpn_gateway(VpnGatewayId=vpng_id, VpcId=vpc.id)
273
274    gateway = client.describe_vpn_gateways(VpnGatewayIds=[vpng_id])["VpnGateways"][0]
275    attachments = gateway["VpcAttachments"]
276    attachments.should.equal([{"State": "attached", "VpcId": vpc.id}])
277
278
279# Has boto3 equivalent
280@mock_ec2_deprecated
281def test_delete_vpn_gateway():
282    conn = boto.connect_vpc("the_key", "the_secret")
283    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
284
285    conn.delete_vpn_gateway(vpn_gateway.id)
286    vgws = conn.get_all_vpn_gateways()
287    vgws.should.have.length_of(1)
288    vgws[0].state.should.equal("deleted")
289
290
291@mock_ec2
292def test_delete_vpn_gateway_boto3():
293    client = boto3.client("ec2", region_name="us-west-1")
294    vpn_gateway = client.create_vpn_gateway(
295        Type="ipsec.1", AvailabilityZone="us-east-1a"
296    )["VpnGateway"]
297    vpng_id = vpn_gateway["VpnGatewayId"]
298
299    client.delete_vpn_gateway(VpnGatewayId=vpng_id)
300    gateways = client.describe_vpn_gateways(VpnGatewayIds=[vpng_id])["VpnGateways"]
301    gateways.should.have.length_of(1)
302    gateways[0].should.have.key("State").equal("deleted")
303
304
305# Has boto3 equivalent
306@mock_ec2_deprecated
307def test_vpn_gateway_tagging():
308    conn = boto.connect_vpc("the_key", "the_secret")
309    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
310    vpn_gateway.add_tag("a key", "some value")
311
312    tag = conn.get_all_tags()[0]
313    tag.name.should.equal("a key")
314    tag.value.should.equal("some value")
315
316    # Refresh the subnet
317    vpn_gateway = conn.get_all_vpn_gateways()[0]
318    vpn_gateway.tags.should.have.length_of(1)
319    vpn_gateway.tags["a key"].should.equal("some value")
320
321
322@mock_ec2
323def test_vpn_gateway_tagging_boto3():
324    client = boto3.client("ec2", region_name="us-west-1")
325    vpn_gateway = client.create_vpn_gateway(
326        Type="ipsec.1", AvailabilityZone="us-east-1a"
327    )["VpnGateway"]
328    client.create_tags(
329        Resources=[vpn_gateway["VpnGatewayId"]],
330        Tags=[{"Key": "a key", "Value": "some value"}],
331    )
332
333    all_tags = retrieve_all_tagged(client)
334    ours = [a for a in all_tags if a["ResourceId"] == vpn_gateway["VpnGatewayId"]][0]
335    ours.should.have.key("Key").equal("a key")
336    ours.should.have.key("Value").equal("some value")
337
338    vpn_gateway = client.describe_vpn_gateways()["VpnGateways"][0]
339    # TODO: Fixme: Tags is currently empty
340    # vpn_gateway["Tags"].should.equal([{'Key': 'a key', 'Value': 'some value'}])
341
342
343# Has boto3 equivalent
344@mock_ec2_deprecated
345def test_detach_vpn_gateway():
346
347    conn = boto.connect_vpc("the_key", "the_secret")
348    vpc = conn.create_vpc("10.0.0.0/16")
349    vpn_gateway = conn.create_vpn_gateway("ipsec.1", "us-east-1a")
350
351    conn.attach_vpn_gateway(vpn_gateway_id=vpn_gateway.id, vpc_id=vpc.id)
352
353    gateway = conn.get_all_vpn_gateways()[0]
354    attachments = gateway.attachments
355    attachments.should.have.length_of(1)
356    attachments[0].vpc_id.should.equal(vpc.id)
357    attachments[0].state.should.equal("attached")
358
359    conn.detach_vpn_gateway(vpn_gateway_id=vpn_gateway.id, vpc_id=vpc.id)
360
361    gateway = conn.get_all_vpn_gateways()[0]
362    attachments = gateway.attachments
363    attachments.should.have.length_of(1)
364    attachments[0].state.should.equal("detached")
365
366
367@mock_ec2
368def test_detach_vpn_gateway_boto3():
369    ec2 = boto3.resource("ec2", region_name="us-west-1")
370    client = boto3.client("ec2", region_name="us-west-1")
371
372    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
373    vpn_gateway = client.create_vpn_gateway(
374        Type="ipsec.1", AvailabilityZone="us-east-1a"
375    )
376    vpn_gateway = vpn_gateway["VpnGateway"]
377    vpng_id = vpn_gateway["VpnGatewayId"]
378
379    client.attach_vpn_gateway(VpnGatewayId=vpng_id, VpcId=vpc.id)
380
381    gateway = client.describe_vpn_gateways(VpnGatewayIds=[vpng_id])["VpnGateways"][0]
382    attachments = gateway["VpcAttachments"]
383    attachments.should.equal([{"State": "attached", "VpcId": vpc.id}])
384
385    client.detach_vpn_gateway(VpnGatewayId=vpng_id, VpcId=vpc.id)
386
387    gateway = client.describe_vpn_gateways(VpnGatewayIds=[vpng_id])["VpnGateways"][0]
388    attachments = gateway["VpcAttachments"]
389    attachments.should.equal([{"State": "detached", "VpcId": vpc.id}])
390
391
392def retrieve_all(client, filters=[]):  # pylint: disable=W0102
393    resp = client.describe_vpn_gateways(Filters=filters)
394    all_gateways = resp["VpnGateways"]
395    token = resp.get("NextToken")
396    while token:
397        resp = client.describe_vpn_gateways(Filters=filters)
398        all_gateways.extend(resp["VpnGateways"])
399        token = resp.get("NextToken")
400    return all_gateways
401