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 Subnet(**subnet)
+def get_subnet_by_id(neutron, subnet_id):
+ """
+ Returns a SNAPS-OO Subnet domain object for a given ID
+ :param neutron: the OpenStack neutron client
+ :param subnet_id: the subnet ID
+ :return: a Subnet object
+ """
+ os_subnet = neutron.show_subnet(subnet_id)
+ if os_subnet and 'subnet' in os_subnet:
+ return Subnet(**os_subnet['subnet'])
+
+
+def get_subnets_by_network(neutron, network):
+ """
+ Returns a list of SNAPS-OO Subnet domain objects
+ :param neutron: the OpenStack neutron client
+ :param network: the SNAPS-OO Network domain object
+ :return: a list of Subnet objects
+ """
+ out = list()
+
+ os_subnets = neutron.list_subnets(network_id=network.id)
+
+ for os_subnet in os_subnets['subnets']:
+ out.append(Subnet(**os_subnet))
+
+ return out
+
+
def create_router(neutron, os_creds, router_settings):
"""
Creates a router for OpenStack
json_body = router_settings.dict_for_neutron(neutron, os_creds)
logger.info('Creating router with name - ' + router_settings.name)
os_router = neutron.create_router(json_body)
- return Router(**os_router['router'])
+ return __map_router(neutron, os_router['router'])
else:
logger.error("Failed to create router.")
raise NeutronException('Failed to create router')
neutron.delete_router(router=router.id)
+def get_router_by_id(neutron, router_id):
+ """
+ Returns a router with a given ID, else None if not found
+ :param neutron: the client
+ :param router_id: the Router ID
+ :return: a SNAPS-OO Router domain object
+ """
+ router = neutron.show_router(router_id)
+ if router:
+ return __map_router(neutron, router['router'])
+
+
def get_router(neutron, router_settings=None, router_name=None):
"""
Returns the first router object (dictionary) found the given the settings
return None
routers = neutron.list_routers(**router_filter)
+
for routerInst in routers['routers']:
- return Router(**routerInst)
+ return __map_router(neutron, routerInst)
+
return None
+def __map_router(neutron, os_router):
+ """
+ Takes an OpenStack router instance and maps it to a SNAPS Router domain
+ object
+ :param neutron: the neutron client
+ :param os_router: the OpenStack Router object
+ :return:
+ """
+ device_ports = neutron.list_ports(
+ **{'device_id': os_router['id']})['ports']
+ port_subnets = list()
+
+ # Order by create date
+ sorted_ports = sorted(device_ports, key=lambda dev_port: dev_port['created_at'])
+
+ for port in sorted_ports:
+ subnets = list()
+ for fixed_ip in port['fixed_ips']:
+ subnet = get_subnet_by_id(neutron, fixed_ip['subnet_id'])
+ if subnet and subnet.network_id == port['network_id']:
+ subnets.append(subnet)
+ port_subnets.append((Port(**port), subnets))
+
+ os_router['port_subnets'] = port_subnets
+ return Router(**os_router)
+
+
def add_interface_router(neutron, router, subnet=None, port=None):
"""
Adds an interface router for OpenStack for either a subnet or port.
port_filter = dict()
if port_settings:
- port_filter['name'] = port_settings.name
+ if port_settings.name and len(port_settings.name) > 0:
+ port_filter['name'] = port_settings.name
if port_settings.admin_state_up:
port_filter['admin_state_up'] = port_settings.admin_state_up
if port_settings.device_id:
port_filter['device_id'] = port_settings.device_id
if port_settings.mac_address:
port_filter['mac_address'] = port_settings.mac_address
+ if port_settings.network_name:
+ network = get_network(neutron,
+ network_name=port_settings.network_name)
+ port_filter['network_id'] = network.id
elif port_name:
port_filter['name'] = port_name
ports = neutron.list_ports(**port_filter)
for port in ports['ports']:
- return Port(name=port['name'], id=port['id'],
- ips=port['fixed_ips'], mac_address=port['mac_address'])
+ return Port(**port)
+ return None
+
+
+def get_port_by_id(neutron, port_id):
+ """
+ Returns a SNAPS-OO Port domain object for the given ID or none if not found
+ :param neutron: the client
+ :param port_id: the to query
+ :return: a SNAPS-OO Port domain object or None
+ """
+ port = neutron.show_port(port_id)
+ if port:
+ return Port(**port['port'])
return None
+def get_ports(neutron, network, ips=None):
+ """
+ Returns a list of SNAPS-OO Port objects for all OpenStack Port objects that
+ are associated with the 'network' parameter
+ :param neutron: the client
+ :param network: SNAPS-OO Network domain object
+ :param ips: the IPs to lookup if not None
+ :return: a SNAPS-OO Port domain object or None if not found
+ """
+ out = list()
+ ports = neutron.list_ports(**{'network_id': network.id})
+ for port in ports['ports']:
+ if ips:
+ for fixed_ips in port['fixed_ips']:
+ if ('ip_address' in fixed_ips and
+ fixed_ips['ip_address'] in ips) or ips is None:
+ out.append(Port(**port))
+ break
+ else:
+ out.append(Port(**port))
+
+ return out
+
+
def create_security_group(neutron, keystone, sec_grp_settings):
"""
Creates a security group object in OpenStack
Returns all of the floating IPs
When ports is not None, FIPs returned must be associated with one of the
ports in the list and a tuple 2 where the first element being the port's
- name and the second being the FloatingIp SNAPS-OO domain object.
+ ID and the second being the FloatingIp SNAPS-OO domain object.
When ports is None, all known FloatingIp SNAPS-OO domain objects will be
returned in a list
:param neutron: the Neutron client
- :param ports: a list of SNAPS-OO Port objects to join
- :return: a list of tuple 2 (port_name, SNAPS FloatingIp) objects when ports
+ :param ports: a list of tuple 2 where index 0 is the port name and index 1
+ is the SNAPS-OO Port object
+ :return: a list of tuple 2 (port_id, SNAPS FloatingIp) objects when ports
is not None else a list of Port objects
"""
out = list()
for fip in fips['floatingips']:
if ports:
for port_name, port in ports:
- if fip['port_id'] == port.id:
- out.append((port.name, FloatingIp(
- inst_id=fip['id'], ip=fip['floating_ip_address'])))
+ if port and port.id == fip['port_id']:
+ out.append((port.id, FloatingIp(**fip)))
break
else:
- out.append(FloatingIp(inst_id=fip['id'],
- ip=fip['floating_ip_address']))
+ out.append(FloatingIp(**fip))
return out
body={'floatingip':
{'floating_network_id': ext_net.id}})
- return FloatingIp(inst_id=fip['floatingip']['id'],
+ return FloatingIp(id=fip['floatingip']['id'],
ip=fip['floatingip']['floating_ip_address'])
else:
raise NeutronException(
floating_ip.ip)
os_fip = __get_os_floating_ip(neutron, floating_ip)
if os_fip:
- return FloatingIp(
- inst_id=os_fip['id'], ip=os_fip['floating_ip_address'])
+ return FloatingIp(id=os_fip['id'], ip=os_fip['floating_ip_address'])
def __get_os_floating_ip(neutron, floating_ip):
return neutron.delete_floatingip(floating_ip.id)
+def get_network_quotas(neutron, project_id):
+ """
+ Returns a list of all available keypairs
+ :param neutron: the neutron 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
Code Review / snaps.git / blobdiff