# Copyright (c) 2014 Rackspace, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or # implied. # See the License for the specific language governing permissions and # limitations under the License. import random import uuid import ddt import fastly import mock from poppy.provider.fastly import services from poppy.transport.pecan.models.request import service from tests.unit import base @ddt.ddt class TestServices(base.TestCase): @mock.patch('fastly.FastlyConnection') @mock.patch('fastly.FastlyService') @mock.patch('fastly.FastlyVersion') @mock.patch('poppy.provider.fastly.services.ServiceController.client') @mock.patch('poppy.provider.fastly.driver.CDNProvider') def setUp(self, mock_connection, mock_service, mock_version, mock_controllerclient, mock_driver): super(TestServices, self).setUp() self.driver = mock_driver() self.driver.provider_name = 'Fastly' self.mock_service = mock_service self.mock_version = mock_version self.service_name = 'scarborough' mock_service_id = '%020x' % random.randrange(16 ** 20) self.mock_create_version_resp = { u'service_id': mock_service_id, u'number': 5} self.service_instance = self.mock_service() self.service_instance.id = mock_service_id self.controller = services.ServiceController(self.driver) self.version = self.mock_version( self.controller.client, self.mock_create_version_resp) self.controller.client.create_service.return_value = ( self.service_instance) self.controller.client.create_version.return_value = self.version @ddt.file_data('data_service.json') def test_create_with_create_service_exception(self, service_json): # ASSERTIONS # create_service service_obj = service.load_from_json(service_json) self.controller.client.create_service.side_effect = fastly.FastlyError( Exception('Creating service failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_create_version_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_service.side_effect = None service_obj = service.load_from_json(service_json) # create_version self.controller.client.create_version.side_effect = fastly.FastlyError( Exception('Creating version failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_create_domain_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_version.side_effect = None service_obj = service.load_from_json(service_json) # create domains self.controller.client.create_domain.side_effect = fastly.FastlyError( Exception('Creating domains failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_create_backend_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_domain.side_effect = None service_obj = service.load_from_json(service_json) # create backends self.controller.client.create_backend.side_effect = fastly.FastlyError( Exception('Creating backend failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_check_domains_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_backend.side_effect = None service_obj = service.load_from_json(service_json) self.controller.client.check_domains.side_effect = fastly.FastlyError( Exception('Check_domains failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_list_versions_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_backend.side_effect = None service_obj = service.load_from_json(service_json) self.controller.client.list_versions.side_effect = fastly.FastlyError( Exception('List_versions failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_activate_version_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.create_backend.side_effect = None service_obj = service.load_from_json(service_json) self.controller.client.active_version.side_effect = fastly.FastlyError( Exception('Active_version failed.')) resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create_with_general_exception(self, service_json): self.controller.client.reset_mock() self.controller.client.check_domains.side_effect = None service_obj = service.load_from_json(service_json) # test a general exception self.controller.client.create_service.side_effect = Exception( 'Wild exception occurred.') resp = self.controller.create(service_obj) self.assertIn('error', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_create(self, service_json): # instantiate # this case needs to set all return value for each call service_obj = service.load_from_json(service_json) controller = services.ServiceController(self.driver) controller.client.create_service.return_value = self.service_instance controller.client.create_version.return_value = self.version controller.client.list_versions.return_value = [self.version] controller.client.active_version.return_value = self.version fastly_fake_domain_check = type( 'FastlyDomain', (object,), { 'name': 'fake_domain.global.prod.fastly.net'}) controller.client.check_domains.return_value = [ mock.Mock(domain=fastly_fake_domain_check) ] resp = controller.create(service_obj) controller.client.create_service.assert_called_once_with( controller.current_customer.id, service_obj.name) controller.client.create_version.assert_called_once_with( self.service_instance.id) controller.client.create_domain.assert_any_call( self.service_instance.id, self.version.number, service_obj.domains[0].domain) controller.client.create_domain.assert_any_call( self.service_instance.id, self.version.number, service_obj.domains[1].domain) controller.client.check_domains.assert_called_once_with( self.service_instance.id, self.version.number) controller.client.create_backend.assert_has_any_call( self.service_instance.id, 1, service_obj.origins[0].origin.replace(":", "-"), service_obj.origins[0].origin, service_obj.origins[0].ssl, service_obj.origins[0].port) self.assertIn('links', resp[self.driver.provider_name]) def test_delete_with_exception(self): provider_service_id = uuid.uuid1() # instantiate controller = services.ServiceController(self.driver) # mock.patch return values service = self.mock_service() service.id = uuid.uuid1() controller.client.get_service_by_name.return_value = service # test exception exception = fastly.FastlyError(Exception('ding')) controller.client.delete_service.side_effect = exception resp = controller.delete(provider_service_id) self.assertIn('error', resp[self.driver.provider_name]) def test_delete(self): provider_service_id = uuid.uuid1() # instantiate controller = services.ServiceController(self.driver) # clear run controller.client.delete_service.side_effect = None resp = controller.delete(provider_service_id) controller.client.delete_service.assert_called_once_with( provider_service_id) self.assertIn('id', resp[self.driver.provider_name]) @ddt.file_data('data_service.json') def test_update(self, service_json): provider_service_id = uuid.uuid1() controller = services.ServiceController(self.driver) resp = controller.update(provider_service_id, service_json) self.assertIn('id', resp[self.driver.provider_name]) def test_client(self): controller = services.ServiceController(self.driver) self.assertNotEqual(controller.client(), None) class TestProviderValidation(base.TestCase): """Tests for provider side validation methods. This class tests the validation methods to verify the data accuracy at the provider side. """ @mock.patch('poppy.provider.fastly.services.ServiceController.client') @mock.patch('poppy.provider.fastly.driver.CDNProvider') def setUp(self, mock_client, mock_driver): super(TestProviderValidation, self).setUp() self.driver = mock_driver() self.controller = services.ServiceController(self.driver) self.client = mock_client self.service_name = uuid.uuid1() service_json = {"name": "mocksite.com", "domains": [{"domain": "parsely.sage.com"}], "origins": [{"origin": "mockdomain.com", "ssl": False, "port": 80}], "flavorRef": "standard"} service_obj = service.load_from_json(service_json) self.controller.create(service_obj) @mock.patch('fastly.FastlyService') def test_get_service_details(self, mock_service): service = mock_service() service.id = uuid.uuid1() service.version = [{'number': 1}] self.controller.client.get_service_by_name.return_value = service self.controller.client.list_domains.return_value = ( [{u'comment': u'A new domain.', u'service_id': u'75hthkv6jrGW4aVjSReT0w', u'version': 1, u'name': u'www.testr.com'}]) self.controller.client.list_cache_settings.return_value = ( [{u'stale_ttl': u'0', u'name': u'test-cache', u'ttl': u'3600', u'version': u'1', u'cache_condition': '', u'action': '', u'service_id': u'75hthkv6jrGW4aVjSReT0w'}]) self.controller.client.list_backends.return_value = ( [{u'comment': '', u'shield': None, u'weight': 100, u'between_bytes_timeout': 10000, u'ssl_client_key': None, u'first_byte_timeout': 15000, u'auto_loadbalance': False, u'use_ssl': False, u'port': 80, u'ssl_hostname': None, u'hostname': u'www.testr.com', u'error_threshold': 0, u'max_conn': 20, u'version': 1, u'ipv4': None, u'ipv6': None, u'connect_timeout': 1000, u'ssl_ca_cert': None, u'request_condition': '', u'healthcheck': None, u'address': u'www.testr.com', u'ssl_client_cert': None, u'name': u'bbb', u'client_cert': None, u'service_id': u'75hthkv6jrGW4aVjSReT0w'}]) resp = self.controller.get(self.service_name) self.assertEqual(resp[self.driver.provider_name]['domains'], ['www.testr.com']) self.assertEqual(resp[self.driver.provider_name]['caching'], [{'name': 'test-cache', 'ttl': 3600, 'rules': ''}]) self.assertEqual(resp[self.driver.provider_name]['origins'], [{'origin': u'www.testr.com', 'port': 80, 'ssl': False}]) def test_get_service_details_error(self): error = fastly.FastlyError('DIMMM') self.controller.client.get_service_by_name.side_effect = error resp = self.controller.get('rror-service') self.assertIn('error', resp[self.driver.provider_name]) def test_get_service_details_exception(self): exception = Exception('DOOM') self.controller.client.get_service_by_name.side_effect = exception resp = self.controller.get('magic-service') self.assertIn('error', resp[self.driver.provider_name])