# Copyright (c) 2014 Skytap http://skytap.com/
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish, dis-
# tribute, sublicense, and/or sell copies of the Software, and to permit
# persons to whom the Software is furnished to do so, subject to the fol-
# lowing conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL-
# ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
# SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
# WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
from tests.unit import mock, unittest
from tests.unit import AWSMockServiceTestCase
from boto.vpc import VpcPeeringConnection, VPCConnection, Subnet
class TestDescribeVpcPeeringConnections(AWSMockServiceTestCase):
DESCRIBE_VPC_PEERING_CONNECTIONS= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
-
pcx-111aaa22
777788889999
vpc-1a2b3c4d
172.31.0.0/16
111122223333
vpc-aa22cc33
pending-acceptance
Pending Acceptance by 111122223333
2014-02-17T16:00:50.000Z
-
pcx-444bbb88
1237897234
vpc-2398abcd
172.30.0.0/16
98654313
vpc-0983bcda
pending-acceptance
Pending Acceptance by 98654313
2015-02-17T16:00:50.000Z
"""
connection_class = VPCConnection
def default_body(self):
return self.DESCRIBE_VPC_PEERING_CONNECTIONS
def test_get_vpc_peering_connections(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.get_all_vpc_peering_connections(
['pcx-111aaa22', 'pcx-444bbb88'], filters=[('status-code', ['pending-acceptance'])])
self.assertEqual(len(api_response), 2)
for vpc_peering_connection in api_response:
if vpc_peering_connection.id == 'pcx-111aaa22':
self.assertEqual(vpc_peering_connection.id, 'pcx-111aaa22')
self.assertEqual(vpc_peering_connection.status_code, 'pending-acceptance')
self.assertEqual(vpc_peering_connection.status_message, 'Pending Acceptance by 111122223333')
self.assertEqual(vpc_peering_connection.requester_vpc_info.owner_id, '777788889999')
self.assertEqual(vpc_peering_connection.requester_vpc_info.vpc_id, 'vpc-1a2b3c4d')
self.assertEqual(vpc_peering_connection.requester_vpc_info.cidr_block, '172.31.0.0/16')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.owner_id, '111122223333')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.vpc_id, 'vpc-aa22cc33')
self.assertEqual(vpc_peering_connection.expiration_time, '2014-02-17T16:00:50.000Z')
else:
self.assertEqual(vpc_peering_connection.id, 'pcx-444bbb88')
self.assertEqual(vpc_peering_connection.status_code, 'pending-acceptance')
self.assertEqual(vpc_peering_connection.status_message, 'Pending Acceptance by 98654313')
self.assertEqual(vpc_peering_connection.requester_vpc_info.owner_id, '1237897234')
self.assertEqual(vpc_peering_connection.requester_vpc_info.vpc_id, 'vpc-2398abcd')
self.assertEqual(vpc_peering_connection.requester_vpc_info.cidr_block, '172.30.0.0/16')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.owner_id, '98654313')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.vpc_id, 'vpc-0983bcda')
self.assertEqual(vpc_peering_connection.expiration_time, '2015-02-17T16:00:50.000Z')
class TestCreateVpcPeeringConnection(AWSMockServiceTestCase):
CREATE_VPC_PEERING_CONNECTION= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
pcx-73a5401a
777788889999
vpc-1a2b3c4d
10.0.0.0/28
123456789012
vpc-a1b2c3d4
initiating-request
Initiating Request to 123456789012
2014-02-18T14:37:25.000Z
"""
connection_class = VPCConnection
def default_body(self):
return self.CREATE_VPC_PEERING_CONNECTION
def test_create_vpc_peering_connection(self):
self.set_http_response(status_code=200)
vpc_peering_connection = self.service_connection.create_vpc_peering_connection('vpc-1a2b3c4d', 'vpc-a1b2c3d4', '123456789012')
self.assertEqual(vpc_peering_connection.id, 'pcx-73a5401a')
self.assertEqual(vpc_peering_connection.status_code, 'initiating-request')
self.assertEqual(vpc_peering_connection.status_message, 'Initiating Request to 123456789012')
self.assertEqual(vpc_peering_connection.requester_vpc_info.owner_id, '777788889999')
self.assertEqual(vpc_peering_connection.requester_vpc_info.vpc_id, 'vpc-1a2b3c4d')
self.assertEqual(vpc_peering_connection.requester_vpc_info.cidr_block, '10.0.0.0/28')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.owner_id, '123456789012')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.vpc_id, 'vpc-a1b2c3d4')
self.assertEqual(vpc_peering_connection.expiration_time, '2014-02-18T14:37:25.000Z')
class TestDeleteVpcPeeringConnection(AWSMockServiceTestCase):
DELETE_VPC_PEERING_CONNECTION= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
true
"""
connection_class = VPCConnection
def default_body(self):
return self.DELETE_VPC_PEERING_CONNECTION
def test_delete_vpc_peering_connection(self):
self.set_http_response(status_code=200)
self.assertEquals(self.service_connection.delete_vpc_peering_connection('pcx-12345678'), True)
class TestDeleteVpcPeeringConnectionShortForm(unittest.TestCase):
DESCRIBE_VPC_PEERING_CONNECTIONS= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
-
pcx-111aaa22
777788889999
vpc-1a2b3c4d
172.31.0.0/16
111122223333
vpc-aa22cc33
pending-acceptance
Pending Acceptance by 111122223333
2014-02-17T16:00:50.000Z
"""
DELETE_VPC_PEERING_CONNECTION= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
true
"""
def test_delete_vpc_peering_connection(self):
vpc_conn = VPCConnection(aws_access_key_id='aws_access_key_id',
aws_secret_access_key='aws_secret_access_key')
mock_response = mock.Mock()
mock_response.read.return_value = self.DESCRIBE_VPC_PEERING_CONNECTIONS
mock_response.status = 200
vpc_conn.make_request = mock.Mock(return_value=mock_response)
vpc_peering_connections = vpc_conn.get_all_vpc_peering_connections()
self.assertEquals(1, len(vpc_peering_connections))
vpc_peering_connection = vpc_peering_connections[0]
mock_response = mock.Mock()
mock_response.read.return_value = self.DELETE_VPC_PEERING_CONNECTION
mock_response.status = 200
vpc_conn.make_request = mock.Mock(return_value=mock_response)
self.assertEquals(True, vpc_peering_connection.delete())
self.assertIn('DeleteVpcPeeringConnection', vpc_conn.make_request.call_args_list[0][0])
self.assertNotIn('DeleteVpc', vpc_conn.make_request.call_args_list[0][0])
class TestRejectVpcPeeringConnection(AWSMockServiceTestCase):
REJECT_VPC_PEERING_CONNECTION= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
true
"""
connection_class = VPCConnection
def default_body(self):
return self.REJECT_VPC_PEERING_CONNECTION
def test_reject_vpc_peering_connection(self):
self.set_http_response(status_code=200)
self.assertEquals(self.service_connection.reject_vpc_peering_connection('pcx-12345678'), True)
class TestAcceptVpcPeeringConnection(AWSMockServiceTestCase):
ACCEPT_VPC_PEERING_CONNECTION= b"""
7a62c49f-347e-4fc4-9331-6e8eEXAMPLE
pcx-1a2b3c4d
123456789012
vpc-1a2b3c4d
10.0.0.0/28
777788889999
vpc-111aaa22
10.0.1.0/28
active
Active
"""
connection_class = VPCConnection
def default_body(self):
return self.ACCEPT_VPC_PEERING_CONNECTION
def test_accept_vpc_peering_connection(self):
self.set_http_response(status_code=200)
vpc_peering_connection = self.service_connection.accept_vpc_peering_connection('pcx-1a2b3c4d')
self.assertEqual(vpc_peering_connection.id, 'pcx-1a2b3c4d')
self.assertEqual(vpc_peering_connection.status_code, 'active')
self.assertEqual(vpc_peering_connection.status_message, 'Active')
self.assertEqual(vpc_peering_connection.requester_vpc_info.owner_id, '123456789012')
self.assertEqual(vpc_peering_connection.requester_vpc_info.vpc_id, 'vpc-1a2b3c4d')
self.assertEqual(vpc_peering_connection.requester_vpc_info.cidr_block, '10.0.0.0/28')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.owner_id, '777788889999')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.vpc_id, 'vpc-111aaa22')
self.assertEqual(vpc_peering_connection.accepter_vpc_info.cidr_block, '10.0.1.0/28')
if __name__ == '__main__':
unittest.main()