| | | OpenStackProject class to ensure that clean will not raise|
| | | an exception |
+----------------------------------+---------------+-----------------------------------------------------------+
+| test_update_quotas | 2 & 3 | Tests the ability to update quota values |
+| | nova & neutron| |
++----------------------------------+---------------+-----------------------------------------------------------+
create_project_tests.py - CreateProjectUserTests
------------------------------------------------
Ensures that all required members are included when constructing a
Domain domain object
+ComputeQuotasDomainObjectTests
+------------------------------
+
+Ensures that all required members are included when constructing a
+ComputeQuotas domain object
+
+NetworkQuotasDomainObjectTests
+------------------------------
+
+Ensures that all required members are included when constructing a
+NetworkQuotas domain object
+
RoleDomainObjectTests
---------------------
class Project:
"""
- SNAPS domain object for Projects. Should contain attributes that
+ SNAPS domain class for Projects. Should contain attributes that
are shared amongst cloud providers
"""
def __init__(self, name, project_id, domain_id=None):
class Domain:
"""
- SNAPS domain object for OpenStack Keystone v3+ domains.
+ SNAPS domain class for OpenStack Keystone v3+ domains.
"""
def __init__(self, name, domain_id=None):
"""
def __eq__(self, other):
return self.name == other.name and self.id == other.id
+
+
+class ComputeQuotas:
+ """
+ SNAPS domain class for holding project quotas for compute services
+ """
+ def __init__(self, nova_quotas=None, **kwargs):
+ """
+ Constructor
+ :param nova_quotas: the OS nova quota object
+ """
+ if nova_quotas:
+ self.metadata_items = nova_quotas.metadata_items
+ self.cores = nova_quotas.cores # aka. VCPUs
+ self.instances = nova_quotas.instances
+ self.injected_files = nova_quotas.injected_files
+ self.injected_file_content_bytes = nova_quotas.injected_file_content_bytes
+ self.ram = nova_quotas.ram
+ self.fixed_ips = nova_quotas.fixed_ips
+ self.key_pairs = nova_quotas.key_pairs
+ else:
+ self.metadata_items = kwargs.get('metadata_items')
+ self.cores = kwargs.get('cores') # aka. VCPUs
+ self.instances = kwargs.get('instances')
+ self.injected_files = kwargs.get('injected_files')
+ self.injected_file_content_bytes = kwargs.get(
+ 'injected_file_content_bytes')
+ self.ram = kwargs.get('ram')
+ self.fixed_ips = kwargs.get('fixed_ips')
+ self.key_pairs = kwargs.get('key_pairs')
+
+ def __eq__(self, other):
+ return (self.metadata_items == other.metadata_items and
+ self.cores == other.cores and
+ self.instances == other.instances and
+ self.injected_files == other.injected_files and
+ self.injected_file_content_bytes == other.injected_file_content_bytes and
+ self.fixed_ips == other.fixed_ips and
+ self.key_pairs == other.key_pairs)
+
+
+class NetworkQuotas:
+ """
+ SNAPS domain class for holding project quotas for networking services
+ """
+ def __init__(self, **neutron_quotas):
+ """
+ Constructor
+ :param neutron_quotas: the OS network quota object
+ """
+
+ # Networks settings here
+ self.security_group = neutron_quotas['security_group']
+ self.security_group_rule = neutron_quotas['security_group_rule']
+ self.floatingip = neutron_quotas['floatingip']
+ self.network = neutron_quotas['network']
+ self.port = neutron_quotas['port']
+ self.router = neutron_quotas['router']
+ self.subnet = neutron_quotas['subnet']
+
+ def __eq__(self, other):
+ return (self.security_group == other.security_group and
+ self.security_group_rule == other.security_group_rule and
+ self.floatingip == other.floatingip and
+ self.network == other.network and
+ self.port == other.port and
+ self.router == other.router and
+ self.subnet == other.subnet)
# limitations under the License.
import unittest
-from snaps.domain.project import Project, Domain
+from snaps.domain.project import Project, Domain, ComputeQuotas, NetworkQuotas
class ProjectDomainObjectTests(unittest.TestCase):
domain = Domain(domain_id='123-456', name='foo')
self.assertEqual('foo', domain.name)
self.assertEqual('123-456', domain.id)
+
+
+class ComputeQuotasDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.project.ComputeQuotas class
+ """
+
+ def test_construction_positional(self):
+ quotas = ComputeQuotas(
+ metadata_items=64, cores=5, instances= 4, injected_files= 3,
+ injected_file_content_bytes=5120,ram=25600, fixed_ips=100,
+ key_pairs=50)
+ self.assertEqual(64, quotas.metadata_items)
+ self.assertEqual(5, quotas.cores)
+ self.assertEqual(4, quotas.instances)
+ self.assertEqual(3, quotas.injected_files)
+ self.assertEqual(5120, quotas.injected_file_content_bytes)
+ self.assertEqual(25600, quotas.ram)
+ self.assertEqual(100, quotas.fixed_ips)
+ self.assertEqual(50, quotas.key_pairs)
+
+ def test_construction_named_minimal(self):
+ quotas = ComputeQuotas(
+ **{'metadata_items': 64, 'cores': 5, 'instances': 4,
+ 'injected_files': 3, 'injected_file_content_bytes': 5120,
+ 'ram': 25600, 'fixed_ips': 100, 'key_pairs': 50})
+ self.assertEqual(64, quotas.metadata_items)
+ self.assertEqual(5, quotas.cores)
+ self.assertEqual(4, quotas.instances)
+ self.assertEqual(3, quotas.injected_files)
+ self.assertEqual(5120, quotas.injected_file_content_bytes)
+ self.assertEqual(25600, quotas.ram)
+ self.assertEqual(100, quotas.fixed_ips)
+ self.assertEqual(50, quotas.key_pairs)
+
+
+class NetworkQuotasDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.project.NetworkQuotas class
+ """
+
+ def test_construction_positional(self):
+ quotas = NetworkQuotas(
+ security_group=5, security_group_rule=50,
+ floatingip=25, network=5, port=25, router=6, subnet=7)
+ self.assertEqual(5, quotas.security_group)
+ self.assertEqual(50, quotas.security_group_rule)
+ self.assertEqual(25, quotas.floatingip)
+ self.assertEqual(5, quotas.network)
+ self.assertEqual(25, quotas.port)
+ self.assertEqual(6, quotas.router)
+ self.assertEqual(7, quotas.subnet)
+
+ def test_construction_named_minimal(self):
+ quotas = NetworkQuotas(
+ **{'security_group': 5, 'security_group_rule': 50,
+ 'floatingip': 25, 'network': 5, 'port': 25, 'router': 6,
+ 'subnet': 7})
+ self.assertEqual(5, quotas.security_group)
+ self.assertEqual(50, quotas.security_group_rule)
+ self.assertEqual(25, quotas.floatingip)
+ self.assertEqual(5, quotas.network)
+ self.assertEqual(25, quotas.port)
+ self.assertEqual(6, quotas.router)
+ self.assertEqual(7, quotas.subnet)
import logging
from keystoneclient.exceptions import NotFound
-from snaps.openstack.utils import keystone_utils, neutron_utils
+
+from snaps.openstack.utils import keystone_utils, neutron_utils, nova_utils
__author__ = 'spisarski'
keystone_utils.grant_user_role_to_project(self.__keystone, self.__role,
user, self.__project)
+ def get_compute_quotas(self):
+ """
+ Returns the compute quotas as an instance of the ComputeQuotas class
+ :return:
+ """
+ nova = nova_utils.nova_client(self.__os_creds)
+ return nova_utils.get_compute_quotas(nova, self.__project.id)
+
+ def get_network_quotas(self):
+ """
+ Returns the network quotas as an instance of the NetworkQuotas class
+ :return:
+ """
+ neutron = neutron_utils.neutron_client(self.__os_creds)
+ return neutron_utils.get_network_quotas(neutron, self.__project.id)
+
+ def update_compute_quotas(self, compute_quotas):
+ """
+ Updates the compute quotas for this project
+ :param compute_quotas: a ComputeQuotas object.
+ """
+ nova = nova_utils.nova_client(self.__os_creds)
+ nova_utils.update_quotas(nova, self.__project.id, compute_quotas)
+
+ def update_network_quotas(self, network_quotas):
+ """
+ Updates the network quotas for this project
+ :param network_quotas: a NetworkQuotas object.
+ """
+ neutron = neutron_utils.neutron_client(self.__os_creds)
+ neutron_utils.update_quotas(neutron, self.__project.id, network_quotas)
+
class ProjectSettings:
"""
from keystoneclient.exceptions import BadRequest
+from snaps.domain.project import ComputeQuotas, NetworkQuotas
from snaps.openstack.create_project import (
OpenStackProject, ProjectSettings, ProjectSettingsError)
from snaps.openstack.create_security_group import OpenStackSecurityGroup
from snaps.openstack.create_user import OpenStackUser
from snaps.openstack.create_user import UserSettings
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import keystone_utils, nova_utils, neutron_utils
__author__ = 'spisarski'
self.assertTrue(validate_project(self.keystone, self.project_settings,
created_project))
+ def test_update_quotas(self):
+ """
+ Tests the creation of an OpenStack project where the quotas get
+ updated.
+ """
+ self.project_creator = OpenStackProject(self.os_creds,
+ self.project_settings)
+ created_project = self.project_creator.create()
+ self.assertIsNotNone(created_project)
+
+ retrieved_project = keystone_utils.get_project(
+ keystone=self.keystone, project_settings=self.project_settings)
+ self.assertIsNotNone(retrieved_project)
+ self.assertEqual(created_project, retrieved_project)
+ self.assertTrue(validate_project(self.keystone, self.project_settings,
+ created_project))
+
+ update_compute_quotas = ComputeQuotas(
+ **{'metadata_items': 64, 'cores': 5, 'instances': 5,
+ 'injected_files': 3, 'injected_file_content_bytes': 5120,
+ 'ram': 25600, 'fixed_ips': 100, 'key_pairs': 50})
+ self.project_creator.update_compute_quotas(update_compute_quotas)
+
+ update_network_quotas = NetworkQuotas(
+ **{'security_group': 5, 'security_group_rule': 50,
+ 'floatingip': 25, 'network': 5, 'port': 25, 'router': 6,
+ 'subnet': 7})
+ self.project_creator.update_network_quotas(update_network_quotas)
+
+ self.assertEqual(update_compute_quotas,
+ self.project_creator.get_compute_quotas())
+ self.assertEqual(update_network_quotas,
+ self.project_creator.get_network_quotas())
+
+ nova = nova_utils.nova_client(self.os_creds)
+ new_compute_quotas = nova_utils.get_compute_quotas(
+ nova, self.project_creator.get_project().id)
+ self.assertEqual(update_compute_quotas, new_compute_quotas)
+
+ neutron = neutron_utils.neutron_client(self.os_creds)
+ new_network_quotas = neutron_utils.get_network_quotas(
+ neutron, self.project_creator.get_project().id)
+ self.assertEqual(update_network_quotas, new_network_quotas)
+
class CreateProjectUserTests(OSComponentTestCase):
"""
def get_project(keystone=None, os_creds=None, project_settings=None,
project_name=None):
"""
- Returns the first project object or None if not found
+ Returns the first project where the project_settings is used for the query
+ if not None, else the project_name parameter is used for the query. If both
+ parameters are None, None is returned
:param keystone: the Keystone client
:param os_creds: the OpenStack credentials used to obtain the Keystone
client if the keystone parameter is None
proj_filter['description'] = project_settings.description
proj_filter['domain_name'] = project_settings.domain_name
proj_filter['enabled'] = project_settings.enabled
+ else:
+ return None
if keystone.version == V2_VERSION_STR:
projects = keystone.tenants.list()
from snaps.domain.network import (
Port, SecurityGroup, SecurityGroupRule, Router, InterfaceRouter, Subnet,
Network)
+from snaps.domain.project import NetworkQuotas
from snaps.domain.vm_inst import FloatingIp
from snaps.openstack.utils import keystone_utils
return neutron.delete_floatingip(floating_ip.id)
+def get_network_quotas(neutron, project_id):
+ """
+ Returns a list of all available keypairs
+ :param nova: the Nova client
+ :param project_id: the project's ID of the quotas to lookup
+ :return: an object of type NetworkQuotas or None if not found
+ """
+ quota = neutron.show_quota(project_id)
+ if quota:
+ return NetworkQuotas(**quota['quota'])
+
+
+def update_quotas(neutron, project_id, network_quotas):
+ """
+ Updates the networking quotas for a given project
+ :param neutron: the Neutron client
+ :param project_id: the project's ID that requires quota updates
+ :param network_quotas: an object of type NetworkQuotas containing the
+ values to update
+ :return:
+ """
+ update_body = dict()
+ update_body['security_group'] = network_quotas.security_group
+ update_body['security_group_rule'] = network_quotas.security_group_rule
+ update_body['floatingip'] = network_quotas.floatingip
+ update_body['network'] = network_quotas.network
+ update_body['port'] = network_quotas.port
+ update_body['router'] = network_quotas.router
+ update_body['subnet'] = network_quotas.subnet
+
+ return neutron.update_quota(project_id, {'quota': update_body})
+
+
class NeutronException(Exception):
"""
Exception when calls to the Keystone client cannot be served properly
from snaps.domain.flavor import Flavor
from snaps.domain.keypair import Keypair
+from snaps.domain.project import ComputeQuotas
from snaps.domain.vm_inst import VmInst
from snaps.openstack.utils import keystone_utils, glance_utils, neutron_utils
vm.add_floating_ip(floating_ip.ip, ip_addr)
+def get_compute_quotas(nova, project_id):
+ """
+ Returns a list of all available keypairs
+ :param nova: the Nova client
+ :param project_id: the project's ID of the quotas to lookup
+ :return: an object of type ComputeQuotas or None if not found
+ """
+ quotas = nova.quotas.get(tenant_id=project_id)
+ if quotas:
+ return ComputeQuotas(quotas)
+
+
+def update_quotas(nova, project_id, compute_quotas):
+ """
+ Updates the compute quotas for a given project
+ :param nova: the Nova client
+ :param project_id: the project's ID that requires quota updates
+ :param compute_quotas: an object of type ComputeQuotas containing the
+ values to update
+ :return:
+ """
+ update_values = dict()
+ update_values['metadata_items'] = compute_quotas.metadata_items
+ update_values['cores'] = compute_quotas.cores
+ update_values['instances'] = compute_quotas.instances
+ update_values['injected_files'] = compute_quotas.injected_files
+ update_values['injected_file_content_bytes'] = compute_quotas.injected_file_content_bytes
+ update_values['ram'] = compute_quotas.ram
+ update_values['fixed_ips'] = compute_quotas.fixed_ips
+ update_values['key_pairs'] = compute_quotas.key_pairs
+
+ return nova.quotas.update(project_id, **update_values)
+
+
class NovaException(Exception):
"""
Exception when calls to the Keystone client cannot be served properly
PortDomainObjectTests, RouterDomainObjectTests,
InterfaceRouterDomainObjectTests, NetworkObjectTests, SubnetObjectTests)
from snaps.domain.test.project_tests import (
- ProjectDomainObjectTests, DomainDomainObjectTests)
+ ProjectDomainObjectTests, DomainDomainObjectTests,
+ ComputeQuotasDomainObjectTests, NetworkQuotasDomainObjectTests)
from snaps.domain.test.role_tests import RoleDomainObjectTests
from snaps.domain.test.stack_tests import StackDomainObjectTests
from snaps.domain.test.user_tests import UserDomainObjectTests
ProjectDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
DomainDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ ComputeQuotasDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ NetworkQuotasDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
RoleDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
Code Review / snaps.git / commitdiff