Commit bac9c8ac authored by Stavros Sachtouris's avatar Stavros Sachtouris
Browse files

Complete clients.networkinh + unittests

Refs: #4546
parent 25d2ff8e
......@@ -218,3 +218,75 @@ class NetworkingClient(NetworkingRestClient):
def delete_subnet(self, subnet_id):
r = self.subnets_delete(subnet_id, success=204)
return r.headers
def list_ports(self):
r = self.ports_get(success=200)
return r.json['ports']
def create_port(
self, network_id,
name=None, status=None, admin_state_up=None, mac_address=None,
fixed_ips=None, security_groups=None):
:param network_id: (str)
:param name: (str)
:param status: (str)
:param admin_state_up: (bool) Router administrative status (UP / DOWN)
:param mac_address: (str)
:param fixed_ips: (str)
:param security_groups: (list)
port = dict(network_id=network_id)
if name:
port['name'] = name
if status:
port['status'] = status
if admin_state_up not in (None, ):
port['admin_state_up'] = bool(admin_state_up)
if mac_address:
port['mac_address'] = mac_address
if fixed_ips:
port['fixed_ips'] = fixed_ips
if security_groups:
port['security_groups'] = security_groups
r = self.ports_post(json_data=dict(port=port), success=201)
return r.json['port']
def create_ports(self, ports):
"""Atomic operation for batch port creation (all or nothing)
:param ports: (list of dicts) {key: ...} with all parameters in the
method create_port, where method mandatory / optional paramteres
respond to mandatory / optional paramters in ports items
:returns: (list of dicts) created portss details
:raises ValueError: if ports parameter is incorrectly formated
:raises ClientError: if the request failed or didn't return 201
msg = 'The ports parameter must be list or tuple'
assert (
isinstance(ports, list) or isinstance(ports, tuple)), msg
for port in ports:
msg = 'Subnet specification %s is not a dict' % port
assert isinstance(port, dict), msg
err = set(port).difference((
'network_id', 'status', 'name', 'admin_state_up',
'mac_address', 'fixed_ips', 'security_groups'))
if err:
raise ValueError(
'Invalid key(s): %s in port specification %s' % (
err, port))
msg = 'network_id is missing in port spec: %s' % port
assert port.get('network_id', None), msg
except AssertionError as ae:
raise ValueError('%s' % ae)
r = self.ports_post(json_data=dict(ports=list(ports)), success=201)
return r.json['ports']
def get_port_details(self, port_id):
r = self.ports_get(port_id, success=201)
return r.json['ports']
def delete_port(self, port_id):
r = self.ports_delete(port_id, success=204)
return r.headers
......@@ -442,6 +442,82 @@ class NetworkingClient(TestCase):
self.assertEqual(self.client.delete_subnet(netid), 'ret headers')
subnets_delete.assert_called_once_with(netid, success=204)
def test_list_ports(self, ports_get):
FakeObject.json = dict(ports='ret val')
self.assertEqual(self.client.list_ports(), 'ret val')
def test_create_port(self, ports_post):
for (
name, status, admin_state_up,
mac_address, fixed_ips, security_groups
) in product(
('name', None), ('status', None), (True, False, None),
('maddr', None), ('some ips', None), ([1, 2, 3], None)):
kwargs = dict(
name=name, status=status, admin_state_up=admin_state_up,
mac_address=mac_address, fixed_ips=fixed_ips,
FakeObject.json, network_id = dict(port='ret val'), 'name'
self.client.create_port(network_id, **kwargs), 'ret val')
req = dict(network_id=network_id)
for k, v in kwargs.items():
if v not in (None, ):
req[k] = v
expargs = dict(json_data=dict(port=req), success=201)
self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
def test_create_ports(self, ports_post):
for ports in (
None, dict(network_id='name'), 'nets', [1, 2, 3], [{'k': 'v'}],
[dict(name=True, mac_address='mac')],
[dict(network_id='n1', invalid='mistake'), ],
[dict(network_id='valid', name='valid'), {'err': 'nop'}]):
ValueError, self.client.create_ports, ports)
FakeObject.json = dict(ports='ret val')
for ports in (
[dict(network_id='n1'), dict(network_id='n 2', name='name')],
dict(network_id='n1', name='n 6', status='status'),
dict(network_id='n 2', admin_state_up=True, fixed_ips='f'),
dict(network_id='n 2', mac_address='mc', name='n.a.m.e.'),
dict(network_id='n-4', security_groups='s~G', name='w. 6'),
dict(network_id='n_5', admin_state_up=False, name='f a')],
dict(network_id='n.e.t', name='c-5'),
dict(network_id='net 2', status='YEAH'))):
self.assertEqual(self.client.create_ports(ports), 'ret val')
expargs = dict(json_data=dict(ports=list(ports)), success=201)
self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
def test_get_port_details(self, ports_get):
portid, FakeObject.json = 'portid', dict(ports='ret val')
self.assertEqual(self.client.get_port_details(portid), 'ret val')
ports_get.assert_called_once_with(portid, success=201)
def test_delete_port(self, ports_delete):
portid, FakeObject.headers = 'portid', 'ret headers'
self.assertEqual(self.client.delete_port(portid), 'ret headers')
ports_delete.assert_called_once_with(portid, success=204)
if __name__ == '__main__':
from sys import argv
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment