Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# Mock up test by clearing credentials on a per provider basis
cloned_config = self.provider.config.copy()
if self.provider.PROVIDER_ID == 'aws':
cloned_config['aws_access_key'] = "dummy_a_key"
cloned_config['aws_secret_key'] = "dummy_s_key"
elif self.provider.PROVIDER_ID == 'openstack':
cloned_config['os_username'] = "cb_dummy"
cloned_config['os_password'] = "cb_dummy"
elif self.provider.PROVIDER_ID == 'azure':
cloned_config['azure_subscription_id'] = "cb_dummy"
elif self.provider.PROVIDER_ID == 'gcp':
cloned_config['gcp_service_creds_dict'] = {'dummy': 'dict'}
with self.assertRaises(ProviderConnectionException):
cloned_provider = CloudProviderFactory().create_provider(
self.provider.PROVIDER_ID, cloned_config)
cloned_provider.authenticate()
def test_middleware_inherited(self):
start_count = 10
class SomeDummyClass(object):
count = start_count
@intercept(event_pattern="*", priority=2499)
def return_incremented(self, event_args, *args, **kwargs):
self.count += 1
return self.count
factory = CloudProviderFactory()
some_obj = SomeDummyClass()
factory.middleware.add(some_obj)
provider_name = cb_helpers.get_env("CB_TEST_PROVIDER", "aws")
first_prov = factory.create_provider(provider_name, {})
# Any dispatched event should be intercepted and increment the count
first_prov.storage.volumes.get("anything")
self.assertEqual(first_prov.networking.networks.get("anything"),
start_count + 2)
second_prov = factory.create_provider(provider_name, {})
# This count should be independent of the previous one
self.assertEqual(second_prov.networking.networks.get("anything"),
start_count + 3)
def create_provider_instance(self):
provider_name = cb_helpers.get_env("CB_TEST_PROVIDER", "aws")
factory = CloudProviderFactory()
provider_class = factory.get_provider_class(provider_name)
config = {'default_wait_interval':
self.get_provider_wait_interval(provider_class),
'default_result_limit': 5}
return provider_class(config)
def test_create_provider_valid(self):
# Creating a provider with a known name should return
# a valid implementation
self.assertIsInstance(CloudProviderFactory().create_provider(
factory.ProviderList.AWS, {}),
interfaces.CloudProvider,
"create_provider did not return a valid VM type")
def create_provider_instance(self):
provider_name = os.environ.get("CB_TEST_PROVIDER", "azure")
use_mock_drivers = parse_bool(
os.environ.get("CB_USE_MOCK_PROVIDERS", "False"))
factory = CloudProviderFactory()
provider_class = factory.get_provider_class(provider_name,
get_mock=use_mock_drivers)
config = {'default_wait_interval':
self.get_provider_wait_interval(provider_class),
'azure_subscription_id':
'7904d702-e01c-4826-8519-f5a25c866a96',
'azure_client_id':
'69621fe1-f59f-43de-8799-269007c76b95',
'azure_secret':
'Orcw9U5Kd4cUDntDABg0dygN32RQ4FGBYyLRaJ/BlrM=',
'azure_tenant':
'75ec242e-054d-4b22-98a9-a4602ebb6027',
'azure_resource_group': 'CB-TEST-RG',
'azure_storage_account': 'cbtestsa',
'azure_vm_default_user_name': 'cbtestuser'
}
def create_provider_instance(self):
provider_name = os.environ.get("CB_TEST_PROVIDER", "azure")
use_mock_drivers = parse_bool(
os.environ.get("CB_USE_MOCK_PROVIDERS", "False"))
factory = CloudProviderFactory()
provider_class = factory.get_provider_class(provider_name,
get_mock=use_mock_drivers)
config = {'default_wait_interval':
self.get_provider_wait_interval(provider_class),
'azure_subscription_id':
'7904d702-e01c-4826-8519-f5a25c866a96',
'azure_client_id':
'69621fe1-f59f-43de-8799-269007c76b95',
'azure_secret':
'Orcw9U5Kd4cUDntDABg0dygN32RQ4FGBYyLRaJ/BlrM=',
'azure_tenant':
'75ec242e-054d-4b22-98a9-a4602ebb6027',
'azure_resource_group': 'CB-TEST-TEST-RG',
'azure_storage_account': 'cbtestsa134',
'azure_vm_default_user_name': 'cbtestuser'
}
def test_find_provider_include_mocks(self):
self.assertTrue(
any(cls for cls
in CloudProviderFactory().get_all_provider_classes()
if issubclass(cls, TestMockHelperMixin)),
"expected to find at least one mock provider")
:type cloud: Cloud
:param cloud: The cloud to create a provider for
:rtype: ``object`` of :class:`.dict`
:return: A dict containing the necessary credentials for the cloud.
"""
# In case a base class instance is sent in, attempt to retrieve the actual
# subclass.
if type(cloud) is models.Cloud:
cloud = models.Cloud.objects.get_subclass(slug=cloud.slug)
if isinstance(cloud, models.OpenStack):
config = {'os_auth_url': cloud.auth_url,
'os_region_name': cloud.region_name}
config.update(cred_dict)
return CloudProviderFactory().create_provider(ProviderList.OPENSTACK,
config)
elif isinstance(cloud, models.AWS):
config = {'ec2_is_secure': cloud.compute.ec2_is_secure,
'ec2_region_name': cloud.compute.ec2_region_name,
'ec2_region_endpoint': cloud.compute.ec2_region_endpoint,
'ec2_port': cloud.compute.ec2_port,
'ec2_conn_path': cloud.compute.ec2_conn_path,
's3_host': cloud.object_store.s3_host,
's3_port': cloud.object_store.s3_port,
's3_conn_path': cloud.object_store.s3_conn_path}
config.update(cred_dict)
return CloudProviderFactory().create_provider(ProviderList.AWS,
config)
elif isinstance(cloud, models.Azure):
config = {'azure_region_name': cloud.region_name,
'azure_resource_group': cloud.resource_group,
"""DRF serializers for the CloudMan Create API endpoints."""
from rest_framework import serializers
from cloudbridge.cloud.factory import CloudProviderFactory
from .api import CMInfrastructureAPI
from .drf_helpers import CustomHyperlinkedIdentityField
class CMCloudSerializer(serializers.Serializer):
cloud_id = serializers.CharField(read_only=True)
name = serializers.CharField()
provider_id = serializers.ChoiceField(
choices=[p for p in CloudProviderFactory().list_providers()])
provider_config = serializers.DictField()
nodes = CustomHyperlinkedIdentityField(view_name='node-list',
lookup_field='cloud_id',
lookup_url_kwarg='cloud_pk')
def create(self, valid_data):
return CMInfrastructureAPI().clouds.create(
valid_data.get('name'), valid_data.get('provider_id'),
valid_data.get('provider_config'))
class CMCloudNodeSerializer(serializers.Serializer):
id = serializers.CharField(read_only=True)
cloud_id = serializers.CharField(read_only=True)
name = serializers.CharField()
instance_type = serializers.CharField()
from test import helpers
from cloudbridge.cloud.factory import CloudProviderFactory
from cloudbridge.cloud.factory import ProviderList
config = {}
provider = CloudProviderFactory().create_provider(
ProviderList.OPENSTACK,
config)
# print provider.security.list_key_pairs()
# print provider.compute.instance_types.list()
# print next(provider.compute.instance_types.find_by_name("m1.small"))
instance = helpers.get_test_instance(provider)
# print provider.security.list_security_groups()