http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/backup/fixtures/dimensiondata/oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOJOB.xml ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/backup/fixtures/dimensiondata/oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOJOB.xml b/apache-libcloud-1.0.0rc2/libcloud/test/backup/fixtures/dimensiondata/oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOJOB.xml deleted file mode 100644 index 45c97ce..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/backup/fixtures/dimensiondata/oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOJOB.xml +++ /dev/null @@ -1,11 +0,0 @@ -<?xml version="1.0" encoding="UTF-8" standalone="yes"?> -<ns9:BackupDetails assetId="12769311-938c-4669-9460-7723eb194269" servicePlan="Enterprise" state="NORMAL" xmlns:ns16="http://oec.api.opsource.net/schemas/storage"; xmlns="http://oec.api.opsource.net/schemas/admin"; xmlns:ns14="http://oec.api.opsource.net/schemas/directory"; xmlns:ns15="http://oec.api.opsource.net/schemas/multigeo"; xmlns:ns9="http://oec.api.opsource.net/schemas/backup"; xmlns:ns5="http://oec.api.opsource.net/schemas/datacenter"; xmlns:ns12="http://oec.api.opsource.net/schemas/support"; xmlns:ns13="http://oec.api.opsource.net/schemas/manualimport"; xmlns:ns6="http://oec.api.opsource.net/schemas/general"; xmlns:ns7="http://oec.api.opsource.net/schemas/reset"; xmlns:ns10="http://oec.api.opsource.net/schemas/server"; xmlns:ns8="http://oec.api.opsource.net/schemas/network"; xmlns:ns11="http://oec.api.opsource.net/schemas/whitelabel"; xmlns:ns2="http://oec.api.opsource.net/schemas/organization"; xmlns:ns4="http://oec.api.opsource.net/schemas/serverbootstrap"; xmlns:ns3="http://oec.api.o psource.net/schemas/vip"> - <ns9:backupClient id="30b1ff76-c76d-4d7c-b39d-3b72be0384c8" type="FA.Linux" isFileSystem="true" status="Unregistered"> - <ns9:description>Linux File Agent</ns9:description> - <ns9:schedulePolicyName>12AM - 6AM</ns9:schedulePolicyName> - <ns9:storagePolicyName>14 Day Storage Policy</ns9:storagePolicyName> - <ns9:times nextBackup="2016-02-09T00:00:00" lastOnline="2016-02-08T06:10:25"/> - <ns9:totalBackupSizeGb>0</ns9:totalBackupSizeGb> - <ns9:downloadUrl>https://backups-na.cloud-vpn.net/PCS/BackupClientInstallerDownload/cbb8a8c607ca4144e8828814edfc1634c8dd8782</ns9:downloadUrl> - </ns9:backupClient> -</ns9:BackupDetails>
http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_base.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_base.py b/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_base.py deleted file mode 100644 index e254e6d..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_base.py +++ /dev/null @@ -1,29 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 - -from __future__ import with_statement - -import sys - -from libcloud.test import unittest -from libcloud.backup.base import BackupDriver - - -class BaseTestCase(unittest.TestCase): - def setUp(self): - self.driver = BackupDriver('none', 'none') - - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_dimensiondata.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_dimensiondata.py b/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_dimensiondata.py deleted file mode 100644 index 6a80fd4..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/backup/test_dimensiondata.py +++ /dev/null @@ -1,490 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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. - -try: - from lxml import etree as ET -except ImportError: - from xml.etree import ElementTree as ET - -import sys -from libcloud.utils.py3 import httplib - -from libcloud.common.dimensiondata import DimensionDataAPIException -from libcloud.common.types import InvalidCredsError -from libcloud.backup.base import BackupTargetJob -from libcloud.backup.drivers.dimensiondata import DimensionDataBackupDriver as DimensionData -from libcloud.backup.drivers.dimensiondata import DEFAULT_BACKUP_PLAN - -from libcloud.test import MockHttp, unittest -from libcloud.test.backup import TestCaseMixin -from libcloud.test.file_fixtures import BackupFileFixtures - -from libcloud.test.secrets import DIMENSIONDATA_PARAMS - - -class DimensionDataTests(unittest.TestCase, TestCaseMixin): - - def setUp(self): - DimensionData.connectionCls.conn_classes = (None, DimensionDataMockHttp) - DimensionDataMockHttp.type = None - self.driver = DimensionData(*DIMENSIONDATA_PARAMS) - - def test_invalid_region(self): - with self.assertRaises(ValueError): - self.driver = DimensionData(*DIMENSIONDATA_PARAMS, region='blah') - - def test_invalid_creds(self): - DimensionDataMockHttp.type = 'UNAUTHORIZED' - with self.assertRaises(InvalidCredsError): - self.driver.list_targets() - - def test_list_targets(self): - targets = self.driver.list_targets() - self.assertEqual(len(targets), 2) - self.assertEqual(targets[0].id, '5579f3a7-4c32-4cf5-8a7e-b45c36a35c10') - self.assertEqual(targets[0].address, 'e75ead52-692f-4314-8725-c8a4f4d13a87') - self.assertEqual(targets[0].extra['servicePlan'], 'Enterprise') - - def test_create_target(self): - target = self.driver.create_target( - 'name', - 'e75ead52-692f-4314-8725-c8a4f4d13a87', - extra={'servicePlan': 'Enterprise'}) - self.assertEqual(target.id, 'ee7c4b64-f7af-4a4f-8384-be362273530f') - self.assertEqual(target.address, 'e75ead52-692f-4314-8725-c8a4f4d13a87') - self.assertEqual(target.extra['servicePlan'], 'Enterprise') - - def test_create_target_DEFAULT(self): - DimensionDataMockHttp.type = 'DEFAULT' - target = self.driver.create_target( - 'name', - 'e75ead52-692f-4314-8725-c8a4f4d13a87') - self.assertEqual(target.id, 'ee7c4b64-f7af-4a4f-8384-be362273530f') - self.assertEqual(target.address, 'e75ead52-692f-4314-8725-c8a4f4d13a87') - - def test_create_target_EXISTS(self): - DimensionDataMockHttp.type = 'EXISTS' - with self.assertRaises(DimensionDataAPIException) as context: - self.driver.create_target( - 'name', - 'e75ead52-692f-4314-8725-c8a4f4d13a87', - extra={'servicePlan': 'Enterprise'}) - self.assertEqual(context.exception.code, 'ERROR') - self.assertEqual(context.exception.msg, 'Cloud backup for this server is already enabled or being enabled (state: NORMAL).') - - def test_update_target(self): - target = self.driver.list_targets()[0] - extra = {'servicePlan': 'Essentials'} - new_target = self.driver.update_target(target, extra=extra) - self.assertEqual(new_target.extra['servicePlan'], 'Essentials') - - def test_update_target_DEFAULT(self): - DimensionDataMockHttp.type = 'DEFAULT' - target = 'e75ead52-692f-4314-8725-c8a4f4d13a87' - self.driver.update_target(target) - - def test_update_target_STR(self): - target = 'e75ead52-692f-4314-8725-c8a4f4d13a87' - extra = {'servicePlan': 'Essentials'} - new_target = self.driver.update_target(target, extra=extra) - self.assertEqual(new_target.extra['servicePlan'], 'Essentials') - - def test_delete_target(self): - target = self.driver.list_targets()[0] - self.assertTrue(self.driver.delete_target(target)) - - def test_ex_add_client_to_target(self): - target = self.driver.list_targets()[0] - client = self.driver.ex_list_available_client_types(target)[0] - storage_policy = self.driver.ex_list_available_storage_policies(target)[0] - schedule_policy = self.driver.ex_list_available_schedule_policies(target)[0] - self.assertTrue( - self.driver.ex_add_client_to_target(target, client, storage_policy, - schedule_policy, 'ON_FAILURE', 'nob...@example.com') - ) - - def test_ex_add_client_to_target_STR(self): - self.assertTrue( - self.driver.ex_add_client_to_target('e75ead52-692f-4314-8725-c8a4f4d13a87', 'FA.Linux', '14 Day Storage Policy', - '12AM - 6AM', 'ON_FAILURE', 'nob...@example.com') - ) - - def test_ex_get_backup_details_for_target(self): - target = self.driver.list_targets()[0] - response = self.driver.ex_get_backup_details_for_target(target) - self.assertEqual(response.service_plan, 'Enterprise') - client = response.clients[0] - self.assertEqual(client.id, '30b1ff76-c76d-4d7c-b39d-3b72be0384c8') - self.assertEqual(client.type.type, 'FA.Linux') - self.assertEqual(client.running_job.progress, 5) - self.assertTrue(isinstance(client.running_job, BackupTargetJob)) - self.assertEqual(len(client.alert.notify_list), 2) - self.assertTrue(isinstance(client.alert.notify_list, list)) - - def test_ex_cancel_target_job(self): - target = self.driver.list_targets()[0] - response = self.driver.ex_get_backup_details_for_target(target) - client = response.clients[0] - self.assertTrue(isinstance(client.running_job, BackupTargetJob)) - success = client.running_job.cancel() - self.assertTrue(success) - - def test_ex_cancel_target_job_with_extras(self): - success = self.driver.cancel_target_job( - None, - ex_client='30b1ff76_c76d_4d7c_b39d_3b72be0384c8', - ex_target='e75ead52_692f_4314_8725_c8a4f4d13a87' - ) - self.assertTrue(success) - - def test_ex_cancel_target_job_FAIL(self): - DimensionDataMockHttp.type = 'FAIL' - with self.assertRaises(DimensionDataAPIException) as context: - self.driver.cancel_target_job( - None, - ex_client='30b1ff76_c76d_4d7c_b39d_3b72be0384c8', - ex_target='e75ead52_692f_4314_8725_c8a4f4d13a87' - ) - self.assertEqual(context.exception.code, 'ERROR') - - """Test a backup info for a target that does not have a client""" - def test_ex_get_backup_details_for_target_NO_CLIENT(self): - DimensionDataMockHttp.type = 'NOCLIENT' - response = self.driver.ex_get_backup_details_for_target('e75ead52-692f-4314-8725-c8a4f4d13a87') - self.assertEqual(response.service_plan, 'Essentials') - self.assertEqual(len(response.clients), 0) - - """Test a backup details that has a client, but no alerting or running jobs""" - def test_ex_get_backup_details_for_target_NO_JOB_OR_ALERT(self): - DimensionDataMockHttp.type = 'NOJOB' - response = self.driver.ex_get_backup_details_for_target('e75ead52-692f-4314_8725-c8a4f4d13a87') - self.assertEqual(response.service_plan, 'Enterprise') - self.assertTrue(isinstance(response.clients, list)) - self.assertEqual(len(response.clients), 1) - client = response.clients[0] - self.assertEqual(client.id, '30b1ff76-c76d-4d7c-b39d-3b72be0384c8') - self.assertEqual(client.type.type, 'FA.Linux') - self.assertIsNone(client.running_job) - self.assertIsNone(client.alert) - - """Test getting backup info for a server that doesn't exist""" - def test_ex_get_backup_details_for_target_DISABLED(self): - DimensionDataMockHttp.type = 'DISABLED' - with self.assertRaises(DimensionDataAPIException) as context: - self.driver.ex_get_backup_details_for_target('e75ead52-692f-4314-8725-c8a4f4d13a87') - self.assertEqual(context.exception.code, 'ERROR') - self.assertEqual(context.exception.msg, 'Server e75ead52-692f-4314-8725-c8a4f4d13a87 has not been provisioned for backup') - - def test_ex_list_available_client_types(self): - target = self.driver.list_targets()[0] - answer = self.driver.ex_list_available_client_types(target) - self.assertEqual(len(answer), 1) - self.assertEqual(answer[0].type, 'FA.Linux') - self.assertEqual(answer[0].is_file_system, True) - self.assertEqual(answer[0].description, 'Linux File system') - - def test_ex_list_available_storage_policies(self): - target = self.driver.list_targets()[0] - answer = self.driver.ex_list_available_storage_policies(target) - self.assertEqual(len(answer), 1) - self.assertEqual(answer[0].name, - '30 Day Storage Policy + Secondary Copy') - self.assertEqual(answer[0].retention_period, 30) - self.assertEqual(answer[0].secondary_location, 'Primary') - - def test_ex_list_available_schedule_policies(self): - target = self.driver.list_targets()[0] - answer = self.driver.ex_list_available_schedule_policies(target) - self.assertEqual(len(answer), 1) - self.assertEqual(answer[0].name, '12AM - 6AM') - self.assertEqual(answer[0].description, 'Daily backup will start between 12AM - 6AM') - - def test_ex_remove_client_from_target(self): - target = self.driver.list_targets()[0] - client = self.driver.ex_get_backup_details_for_target('e75ead52-692f-4314-8725-c8a4f4d13a87').clients[0] - self.assertTrue(self.driver.ex_remove_client_from_target(target, client)) - - def test_ex_remove_client_from_target_STR(self): - self.assertTrue( - self.driver.ex_remove_client_from_target( - 'e75ead52-692f-4314-8725-c8a4f4d13a87', - '30b1ff76-c76d-4d7c-b39d-3b72be0384c8' - ) - ) - - def test_ex_remove_client_from_target_FAIL(self): - DimensionDataMockHttp.type = 'FAIL' - with self.assertRaises(DimensionDataAPIException) as context: - self.driver.ex_remove_client_from_target( - 'e75ead52-692f-4314-8725-c8a4f4d13a87', - '30b1ff76-c76d-4d7c-b39d-3b72be0384c8' - ) - self.assertEqual(context.exception.code, 'ERROR') - self.assertTrue('Backup Client is currently performing another operation' in context.exception.msg) - - def test_priv_target_to_target_address(self): - target = self.driver.list_targets()[0] - self.assertEqual( - self.driver._target_to_target_address(target), - 'e75ead52-692f-4314-8725-c8a4f4d13a87' - ) - - def test_priv_target_to_target_address_STR(self): - self.assertEqual( - self.driver._target_to_target_address('e75ead52-692f-4314-8725-c8a4f4d13a87'), - 'e75ead52-692f-4314-8725-c8a4f4d13a87' - ) - - def test_priv_target_to_target_address_TYPEERROR(self): - with self.assertRaises(TypeError): - self.driver._target_to_target_address([1, 2, 3]) - - def test_priv_client_to_client_id(self): - client = self.driver.ex_get_backup_details_for_target('e75ead52-692f-4314-8725-c8a4f4d13a87').clients[0] - self.assertEqual( - self.driver._client_to_client_id(client), - '30b1ff76-c76d-4d7c-b39d-3b72be0384c8' - ) - - def test_priv_client_to_client_id_STR(self): - self.assertEqual( - self.driver._client_to_client_id('30b1ff76-c76d-4d7c-b39d-3b72be0384c8'), - '30b1ff76-c76d-4d7c-b39d-3b72be0384c8' - ) - - def test_priv_client_to_client_id_TYPEERROR(self): - with self.assertRaises(TypeError): - self.driver._client_to_client_id([1, 2, 3]) - - -class InvalidRequestError(Exception): - def __init__(self, tag): - super(InvalidRequestError, self).__init__("Invalid Request - %s" % tag) - - -class DimensionDataMockHttp(MockHttp): - - fixtures = BackupFileFixtures('dimensiondata') - - def _oec_0_9_myaccount_UNAUTHORIZED(self, method, url, body, headers): - return (httplib.UNAUTHORIZED, "", {}, httplib.responses[httplib.UNAUTHORIZED]) - - def _oec_0_9_myaccount(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_EXISTS(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_DEFAULT(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_INPROGRESS(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_FAIL(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_NOCLIENT(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_DISABLED(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_myaccount_NOJOB(self, method, url, body, headers): - body = self.fixtures.load('oec_0_9_myaccount.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DEFAULT(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DEFAULT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_NOCLIENT(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DEFAULT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_NOJOB(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DEFAULT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DISABLED(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server_e75ead52_692f_4314_8725_c8a4f4d13a87_DEFAULT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server(self, method, url, body, headers): - body = self.fixtures.load( - 'caas_2_1_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_server.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_type(self, method, url, body, headers): - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_type.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_storagePolicy( - self, method, url, body, headers): - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_storagePolicy.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_schedulePolicy( - self, method, url, body, headers): - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_schedulePolicy.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client( - self, method, url, body, headers): - if method == 'POST': - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_SUCCESS_PUT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - else: - raise ValueError("Unknown Method {0}".format(method)) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_NOCLIENT( - self, method, url, body, headers): - # only gets here are implemented - # If we get any other method something has gone wrong - assert(method == 'GET') - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOCLIENT.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_DISABLED( - self, method, url, body, headers): - # only gets here are implemented - # If we get any other method something has gone wrong - assert(method == 'GET') - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_DISABLED.xml') - return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_NOJOB( - self, method, url, body, headers): - # only gets here are implemented - # If we get any other method something has gone wrong - assert(method == 'GET') - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO_NOJOB.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_DEFAULT( - self, method, url, body, headers): - if method != 'POST': - raise InvalidRequestError('Only POST is accepted for this test') - request = ET.fromstring(body) - service_plan = request.get('servicePlan') - if service_plan != DEFAULT_BACKUP_PLAN: - raise InvalidRequestError('The default plan %s should have been passed in. Not %s' % (DEFAULT_BACKUP_PLAN, service_plan)) - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_ENABLE.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup( - self, method, url, body, headers): - if method == 'POST': - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_ENABLE.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - elif method == 'GET': - if url.endswith('disable'): - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_DISABLE.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_INFO.xml') - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - else: - raise ValueError("Unknown Method {0}".format(method)) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_EXISTS( - self, method, url, body, headers): - # only POSTs are implemented - # If we get any other method something has gone wrong - assert(method == 'POST') - body = self.fixtures.load( - 'oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_EXISTS.xml') - return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_modify( - self, method, url, body, headers): - request = ET.fromstring(body) - service_plan = request.get('servicePlan') - if service_plan != 'Essentials': - raise InvalidRequestError("Expected Essentials backup plan in request") - body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_modify.xml') - - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_modify_DEFAULT( - self, method, url, body, headers): - request = ET.fromstring(body) - service_plan = request.get('servicePlan') - if service_plan != DEFAULT_BACKUP_PLAN: - raise InvalidRequestError("Expected % backup plan in test" % DEFAULT_BACKUP_PLAN) - body = self.fixtures.load('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_modify.xml') - - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_30b1ff76_c76d_4d7c_b39d_3b72be0384c8( - self, method, url, body, headers): - if url.endswith('disable'): - body = self.fixtures.load( - ('_remove_backup_client.xml') - ) - elif url.endswith('cancelJob'): - body = self.fixtures.load( - ('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87' - '_backup_client_30b1ff76_c76d_4d7c_b39d_3b72be0384c8_cancelJob.xml') - ) - else: - raise ValueError("Unknown URL: %s" % url) - return (httplib.OK, body, {}, httplib.responses[httplib.OK]) - - def _oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87_backup_client_30b1ff76_c76d_4d7c_b39d_3b72be0384c8_FAIL( - self, method, url, body, headers): - if url.endswith('disable'): - body = self.fixtures.load( - ('_remove_backup_client_FAIL.xml') - ) - elif url.endswith('cancelJob'): - body = self.fixtures.load( - ('oec_0_9_8a8f6abc_2745_4d8a_9cbc_8dabe5a7d0e4_server_e75ead52_692f_4314_8725_c8a4f4d13a87' - '_backup_client_30b1ff76_c76d_4d7c_b39d_3b72be0384c8_cancelJob_FAIL.xml') - ) - else: - raise ValueError("Unknown URL: %s" % url) - return (httplib.BAD_REQUEST, body, {}, httplib.responses[httplib.OK]) - - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/file_fixtures.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/file_fixtures.py b/apache-libcloud-1.0.0rc2/libcloud/test/file_fixtures.py deleted file mode 100644 index 59f8e2e..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/file_fixtures.py +++ /dev/null @@ -1,97 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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. - -# Helper class for loading large fixture data -from __future__ import with_statement - -import os - -from libcloud.utils.py3 import PY3 -from libcloud.utils.py3 import u - -FIXTURES_ROOT = { - 'common': 'common/fixtures', - 'compute': 'compute/fixtures', - 'storage': 'storage/fixtures', - 'loadbalancer': 'loadbalancer/fixtures', - 'dns': 'dns/fixtures', - 'backup': 'backup/fixtures', - 'openstack': 'compute/fixtures/openstack', - 'container': 'container/fixtures' -} - - -class FileFixtures(object): - def __init__(self, fixtures_type, sub_dir=''): - script_dir = os.path.abspath(os.path.split(__file__)[0]) - self.root = os.path.join(script_dir, FIXTURES_ROOT[fixtures_type], - sub_dir) - - def load(self, file): - path = os.path.join(self.root, file) - if os.path.exists(path): - if PY3: - kwargs = {'encoding': 'utf-8'} - else: - kwargs = {} - - with open(path, 'r', **kwargs) as fh: - content = fh.read() - return u(content) - else: - raise IOError(path) - - -class ComputeFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(ComputeFileFixtures, self).__init__(fixtures_type='compute', - sub_dir=sub_dir) - - -class StorageFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(StorageFileFixtures, self).__init__(fixtures_type='storage', - sub_dir=sub_dir) - - -class LoadBalancerFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(LoadBalancerFileFixtures, self).__init__( - fixtures_type='loadbalancer', - sub_dir=sub_dir) - - -class DNSFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(DNSFileFixtures, self).__init__(fixtures_type='dns', - sub_dir=sub_dir) - - -class OpenStackFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(OpenStackFixtures, self).__init__(fixtures_type='openstack', - sub_dir=sub_dir) - - -class ContainerFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(ContainerFileFixtures, self).__init__(fixtures_type='container', - sub_dir=sub_dir) - - -class BackupFileFixtures(FileFixtures): - def __init__(self, sub_dir=''): - super(BackupFileFixtures, self).__init__(fixtures_type='backup', - sub_dir=sub_dir) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/pricing_test.json ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/pricing_test.json b/apache-libcloud-1.0.0rc2/libcloud/test/pricing_test.json deleted file mode 100644 index 9277874..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/pricing_test.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "compute": { - "foo": { - "1": 1.00, - "2": 2.00 - } - }, - - "updated": 1309019791 -} http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/secrets.py-dist ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/secrets.py-dist b/apache-libcloud-1.0.0rc2/libcloud/test/secrets.py-dist deleted file mode 100644 index 4412f13..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/secrets.py-dist +++ /dev/null @@ -1,94 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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. - -# Make a copy of this file named 'secrets.py' and add your credentials there. -# Note you can run unit tests without setting your credentials. - -BLUEBOX_PARAMS = ('customer_id', 'api_key') -BRIGHTBOX_PARAMS = ('client_id', 'client_secret') -EC2_PARAMS = ('access_id', 'secret') -ECP_PARAMS = ('user_name', 'password') -GANDI_PARAMS = ('user',) -GCE_PARAMS = ('em...@developer.gserviceaccount.com', 'key') # Service Account Authentication -# GCE_PARAMS = ('client_id', 'client_secret') # Installed App Authentication -GCE_KEYWORD_PARAMS = {'project': 'project_name'} -HOSTINGCOM_PARAMS = ('user', 'secret') -IBM_PARAMS = ('user', 'secret') -ONAPP_PARAMS = ('key',) -# OPENSTACK_PARAMS = ('user_name', 'api_key', secure_bool, 'host', port_int) -OPENSTACK_PARAMS = ('user_name', 'api_key', False, 'host', 8774) -OPENNEBULA_PARAMS = ('user', 'key') -DIMENSIONDATA_PARAMS = ('user', 'password') -OPSOURCE_PARAMS = ('user', 'password') -RUNABOVE_PARAMS = ('application_key', 'application_secret', 'consumer_key') -RACKSPACE_PARAMS = ('user', 'key') -RACKSPACE_NOVA_PARAMS = ('user_name', 'api_key', False, 'host', 8774) -SLICEHOST_PARAMS = ('key',) -SOFTLAYER_PARAMS = ('user', 'api_key') -VCLOUD_PARAMS = ('user', 'secret') -VOXEL_PARAMS = ('key', 'secret') -VPSNET_PARAMS = ('user', 'key') -JOYENT_PARAMS = ('user', 'key') -VCL_PARAMS = ('user', 'pass', True, 'foo.bar.com') -GRIDSPOT_PARAMS = ('key',) -HOSTVIRTUAL_PARAMS = ('key',) -DIGITALOCEAN_v1_PARAMS = ('user', 'key') -DIGITALOCEAN_v2_PARAMS = ('token',) -CLOUDFRAMES_PARAMS = ('key', 'secret', False, 'host', 8888) -PROFIT_BRICKS_PARAMS = ('user', 'key') -VULTR_PARAMS = ('key') -PACKET_PARAMS = ('api_key') -ECS_PARAMS = ('access_key', 'access_secret') - -# Storage -STORAGE_S3_PARAMS = ('key', 'secret') -STORAGE_OSS_PARAMS = ('key', 'secret') -# Google key = 20 char alphanumeric string starting with GOOG -STORAGE_GOOGLE_STORAGE_PARAMS = ('GOOG0123456789ABCXYZ', 'secret') - -# Azure key is b64 encoded and must be decoded before signing requests -STORAGE_AZURE_BLOBS_PARAMS = ('account', 'cGFzc3dvcmQ=') - -# Loadbalancer -LB_BRIGHTBOX_PARAMS = ('user', 'key') -LB_ELB_PARAMS = ('access_id', 'secret', 'region') -LB_SLB_PARAMS = ('access_id', 'secret', 'region') - -# DNS -DNS_PARAMS_LINODE = ('user', 'key') -DNS_PARAMS_ZERIGO = ('email', 'api token') -DNS_PARAMS_RACKSPACE = ('user', 'key') -DNS_PARAMS_HOSTVIRTUAL = ('key',) -DNS_PARAMS_ROUTE53 = ('access_id', 'secret') -DNS_GANDI = ('user', ) -DNS_PARAMS_GOOGLE = ('email_address', 'key') -DNS_KEYWORD_PARAMS_GOOGLE = {'project': 'project_name'} -DNS_PARAMS_WORLDWIDEDNS = ('user', 'key') -DNS_PARAMS_DNSIMPLE = ('user', 'key') -DNS_PARAMS_POINTDNS = ('user', 'key') -DNS_PARAMS_LIQUIDWEB = ('user', 'key') -DNS_PARAMS_ZONOMI = ('key') -DNS_PARAMS_DURABLEDNS = ('api_user', 'api_key') -DNS_PARAMS_GODADDY = ('customer-id', 'api_user', 'api_key') -DNS_PARAMS_CLOUDFLARE = ('u...@example.com', 'key') -DNS_PARAMS_AURORADNS = ('apikey', 'secretkey') -DNS_PARAMS_NSONE = ('key', ) -DNS_PARAMS_LUADNS = ('user', 'key') -DNS_PARAMS_BUDDYNS = ('key', ) - -# Container -CONTAINER_PARAMS_DOCKER = ('user', 'password') -CONTAINER_PARAMS_ECS = ('user', 'password', 'region') -CONTAINER_PARAMS_KUBERNETES = ('user', 'password') http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_connection.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_connection.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_connection.py deleted file mode 100644 index 96e86ce..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_connection.py +++ /dev/null @@ -1,334 +0,0 @@ -# -*- coding: utf-8 -*- -# Licensed to the Apache Software Foundation (ASF) under one or more§ -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 os -import socket -import sys -import ssl - -from mock import Mock, call, patch - -from libcloud.test import unittest -from libcloud.common.base import Connection -from libcloud.common.base import LoggingConnection -from libcloud.httplib_ssl import LibcloudBaseConnection -from libcloud.httplib_ssl import LibcloudHTTPConnection -from libcloud.utils.misc import retry - - -class BaseConnectionClassTestCase(unittest.TestCase): - def test_parse_proxy_url(self): - conn = LibcloudBaseConnection() - - proxy_url = 'http://127.0.0.1:3128' - result = conn._parse_proxy_url(proxy_url=proxy_url) - self.assertEqual(result[0], 'http') - self.assertEqual(result[1], '127.0.0.1') - self.assertEqual(result[2], 3128) - self.assertEqual(result[3], None) - self.assertEqual(result[4], None) - - proxy_url = 'http://user1:pass1@127.0.0.1:3128' - result = conn._parse_proxy_url(proxy_url=proxy_url) - self.assertEqual(result[0], 'http') - self.assertEqual(result[1], '127.0.0.1') - self.assertEqual(result[2], 3128) - self.assertEqual(result[3], 'user1') - self.assertEqual(result[4], 'pass1') - - proxy_url = 'https://127.0.0.1:3128' - expected_msg = 'Only http proxies are supported' - self.assertRaisesRegexp(ValueError, expected_msg, - conn._parse_proxy_url, - proxy_url=proxy_url) - - proxy_url = 'http://127.0.0.1' - expected_msg = 'proxy_url must be in the following format' - self.assertRaisesRegexp(ValueError, expected_msg, - conn._parse_proxy_url, - proxy_url=proxy_url) - - proxy_url = 'http://@127.0.0.1:3128' - expected_msg = 'URL is in an invalid format' - self.assertRaisesRegexp(ValueError, expected_msg, - conn._parse_proxy_url, - proxy_url=proxy_url) - - proxy_url = 'http://user@127.0.0.1:3128' - expected_msg = 'URL is in an invalid format' - self.assertRaisesRegexp(ValueError, expected_msg, - conn._parse_proxy_url, - proxy_url=proxy_url) - - def test_constructor(self): - conn = LibcloudHTTPConnection(host='localhost', port=80) - self.assertEqual(conn.proxy_scheme, None) - self.assertEqual(conn.proxy_host, None) - self.assertEqual(conn.proxy_port, None) - - proxy_url = 'http://127.0.0.3:3128' - conn.set_http_proxy(proxy_url=proxy_url) - self.assertEqual(conn.proxy_scheme, 'http') - self.assertEqual(conn.proxy_host, '127.0.0.3') - self.assertEqual(conn.proxy_port, 3128) - - proxy_url = 'http://127.0.0.4:3128' - conn = LibcloudHTTPConnection(host='localhost', port=80, - proxy_url=proxy_url) - self.assertEqual(conn.proxy_scheme, 'http') - self.assertEqual(conn.proxy_host, '127.0.0.4') - self.assertEqual(conn.proxy_port, 3128) - - os.environ['http_proxy'] = proxy_url - proxy_url = 'http://127.0.0.5:3128' - conn = LibcloudHTTPConnection(host='localhost', port=80, - proxy_url=proxy_url) - self.assertEqual(conn.proxy_scheme, 'http') - self.assertEqual(conn.proxy_host, '127.0.0.5') - self.assertEqual(conn.proxy_port, 3128) - - -class ConnectionClassTestCase(unittest.TestCase): - def setUp(self): - self.originalConnect = Connection.connect - self.originalResponseCls = Connection.responseCls - - Connection.connect = Mock() - Connection.responseCls = Mock() - Connection.allow_insecure = True - - def tearDown(self): - Connection.connect = self.originalConnect - Connection.responseCls = Connection.responseCls - Connection.allow_insecure = True - - def test_dont_allow_insecure(self): - Connection.allow_insecure = True - Connection(secure=False) - - Connection.allow_insecure = False - - expected_msg = (r'Non https connections are not allowed \(use ' - 'secure=True\)') - self.assertRaisesRegexp(ValueError, expected_msg, Connection, - secure=False) - - def test_content_length(self): - con = Connection() - con.connection = Mock() - - # GET method - # No data, no content length should be present - con.request('/test', method='GET', data=None) - call_kwargs = con.connection.request.call_args[1] - self.assertTrue('Content-Length' not in call_kwargs['headers']) - - # '' as data, no content length should be present - con.request('/test', method='GET', data='') - call_kwargs = con.connection.request.call_args[1] - self.assertTrue('Content-Length' not in call_kwargs['headers']) - - # 'a' as data, content length should be present (data in GET is not - # correct, but anyways) - con.request('/test', method='GET', data='a') - call_kwargs = con.connection.request.call_args[1] - self.assertEqual(call_kwargs['headers']['Content-Length'], '1') - - # POST, PUT method - # No data, content length should be present - for method in ['POST', 'PUT', 'post', 'put']: - con.request('/test', method=method, data=None) - call_kwargs = con.connection.request.call_args[1] - self.assertEqual(call_kwargs['headers']['Content-Length'], '0') - - # '' as data, content length should be present - for method in ['POST', 'PUT', 'post', 'put']: - con.request('/test', method=method, data='') - call_kwargs = con.connection.request.call_args[1] - self.assertEqual(call_kwargs['headers']['Content-Length'], '0') - - # No data, raw request, do not touch Content-Length if present - for method in ['POST', 'PUT', 'post', 'put']: - con.request('/test', method=method, data=None, - headers={'Content-Length': '42'}, raw=True) - putheader_call_list = con.connection.putheader.call_args_list - self.assertIn(call('Content-Length', '42'), putheader_call_list) - - # '' as data, raw request, do not touch Content-Length if present - for method in ['POST', 'PUT', 'post', 'put']: - con.request('/test', method=method, data=None, - headers={'Content-Length': '42'}, raw=True) - putheader_call_list = con.connection.putheader.call_args_list - self.assertIn(call('Content-Length', '42'), putheader_call_list) - - # 'a' as data, content length should be present - for method in ['POST', 'PUT', 'post', 'put']: - con.request('/test', method=method, data='a') - call_kwargs = con.connection.request.call_args[1] - self.assertEqual(call_kwargs['headers']['Content-Length'], '1') - - def test_cache_busting(self): - params1 = {'foo1': 'bar1', 'foo2': 'bar2'} - params2 = [('foo1', 'bar1'), ('foo2', 'bar2')] - - con = Connection() - con.connection = Mock() - con.pre_connect_hook = Mock() - con.pre_connect_hook.return_value = {}, {} - con.cache_busting = False - - con.request(action='/path', params=params1) - args, kwargs = con.pre_connect_hook.call_args - self.assertFalse('cache-busting' in args[0]) - self.assertEqual(args[0], params1) - - con.request(action='/path', params=params2) - args, kwargs = con.pre_connect_hook.call_args - self.assertFalse('cache-busting' in args[0]) - self.assertEqual(args[0], params2) - - con.cache_busting = True - - con.request(action='/path', params=params1) - args, kwargs = con.pre_connect_hook.call_args - self.assertTrue('cache-busting' in args[0]) - - con.request(action='/path', params=params2) - args, kwargs = con.pre_connect_hook.call_args - self.assertTrue('cache-busting' in args[0][len(params2)]) - - def test_context_is_reset_after_request_has_finished(self): - context = {'foo': 'bar'} - - def responseCls(connection, response): - connection.called = True - self.assertEqual(connection.context, context) - - con = Connection() - con.called = False - con.connection = Mock() - con.responseCls = responseCls - - con.set_context(context) - self.assertEqual(con.context, context) - - con.request('/') - - # Context should have been reset - self.assertTrue(con.called) - self.assertEqual(con.context, {}) - - # Context should also be reset if a method inside request throws - con = Connection(timeout=1, retry_delay=0.1) - con.connection = Mock() - - con.set_context(context) - self.assertEqual(con.context, context) - con.connection.request = Mock(side_effect=ssl.SSLError()) - - try: - con.request('/') - except ssl.SSLError: - pass - - self.assertEqual(con.context, {}) - - con.connection = Mock() - con.set_context(context) - self.assertEqual(con.context, context) - - con.responseCls = Mock(side_effect=ValueError()) - - try: - con.request('/') - except ValueError: - pass - - self.assertEqual(con.context, {}) - - def test_log_curl(self): - url = '/test/path' - body = None - headers = {} - - con = LoggingConnection() - con.protocol = 'http' - con.host = 'example.com' - con.port = 80 - - for method in ['GET', 'POST', 'PUT', 'DELETE']: - cmd = con._log_curl(method=method, url=url, body=body, - headers=headers) - self.assertEqual(cmd, 'curl -i -X %s --compress http://example.com:80/test/path' % - (method)) - - # Should use --head for head requests - cmd = con._log_curl(method='HEAD', url=url, body=body, headers=headers) - self.assertEqual(cmd, 'curl -i --head --compress http://example.com:80/test/path') - - def _raise_socket_error(self): - raise socket.gaierror('') - - def test_retry_with_sleep(self): - con = Connection() - con.connection = Mock() - connect_method = 'libcloud.common.base.Connection.request' - - with patch(connect_method) as mock_connect: - mock_connect.__name__ = 'mock_connect' - with self.assertRaises(socket.gaierror): - mock_connect.side_effect = socket.gaierror('') - retry_request = retry(timeout=1, retry_delay=.1, - backoff=1) - retry_request(con.request)(action='/') - - self.assertGreater(mock_connect.call_count, 1, - 'Retry logic failed') - - def test_retry_with_timeout(self): - con = Connection() - con.connection = Mock() - connect_method = 'libcloud.common.base.Connection.request' - - with patch(connect_method) as mock_connect: - mock_connect.__name__ = 'mock_connect' - with self.assertRaises(socket.gaierror): - mock_connect.side_effect = socket.gaierror('') - retry_request = retry(timeout=2, retry_delay=.1, - backoff=1) - retry_request(con.request)(action='/') - - self.assertGreater(mock_connect.call_count, 1, - 'Retry logic failed') - - def test_retry_with_backoff(self): - con = Connection() - con.connection = Mock() - connect_method = 'libcloud.common.base.Connection.request' - - with patch(connect_method) as mock_connect: - mock_connect.__name__ = 'mock_connect' - with self.assertRaises(socket.gaierror): - mock_connect.side_effect = socket.gaierror('') - retry_request = retry(timeout=2, retry_delay=.1, - backoff=1) - retry_request(con.request)(action='/') - - self.assertGreater(mock_connect.call_count, 1, - 'Retry logic failed') - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_file_fixtures.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_file_fixtures.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_file_fixtures.py deleted file mode 100644 index 395f315..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_file_fixtures.py +++ /dev/null @@ -1,32 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 sys -import unittest - -from libcloud.test.file_fixtures import ComputeFileFixtures - - -class FileFixturesTests(unittest.TestCase): - - def test_success(self): - f = ComputeFileFixtures('meta') - self.assertEqual("Hello, World!", f.load('helloworld.txt')) - - def test_failure(self): - f = ComputeFileFixtures('meta') - self.assertRaises(IOError, f.load, 'nil') - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_httplib_ssl.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_httplib_ssl.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_httplib_ssl.py deleted file mode 100644 index 996498f..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_httplib_ssl.py +++ /dev/null @@ -1,157 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 os -import sys -import os.path -import socket - -import mock -from mock import patch - -import libcloud.security - -from libcloud.utils.py3 import reload -from libcloud.httplib_ssl import LibcloudHTTPSConnection - -from libcloud.test import unittest - -ORIGINAL_CA_CERS_PATH = libcloud.security.CA_CERTS_PATH - - -class TestHttpLibSSLTests(unittest.TestCase): - - def setUp(self): - libcloud.security.VERIFY_SSL_CERT = False - libcloud.security.CA_CERTS_PATH = ORIGINAL_CA_CERS_PATH - self.httplib_object = LibcloudHTTPSConnection('foo.bar') - - def test_custom_ca_path_using_env_var_doesnt_exist(self): - os.environ['SSL_CERT_FILE'] = '/foo/doesnt/exist' - - try: - reload(libcloud.security) - except ValueError: - e = sys.exc_info()[1] - msg = 'Certificate file /foo/doesnt/exist doesn\'t exist' - self.assertEqual(str(e), msg) - else: - self.fail('Exception was not thrown') - - def test_custom_ca_path_using_env_var_is_directory(self): - file_path = os.path.dirname(os.path.abspath(__file__)) - os.environ['SSL_CERT_FILE'] = file_path - - expected_msg = 'Certificate file can\'t be a directory' - self.assertRaisesRegexp(ValueError, expected_msg, - reload, libcloud.security) - - def test_custom_ca_path_using_env_var_exist(self): - # When setting a path we don't actually check that a valid CA file is - # provided. - # This happens later in the code in httplib_ssl.connect method - file_path = os.path.abspath(__file__) - os.environ['SSL_CERT_FILE'] = file_path - - reload(libcloud.security) - - self.assertEqual(libcloud.security.CA_CERTS_PATH, [file_path]) - - @patch('warnings.warn') - def test_setup_verify(self, _): - libcloud.security.CA_CERTS_PATH = [] - - # Should throw a runtime error - libcloud.security.VERIFY_SSL_CERT = True - - expected_msg = libcloud.security.CA_CERTS_UNAVAILABLE_ERROR_MSG - self.assertRaisesRegexp(RuntimeError, expected_msg, - self.httplib_object._setup_verify) - - libcloud.security.VERIFY_SSL_CERT = False - self.httplib_object._setup_verify() - - @patch('warnings.warn') - def test_setup_ca_cert(self, _): - # verify = False, _setup_ca_cert should be a no-op - self.httplib_object.verify = False - self.httplib_object._setup_ca_cert() - - self.assertEqual(self.httplib_object.ca_cert, None) - - # verify = True, a valid path is provided, self.ca_cert should be set to - # a valid path - self.httplib_object.verify = True - - libcloud.security.CA_CERTS_PATH = [os.path.abspath(__file__)] - self.httplib_object._setup_ca_cert() - - self.assertTrue(self.httplib_object.ca_cert is not None) - - # verify = True, no CA certs are available, exception should be thrown - libcloud.security.CA_CERTS_PATH = [] - - expected_msg = libcloud.security.CA_CERTS_UNAVAILABLE_ERROR_MSG - self.assertRaisesRegexp(RuntimeError, expected_msg, - self.httplib_object._setup_ca_cert) - - @mock.patch('socket.create_connection', mock.MagicMock()) - @mock.patch('socket.socket', mock.MagicMock()) - @mock.patch('ssl.wrap_socket') - def test_connect_throws_friendly_error_message_on_ssl_wrap_connection_reset_by_peer(self, mock_wrap_socket): - # Test that we re-throw a more friendly error message in case - # "connection reset by peer" error occurs when trying to establish a - # SSL connection - libcloud.security.VERIFY_SSL_CERT = True - self.httplib_object.verify = True - self.httplib_object.http_proxy_used = False - - # No connection reset by peer, original exception should be thrown - mock_wrap_socket.side_effect = Exception('foo bar fail') - - expected_msg = 'foo bar fail' - self.assertRaisesRegexp(Exception, expected_msg, - self.httplib_object.connect) - - # Connection reset by peer, wrapped exception with friendly error - # message should be thrown - mock_wrap_socket.side_effect = socket.error('Connection reset by peer') - - expected_msg = 'Failed to establish SSL / TLS connection' - self.assertRaisesRegexp(socket.error, expected_msg, - self.httplib_object.connect) - - # Same error but including errno - with self.assertRaises(socket.error) as cm: - mock_wrap_socket.side_effect = socket.error(104, 'Connection reset by peer') - self.httplib_object.connect() - - e = cm.exception - self.assertEqual(e.errno, 104) - self.assertTrue(expected_msg in str(e)) - - # Test original exception is propagated correctly on non reset by peer - # error - with self.assertRaises(socket.error) as cm: - mock_wrap_socket.side_effect = socket.error(105, 'Some random error') - self.httplib_object.connect() - - e = cm.exception - self.assertEqual(e.errno, 105) - self.assertTrue('Some random error' in str(e)) - - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_init.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_init.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_init.py deleted file mode 100644 index ad709d7..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_init.py +++ /dev/null @@ -1,61 +0,0 @@ -# -*- coding: utf-8 -*- -# Licensed to the Apache Software Foundation (ASF) under one or more§ -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 os -import sys -import logging - -try: - import paramiko - have_paramiko = True -except ImportError: - have_paramiko = False - -from libcloud import _init_once -from libcloud.common.base import LoggingHTTPConnection -from libcloud.common.base import LoggingHTTPSConnection - -from libcloud.test import unittest - - -class TestUtils(unittest.TestCase): - def test_init_once_and_debug_mode(self): - # Debug mode is disabled - _init_once() - - self.assertEqual(LoggingHTTPConnection.log, None) - self.assertEqual(LoggingHTTPSConnection.log, None) - - if have_paramiko: - logger = paramiko.util.logging.getLogger() - paramiko_log_level = logger.getEffectiveLevel() - self.assertEqual(paramiko_log_level, logging.WARNING) - - # Enable debug mode - os.environ['LIBCLOUD_DEBUG'] = '/dev/null' - _init_once() - - self.assertTrue(LoggingHTTPConnection.log is not None) - self.assertTrue(LoggingHTTPSConnection.log is not None) - - if have_paramiko: - logger = paramiko.util.logging.getLogger() - paramiko_log_level = logger.getEffectiveLevel() - self.assertEqual(paramiko_log_level, logging.DEBUG) - - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_pricing.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_pricing.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_pricing.py deleted file mode 100644 index 5b6d132..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_pricing.py +++ /dev/null @@ -1,106 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 os.path -import sys -import unittest - -import libcloud.pricing - -PRICING_FILE_PATH = os.path.join(os.path.dirname(__file__), 'pricing_test.json') - - -class PricingTestCase(unittest.TestCase): - - def test_get_pricing_success(self): - self.assertFalse('foo' in libcloud.pricing.PRICING_DATA['compute']) - - pricing = libcloud.pricing.get_pricing(driver_type='compute', - driver_name='foo', - pricing_file_path=PRICING_FILE_PATH) - self.assertEqual(pricing['1'], 1.0) - self.assertEqual(pricing['2'], 2.0) - - self.assertEqual(libcloud.pricing.PRICING_DATA['compute']['foo']['1'], 1.0) - self.assertEqual(libcloud.pricing.PRICING_DATA['compute']['foo']['2'], 2.0) - - def test_get_pricing_invalid_file_path(self): - try: - libcloud.pricing.get_pricing(driver_type='compute', driver_name='bar', - pricing_file_path='inexistent.json') - except IOError: - pass - else: - self.fail('Invalid pricing file path provided, but an exception was not' - ' thrown') - - def test_get_pricing_invalid_driver_type(self): - try: - libcloud.pricing.get_pricing(driver_type='invalid_type', driver_name='bar', - pricing_file_path='inexistent.json') - except AttributeError: - pass - else: - self.fail('Invalid driver_type provided, but an exception was not' - ' thrown') - - def test_get_pricing_not_in_cache(self): - try: - libcloud.pricing.get_pricing(driver_type='compute', driver_name='inexistent', - pricing_file_path=PRICING_FILE_PATH) - except KeyError: - pass - else: - self.fail('Invalid driver provided, but an exception was not' - ' thrown') - - def test_get_size_price(self): - libcloud.pricing.PRICING_DATA['compute']['foo'] = {2: 2, '3': 3} - price1 = libcloud.pricing.get_size_price(driver_type='compute', - driver_name='foo', - size_id=2) - price2 = libcloud.pricing.get_size_price(driver_type='compute', - driver_name='foo', - size_id='3') - self.assertEqual(price1, 2) - self.assertEqual(price2, 3) - - def test_invalid_pricing_cache(self): - libcloud.pricing.PRICING_DATA['compute']['foo'] = {2: 2} - self.assertTrue('foo' in libcloud.pricing.PRICING_DATA['compute']) - - libcloud.pricing.invalidate_pricing_cache() - self.assertFalse('foo' in libcloud.pricing.PRICING_DATA['compute']) - - def test_invalid_module_pricing_cache(self): - libcloud.pricing.PRICING_DATA['compute']['foo'] = {1: 1} - - self.assertTrue('foo' in libcloud.pricing.PRICING_DATA['compute']) - - libcloud.pricing.invalidate_module_pricing_cache(driver_type='compute', - driver_name='foo') - self.assertFalse('foo' in libcloud.pricing.PRICING_DATA['compute']) - libcloud.pricing.invalidate_module_pricing_cache(driver_type='compute', - driver_name='foo1') - - def test_set_pricing(self): - self.assertFalse('foo' in libcloud.pricing.PRICING_DATA['compute']) - - libcloud.pricing.set_pricing(driver_type='compute', driver_name='foo', - pricing={'foo': 1}) - self.assertTrue('foo' in libcloud.pricing.PRICING_DATA['compute']) - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_response_classes.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_response_classes.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_response_classes.py deleted file mode 100644 index ceeab4c..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_response_classes.py +++ /dev/null @@ -1,151 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 sys -import unittest -import zlib -import gzip - -from mock import Mock - -from libcloud.utils.py3 import httplib, b, StringIO, PY3 -from libcloud.common.base import Response, XmlResponse, JsonResponse -from libcloud.common.types import MalformedResponseError - - -class ResponseClassesTests(unittest.TestCase): - def setUp(self): - self._mock_response = Mock() - self._mock_response.getheaders.return_value = [] - self._mock_response.status = httplib.OK - self._mock_response._original_data = None - self._mock_connection = Mock() - - def test_XmlResponse_class(self): - self._mock_response.read.return_value = '<foo>bar</foo>' - response = XmlResponse(response=self._mock_response, - connection=self._mock_connection) - - parsed = response.parse_body() - self.assertEqual(parsed.tag, 'foo') - self.assertEqual(parsed.text, 'bar') - - def test_XmlResponse_class_malformed_response(self): - self._mock_response.read.return_value = '<foo>' - - try: - XmlResponse(response=self._mock_response, - connection=self._mock_connection) - except MalformedResponseError: - pass - else: - self.fail('Exception was not thrown') - - def test_XmlResponse_class_zero_length_body_strip(self): - self._mock_response.read.return_value = ' ' - - response = XmlResponse(response=self._mock_response, - connection=self._mock_connection) - - parsed = response.parse_body() - self.assertEqual(parsed, '') - - def test_JsonResponse_class_success(self): - self._mock_response.read.return_value = '{"foo": "bar"}' - response = JsonResponse(response=self._mock_response, - connection=self._mock_connection) - - parsed = response.parse_body() - self.assertEqual(parsed, {'foo': 'bar'}) - - def test_JsonResponse_class_malformed_response(self): - self._mock_response.read.return_value = '{"foo": "bar' - - try: - JsonResponse(response=self._mock_response, - connection=self._mock_connection) - except MalformedResponseError: - pass - else: - self.fail('Exception was not thrown') - - def test_JsonResponse_class_zero_length_body_strip(self): - self._mock_response.read.return_value = ' ' - - response = JsonResponse(response=self._mock_response, - connection=self._mock_connection) - - parsed = response.parse_body() - self.assertEqual(parsed, '') - - def test_deflate_encoding(self): - original_data = 'foo bar ponies, wooo zlib' - compressed_data = zlib.compress(b(original_data)) - - self._mock_response.read.return_value = compressed_data - self._mock_response.getheaders.return_value = \ - {'Content-Encoding': 'deflate'} - - response = Response(response=self._mock_response, - connection=self._mock_connection) - - body = response.parse_body() - self.assertEqual(body, original_data) - - self._mock_response.getheaders.return_value = \ - {'Content-Encoding': 'zlib'} - - response = Response(response=self._mock_response, - connection=self._mock_connection) - - body = response.parse_body() - self.assertEqual(body, original_data) - - def test_gzip_encoding(self): - original_data = 'foo bar ponies, wooo gzip' - - if PY3: - from io import BytesIO - string_io = BytesIO() - else: - string_io = StringIO() - - stream = gzip.GzipFile(fileobj=string_io, mode='w') - stream.write(b(original_data)) - stream.close() - compressed_data = string_io.getvalue() - - self._mock_response.read.return_value = compressed_data - self._mock_response.getheaders.return_value = \ - {'Content-Encoding': 'gzip'} - - response = Response(response=self._mock_response, - connection=self._mock_connection) - - body = response.parse_body() - self.assertEqual(body, original_data) - - self._mock_response.getheaders.return_value = \ - {'Content-Encoding': 'x-gzip'} - - response = Response(response=self._mock_response, - connection=self._mock_connection) - - body = response.parse_body() - self.assertEqual(body, original_data) - - -if __name__ == '__main__': - sys.exit(unittest.main()) http://git-wip-us.apache.org/repos/asf/libcloud/blob/8afcda91/apache-libcloud-1.0.0rc2/libcloud/test/test_types.py ---------------------------------------------------------------------- diff --git a/apache-libcloud-1.0.0rc2/libcloud/test/test_types.py b/apache-libcloud-1.0.0rc2/libcloud/test/test_types.py deleted file mode 100644 index 453ffcc..0000000 --- a/apache-libcloud-1.0.0rc2/libcloud/test/test_types.py +++ /dev/null @@ -1,112 +0,0 @@ -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 sys -import unittest - -from libcloud.common.types import LazyList - - -class TestLazyList(unittest.TestCase): - def setUp(self): - super(TestLazyList, self).setUp - self._get_more_counter = 0 - - def tearDown(self): - super(TestLazyList, self).tearDown - - def test_init(self): - data = [1, 2, 3, 4, 5] - ll = LazyList(get_more=self._get_more_exhausted) - ll_list = list(ll) - self.assertEqual(ll_list, data) - - def test_iterator(self): - data = [1, 2, 3, 4, 5] - ll = LazyList(get_more=self._get_more_exhausted) - for i, d in enumerate(ll): - self.assertEqual(d, data[i]) - - def test_empty_list(self): - ll = LazyList(get_more=self._get_more_empty) - - self.assertEqual(list(ll), []) - self.assertEqual(len(ll), 0) - self.assertTrue(10 not in ll) - - def test_iterator_not_exhausted(self): - data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] - ll = LazyList(get_more=self._get_more_not_exhausted) - number_of_iterations = 0 - for i, d in enumerate(ll): - self.assertEqual(d, data[i]) - number_of_iterations += 1 - self.assertEqual(number_of_iterations, 10) - - def test_len(self): - ll = LazyList(get_more=self._get_more_not_exhausted) - ll = LazyList(get_more=self._get_more_not_exhausted) - - self.assertEqual(len(ll), 10) - - def test_contains(self): - ll = LazyList(get_more=self._get_more_not_exhausted) - - self.assertTrue(40 not in ll) - self.assertTrue(1 in ll) - self.assertTrue(5 in ll) - self.assertTrue(10 in ll) - - def test_indexing(self): - ll = LazyList(get_more=self._get_more_not_exhausted) - - self.assertEqual(ll[0], 1) - self.assertEqual(ll[9], 10) - self.assertEqual(ll[-1], 10) - - try: - ll[11] - except IndexError: - pass - else: - self.fail('Exception was not thrown') - - def test_repr(self): - ll1 = LazyList(get_more=self._get_more_empty) - ll2 = LazyList(get_more=self._get_more_exhausted) - ll3 = LazyList(get_more=self._get_more_not_exhausted) - - self.assertEqual(repr(ll1), '[]') - self.assertEqual(repr(ll2), '[1, 2, 3, 4, 5]') - self.assertEqual(repr(ll3), '[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]') - - def _get_more_empty(self, last_key, value_dict): - return [], None, True - - def _get_more_exhausted(self, last_key, value_dict): - data = [1, 2, 3, 4, 5] - return data, 5, True - - def _get_more_not_exhausted(self, last_key, value_dict): - self._get_more_counter += 1 - if not last_key: - data, last_key, exhausted = [1, 2, 3, 4, 5], 5, False - else: - data, last_key, exhausted = [6, 7, 8, 9, 10], 10, True - - return data, last_key, exhausted - -if __name__ == '__main__': - sys.exit(unittest.main())
