Commit 77d1b504 authored by Stavros Sachtouris's avatar Stavros Sachtouris
Browse files

Implement floating_ip(s)post/get/delete rest calls

Refs: #3862
parent 0e8a561c
CHANGELOG for version 0.10 CHANGELOG for version 0.10
Bug Fixes:
Changes:
Features:
...@@ -241,3 +241,23 @@ class ComputeRestClient(Client): ...@@ -241,3 +241,23 @@ class ComputeRestClient(Client):
def floating_ip_pools_get(self, tenant_id, success=200, **kwargs): def floating_ip_pools_get(self, tenant_id, success=200, **kwargs):
path = path4url(tenant_id, 'os-floating-ip-pools') path = path4url(tenant_id, 'os-floating-ip-pools')
return self.get(path, success=success, **kwargs) return self.get(path, success=success, **kwargs)
def floating_ips_get(self, tenant_id, success=200, **kwargs):
path = path4url(tenant_id, 'os-floating-ips')
return self.get(path, success=success, **kwargs)
def floating_ips_post(self, tenant_id, json_data, success=201, **kwargs):
path = path4url(tenant_id, 'os-floating-ips')
if json_data is not None:
json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(json_data))
return self.post(path, data=json_data, success=success, **kwargs)
def floating_ip_get(self, tenant_id, success=200, **kwargs):
path = path4url(tenant_id, 'os-floating-ip')
return self.get(path, success=success, **kwargs)
def floating_ip_delete(self, tenant_id, success=204, **kwargs):
path = path4url(tenant_id, 'os-floating-ip')
return self.delete(path, success=success, **kwargs)
...@@ -240,6 +240,65 @@ class ComputeRestClient(TestCase): ...@@ -240,6 +240,65 @@ class ComputeRestClient(TestCase):
'/%s/os-floating-ip-pools' % tenant_id, '/%s/os-floating-ip-pools' % tenant_id,
success=success, **kwargs)) success=success, **kwargs))
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ips_get(self, get):
for args in product(
('tenant1', 'tenant2'),
(200, 204),
({}, {'k': 'v'})):
tenant_id, success, kwargs = args
r = self.client.floating_ips_get(tenant_id, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(get.mock_calls[-1], call(
'/%s/os-floating-ips' % tenant_id,
success=success, **kwargs))
@patch('%s.set_header' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
def test_floating_ips_post(self, post, SH):
for args in product(
('tenant1', 'tenant2'),
(None, [dict(json="data"), dict(data="json")]),
(202, 204),
({}, {'k': 'v'})):
(tenant_id, json_data, success, kwargs) = args
self.client.floating_ips_post(*args[:3], **kwargs)
if json_data:
json_data = dumps(json_data)
self.assertEqual(SH.mock_calls[-2:], [
call('Content-Type', 'application/json'),
call('Content-Length', len(json_data))])
self.assertEqual(post.mock_calls[-1], call(
'/%s/os-floating-ips' % tenant_id,
data=json_data, success=success,
**kwargs))
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ip_get(self, get):
for args in product(
('tenant1', 'tenant2'),
(200, 204),
({}, {'k': 'v'})):
tenant_id, success, kwargs = args
r = self.client.floating_ip_get(tenant_id, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(get.mock_calls[-1], call(
'/%s/os-floating-ip' % tenant_id,
success=success, **kwargs))
@patch('%s.delete' % rest_pkg, return_value=FR())
def test_floating_ip_delete(self, delete):
for args in product(
('tenant1', 'tenant2'),
(204,),
({}, {'k': 'v'})):
tenant_id, success, kwargs = args
r = self.client.floating_ip_delete(tenant_id, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(delete.mock_calls[-1], call(
'/%s/os-floating-ip' % tenant_id,
success=success, **kwargs))
class ComputeClient(TestCase): class ComputeClient(TestCase):
......
...@@ -169,3 +169,23 @@ class CycladesRestClient(ComputeClient): ...@@ -169,3 +169,23 @@ class CycladesRestClient(ComputeClient):
def floating_ip_pools_get(self, success=200, **kwargs): def floating_ip_pools_get(self, success=200, **kwargs):
path = path4url('os-floating-ip-pools') path = path4url('os-floating-ip-pools')
return self.get(path, success=success, **kwargs) return self.get(path, success=success, **kwargs)
def floating_ips_get(self, success=200, **kwargs):
path = path4url('os-floating-ips')
return self.get(path, success=success, **kwargs)
def floating_ips_post(self, json_data, success=201, **kwargs):
path = path4url('os-floating-ips')
if json_data is not None:
json_data = json.dumps(json_data)
self.set_header('Content-Type', 'application/json')
self.set_header('Content-Length', len(json_data))
return self.post(path, data=json_data, success=success, **kwargs)
def floating_ip_get(self, floating_ip_id, success=200, **kwargs):
path = path4url('os-floating-ip', floating_ip_id)
return self.get(path, success=success, **kwargs)
def floating_ip_delete(self, floating_ip_id, success=200, **kwargs):
path = path4url('os-floating-ip', floating_ip_id)
return self.delete(path, success=success, **kwargs)
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
from mock import patch, call from mock import patch, call
from unittest import TestCase from unittest import TestCase
from itertools import product from itertools import product
from json import dumps
from kamaki.clients import ClientError, cyclades from kamaki.clients import ClientError, cyclades
...@@ -215,6 +216,60 @@ class CycladesRestClient(TestCase): ...@@ -215,6 +216,60 @@ class CycladesRestClient(TestCase):
self.assertEqual(get.mock_calls[-1], call( self.assertEqual(get.mock_calls[-1], call(
'/os-floating-ip-pools', success=success, **kwargs)) '/os-floating-ip-pools', success=success, **kwargs))
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ips_get(self, get):
for args in product(
(200, 204),
({}, {'k': 'v'})):
success, kwargs = args
r = self.client.floating_ips_get(success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(get.mock_calls[-1], call(
'/os-floating-ips', success=success, **kwargs))
@patch('%s.set_header' % rest_pkg)
@patch('%s.post' % rest_pkg, return_value=FR())
def test_floating_ips_post(self, post, SH):
for args in product(
(None, [dict(json="data"), dict(data="json")]),
(202, 204),
({}, {'k': 'v'})):
(json_data, success, kwargs) = args
self.client.floating_ips_post(*args[:2], **kwargs)
if json_data:
json_data = dumps(json_data)
self.assertEqual(SH.mock_calls[-2:], [
call('Content-Type', 'application/json'),
call('Content-Length', len(json_data))])
self.assertEqual(post.mock_calls[-1], call(
'/os-floating-ips',
data=json_data, success=success,
**kwargs))
@patch('%s.get' % rest_pkg, return_value=FR())
def test_floating_ip_get(self, get):
for args in product(
('fip1', 'fip2'),
(200, 204),
({}, {'k': 'v'})):
fip, success, kwargs = args
r = self.client.floating_ip_get(fip, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(get.mock_calls[-1], call(
'/os-floating-ip/%s' % fip, success=success, **kwargs))
@patch('%s.delete' % rest_pkg, return_value=FR())
def test_floating_ip_delete(self, delete):
for args in product(
('fip1', 'fip2'),
(200, 204),
({}, {'k': 'v'})):
fip, success, kwargs = args
r = self.client.floating_ip_delete(fip, success, **kwargs)
self.assertTrue(isinstance(r, FR))
self.assertEqual(delete.mock_calls[-1], call(
'/os-floating-ip/%s' % fip, success=success, **kwargs))
class CycladesClient(TestCase): class CycladesClient(TestCase):
......
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