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