1
# Copyright 2014 Citrix Systems
3
# Licensed under the Apache License, Version 2.0 (the "License"); you may
4
# not use this file except in compliance with the License. You may obtain
5
# a copy of the License at
7
# http://www.apache.org/licenses/LICENSE-2.0
9
# Unless required by applicable law or agreed to in writing, software
10
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
# License for the specific language governing permissions and limitations
16
from neutron.tests.unit import testlib_api
19
from neutron_lbaas.services.loadbalancer.drivers.netscaler import ncc_client
20
from neutron_lbaas.services.loadbalancer.drivers.netscaler \
21
import netscaler_driver
23
NCC_CLIENT_CLASS = ('neutron_lbaas.services.loadbalancer.drivers'
24
'.netscaler.ncc_client.NSClient')
26
TESTURI_SCHEME = 'http'
27
TESTURI_HOSTNAME = '1.1.1.1'
29
TESTURI_PATH = '/ncc_service/1.0'
30
TESTURI = '%s://%s:%s%s' % (TESTURI_SCHEME, TESTURI_HOSTNAME,
31
TESTURI_PORT, TESTURI_PATH)
32
TEST_USERNAME = 'user211'
33
TEST_PASSWORD = '@30xHl5cT'
34
TEST_TENANT_ID = '9c5245a2-0432-9d4c-4829-9bd7028603a1'
35
TESTVIP_ID = '52ab5d71-6bb2-457f-8414-22a4ba55efec'
38
class TestNSClient(testlib_api.WebTestCase):
40
"""A Unit test for the NetScaler NCC client module."""
43
self.log = mock.patch.object(ncc_client, 'LOG').start()
44
super(TestNSClient, self).setUp()
45
# mock the requests.request function call
46
self.request_method_mock = mock.Mock()
47
requests.request = self.request_method_mock
48
self.testclient = self._get_nsclient()
50
def test_instantiate_nsclient_with_empty_uri(self):
51
"""Asserts that a call with empty URI will raise an exception."""
52
self.assertRaises(ncc_client.NCCException, ncc_client.NSClient,
53
'', TEST_USERNAME, TEST_PASSWORD)
55
def test_create_resource_with_no_connection(self):
56
"""Asserts that a call with no connection will raise an exception."""
57
# mock a connection object that fails to establish a connection
58
self.request_method_mock.side_effect = (
59
requests.exceptions.ConnectionError())
60
resource_path = netscaler_driver.VIPS_RESOURCE
61
resource_name = netscaler_driver.VIP_RESOURCE
62
resource_body = self._get_testvip_httpbody_for_create()
63
# call method under test: create_resource() and assert that
64
# it raises an exception
65
self.assertRaises(ncc_client.NCCException,
66
self.testclient.create_resource,
67
TEST_TENANT_ID, resource_path,
68
resource_name, resource_body)
70
def test_create_resource_with_error(self):
71
"""Asserts that a failed create call raises an exception."""
72
# create a mock object to represent a valid http response
73
# with a failure status code.
74
fake_response = requests.Response()
75
fake_response.status_code = requests.codes.unauthorized
76
fake_response.headers = []
77
requests.request.return_value = fake_response
78
resource_path = netscaler_driver.VIPS_RESOURCE
79
resource_name = netscaler_driver.VIP_RESOURCE
80
resource_body = self._get_testvip_httpbody_for_create()
81
# call method under test: create_resource
82
# and assert that it raises the expected exception.
83
self.assertRaises(ncc_client.NCCException,
84
self.testclient.create_resource,
85
TEST_TENANT_ID, resource_path,
86
resource_name, resource_body)
88
def test_create_resource(self):
89
"""Asserts that a correct call will succeed."""
90
# obtain the mock object that corresponds to the call of request()
91
fake_response = requests.Response()
92
fake_response.status_code = requests.codes.created
93
fake_response.headers = []
94
self.request_method_mock.return_value = fake_response
95
resource_path = netscaler_driver.VIPS_RESOURCE
96
resource_name = netscaler_driver.VIP_RESOURCE
97
resource_body = self._get_testvip_httpbody_for_create()
98
# call method under test: create_resource()
99
self.testclient.create_resource(TEST_TENANT_ID, resource_path,
100
resource_name, resource_body)
101
# assert that request() was called
102
# with the expected params.
103
resource_url = "%s/%s" % (self.testclient.service_uri, resource_path)
104
self.request_method_mock.assert_called_once_with(
110
def test_update_resource_with_error(self):
111
"""Asserts that a failed update call raises an exception."""
112
# create a valid http response with a failure status code.
113
fake_response = requests.Response()
114
fake_response.status_code = requests.codes.unauthorized
115
fake_response.headers = []
116
# obtain the mock object that corresponds to the call of request()
117
self.request_method_mock.return_value = fake_response
118
resource_path = "%s/%s" % (netscaler_driver.VIPS_RESOURCE,
120
resource_name = netscaler_driver.VIP_RESOURCE
121
resource_body = self._get_testvip_httpbody_for_update()
122
# call method under test: update_resource() and
123
# assert that it raises the expected exception.
124
self.assertRaises(ncc_client.NCCException,
125
self.testclient.update_resource,
126
TEST_TENANT_ID, resource_path,
127
resource_name, resource_body)
129
def test_update_resource(self):
130
"""Asserts that a correct update call will succeed."""
131
# create a valid http response with a successful status code.
132
fake_response = requests.Response()
133
fake_response.status_code = requests.codes.ok
134
fake_response.headers = []
135
# obtain the mock object that corresponds to the call of request()
136
self.request_method_mock.return_value = fake_response
137
resource_path = "%s/%s" % (netscaler_driver.VIPS_RESOURCE,
139
resource_name = netscaler_driver.VIP_RESOURCE
140
resource_body = self._get_testvip_httpbody_for_update()
141
# call method under test: update_resource.
142
self.testclient.update_resource(TEST_TENANT_ID, resource_path,
143
resource_name, resource_body)
144
resource_url = "%s/%s" % (self.testclient.service_uri, resource_path)
145
# assert that requests.request() was called with the
147
self.request_method_mock.assert_called_once_with(
153
def test_delete_resource_with_error(self):
154
"""Asserts that a failed delete call raises an exception."""
155
# create a valid http response with a failure status code.
156
fake_response = requests.Response()
157
fake_response.status_code = requests.codes.unauthorized
158
fake_response.headers = []
159
resource_path = "%s/%s" % (netscaler_driver.VIPS_RESOURCE,
161
# call method under test: create_resource
162
self.assertRaises(ncc_client.NCCException,
163
self.testclient.remove_resource,
164
TEST_TENANT_ID, resource_path)
166
def test_delete_resource(self):
167
"""Asserts that a correct delete call will succeed."""
168
# create a valid http response with a failure status code.
169
fake_response = requests.Response()
170
fake_response.status_code = requests.codes.ok
171
fake_response.headers = []
172
# obtain the mock object that corresponds to the call of request()
173
self.request_method_mock.return_value = fake_response
174
resource_path = "%s/%s" % (netscaler_driver.VIPS_RESOURCE,
176
resource_url = "%s/%s" % (self.testclient.service_uri, resource_path)
177
# call method under test: create_resource
178
self.testclient.remove_resource(TEST_TENANT_ID, resource_path)
179
# assert that httplib.HTTPConnection request() was called with the
181
self.request_method_mock.assert_called_once_with(
187
def _get_nsclient(self):
188
return ncc_client.NSClient(TESTURI, TEST_USERNAME, TEST_PASSWORD)
190
def _get_testvip_httpbody_for_create(self):
193
'address': '10.0.0.3',
194
'pool_id': 'da477c13-24cd-4c9f-8c19-757a61ef3b9d',
197
'admin_state_up': True,
201
def _get_testvip_httpbody_for_update(self):
203
body['name'] = 'updated vip1'
204
body['admin_state_up'] = False
added
removed
neutron_lbaas/tests/unit/services/loadbalancer/drivers/netscaler/test_ncc_client.py
