# Copyright 2015 Infoblox Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import time
import unittest
import os
from unittest import mock
import requests
from unittest.mock import patch
from requests import exceptions as req_exc
from requests.cookies import RequestsCookieJar
try:
from oslo_serialization import jsonutils
except ImportError: # pragma: no cover
import json as jsonutils
from infoblox_client import connector
from infoblox_client import exceptions
class TestInfobloxConnector(unittest.TestCase):
def setUp(self):
super(TestInfobloxConnector, self).setUp()
self.default_opts = self._prepare_options()
self.connector = connector.Connector(self.default_opts)
@staticmethod
def _create_infoblox_csv():
# build data file for file upload test
file_data = [
'header-network,address,netmask,comment'
'network,10.10.10.0,255.255.255.0,test1',
'network,10.10.11.0,255.255.255.0,test2'
]
with open('tests/ibx_networks.csv', 'w') as fh:
fh.write('\n'.join(file_data))
fh.close()
@staticmethod
def _delete_infoblox_csv():
if os.path.exists('tests/ibx_networks.csv'):
os.unlink('tests/ibx_networks.csv')
@staticmethod
def _prepare_options():
opts = mock.Mock()
opts.host = 'infoblox.example.org'
opts.wapi_version = '1.1'
opts.username = 'admin'
opts.password = 'password'
opts.ssl_verify = False
opts.silent_ssl_warnings = True
opts.http_pool_connections = 10
opts.http_pool_maxsize = 10
opts.http_request_timeout = 10
opts.max_retries = 3
opts.max_results = None
opts.paging = False
return opts
def test_create_object(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_create:
cookie_jar = RequestsCookieJar()
cookie_jar.set('ibapauth',
'ctime={},user=admin,group=admin-group,auth=LOCAL,client=API,mtime=1731651582,su=1,ip=localhost,timeout=60,mjuHjy8l1tY0GhSf+aRcxI7rybaIONUIpjc'.format(
int(time.time())), domain='infoblox.localhost')
self.connector.session.cookies = cookie_jar
patched_create.return_value.status_code = 201
patched_create.return_value.content = '{}'
self.connector.create_object(objtype, payload)
patched_create.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network',
data=jsonutils.dumps(payload),
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
# test if cookies have been set
self.assertEqual(None, self.connector.session.auth)
def test_create_object_with_extattrs(self):
objtype = 'network'
payload = {
'extattrs': {
'Subnet ID': {'value': 'fake_subnet_id'}
},
'ip': '0.0.0.0',
}
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_create:
patched_create.return_value.status_code = 201
patched_create.return_value.content = '{}'
self.connector.create_object(objtype, payload)
patched_create.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network',
data=jsonutils.dumps(payload),
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_create_object_raises_member_assigned(self):
nios_error = (
'{ "Error": "AdmConDataError: None (IBDataConflictError:'
'IB.Data.Conflict:Member 10.39.12.91 is assigned to another '
'network view \'test2\')",'
'"code": "Client.Ibap.Data.Conflict",'
'"text": "Member 10.39.12.91 is assigned to another '
'network view \'test2\'"}')
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_create:
patched_create.return_value.status_code = 400
patched_create.return_value.content = nios_error
self.assertRaises(exceptions.InfobloxMemberAlreadyAssigned,
self.connector.create_object,
'network', {'network': '192.178.1.0/24'})
def test_get_object(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(objtype, payload)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?ip=0.0.0.0',
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_get_objects_with_extattrs(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
extattrs = {
'Subnet ID': {'value': 'fake_subnet_id'}
}
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(objtype, payload, extattrs=extattrs)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?%2ASubnet+ID=fake_subnet_id&ip=0.0.0.0', # noqa: E501
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_get_object_with_default_and_extattrs(self):
objtype = 'network'
extattrs = {'Subnet ID': {'value': 'fake_subnet_id'}}
return_fields = ['default', 'extattrs']
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(
objtype,
extattrs=extattrs,
return_fields=return_fields
)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?%2ASubnet+ID=fake_subnet_id&_return_fields%2B=extattrs',
# noqa: E501
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_get_object_with_specific_return_fields(self):
objtype = 'network'
extattrs = {'Subnet ID': {'value': 'fake_subnet_id'}}
return_fields = ['extattrs']
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(
objtype,
extattrs=extattrs,
return_fields=return_fields
)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?%2ASubnet+ID=fake_subnet_id&_return_fields=extattrs',
# noqa: E501
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_get_objects_with_max_results(self):
objtype = 'network'
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(objtype, {}, max_results=20)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?_max_results=20', # noqa: E501
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_get_objects_with_max_results_as_connector_opt(self):
objtype = 'network'
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
opts = self._prepare_options()
opts.max_results = 10
conn = connector.Connector(opts)
conn.get_object(objtype, {})
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/'
'v1.1/network?_max_results=10',
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_max_results_priority(self):
objtype = 'network'
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
opts = self._prepare_options()
opts.max_results = 10
conn = connector.Connector(opts)
# max_results passed to get_object should have higher priority
# over max_results connector option
conn.get_object(objtype, {}, max_results=-20)
patched_get.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network?_max_results=-20', # noqa: E501
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_update_object(self):
ref = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'put',
return_value=mock.Mock()) as patched_update:
patched_update.return_value.status_code = 200
patched_update.return_value.content = '{}'
self.connector.update_object(ref, payload)
patched_update.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network',
data=jsonutils.dumps(payload),
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_update_object_with_http_error(self):
ref = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'put',
return_value=mock.Mock()) as patched_update:
patched_update.return_value.status_code = 400
patched_update.return_value.content = '{}'
self.assertRaises(exceptions.InfobloxCannotUpdateObject,
self.connector.update_object, ref, payload)
def test_update_object_with_http_error_503(self):
ref = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'put',
return_value=mock.Mock()) as patched_update:
patched_update.return_value.status_code = 503
patched_update.return_value.content = 'Temporary Unavailable'
self.assertRaises(exceptions.InfobloxGridTemporaryUnavailable,
self.connector.update_object, ref, payload)
def test_delete_object(self):
ref = 'network'
with patch.object(requests.Session, 'delete',
return_value=mock.Mock()) as patched_delete:
patched_delete.return_value.status_code = 200
patched_delete.return_value.content = '{}'
self.connector.delete_object(ref)
patched_delete.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/network',
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_delete_object_with_http_error(self):
ref = 'network'
with patch.object(requests.Session, 'delete',
return_value=mock.Mock()) as patched_delete:
patched_delete.return_value.status_code = 400
patched_delete.return_value.content = '{}'
self.assertRaises(exceptions.InfobloxCannotDeleteObject,
self.connector.delete_object, ref)
def test_delete_object_with_http_error_503(self):
ref = 'network'
with patch.object(requests.Session, 'delete',
return_value=mock.Mock()) as patched_delete:
patched_delete.return_value.status_code = 503
patched_delete.return_value.content = 'Temporary Unavailable'
self.assertRaises(exceptions.InfobloxGridTemporaryUnavailable,
self.connector.delete_object, ref)
def test_construct_url_absolute_path_fails(self):
pathes = ('/starts_with_slash', '', None)
for path in pathes:
self.assertRaises(ValueError,
self.connector._construct_url, path)
def test_construct_url_with_query_params_and_extattrs(self):
query_params = {'some_option': 'some_value'}
ext_attrs = {'Subnet ID': {'value': 'fake_subnet_id'}}
url = self.connector._construct_url('network',
query_params=query_params,
extattrs=ext_attrs)
self.assertEqual(
'https://infoblox.example.org/wapi/v1.1/network?%2ASubnet+ID=fake_subnet_id&some_option=some_value',
# noqa: E501
url)
def test_construct_url_with_query_params_containing_array(self):
query_params = {'array_option': ['value1', 'value2']}
url = self.connector._construct_url('network',
query_params=query_params)
self.assertEqual('https://infoblox.example.org/wapi/v1.1/network?array_option=value1&array_option=value2',
# noqa: E501
url)
def test_construct_url_with_force_proxy(self):
ext_attrs = {'Subnet ID': {'value': 'fake_subnet_id'}}
url = self.connector._construct_url('network',
extattrs=ext_attrs,
force_proxy=True)
self.assertEqual('https://infoblox.example.org/wapi/v1.1/network?%2ASubnet+ID=fake_subnet_id&_proxy_search=GM',
# noqa: E501
url)
def test_get_object_with_proxy_flag(self):
self.connector._get_object = mock.MagicMock(return_value=None)
self.connector._construct_url = mock.MagicMock()
self.connector.cloud_api_enabled = True
result = self.connector.get_object('network', force_proxy=True)
self.assertEqual(None, result)
self.connector._construct_url.assert_called_with('network',
{},
None,
force_proxy=True)
self.connector._get_object.assert_called_with('network', mock.ANY)
def test_get_object_without_proxy_flag(self):
self.connector._get_object = mock.MagicMock(
side_effect=[requests.HTTPError(), None],
)
self.connector._construct_url = mock.MagicMock()
self.connector.cloud_api_enabled = True
result = self.connector.get_object('network')
self.assertEqual(None, result)
construct_calls = [mock.call('network', {}, None, force_proxy=False),
mock.call('network', {}, None, force_proxy=True)]
self.connector._construct_url.assert_has_calls(construct_calls)
def test__get_object_raises_search_error(self):
url = 'http://some-url/'
response = requests.Response()
response.status_code = 404
response._content = 'Object not found'
response.url = url
self.connector.session = mock.Mock()
self.connector.session.get.return_value = response
self.connector.session.cookies = RequestsCookieJar()
with self.assertRaises(requests.HTTPError):
self.connector._get_object('network', url)
def test_get_object_with_pagination_with_no_result(self):
self.connector._get_object = mock.MagicMock(return_value=None)
result = self.connector.get_object('network', paging=True)
self.assertEqual(None, result)
def test_get_object_with_pagination_with_result(self):
self.connector._get_object = mock.MagicMock(return_value={"result": ["data"]}) # noqa: E501
result = self.connector.get_object('network', paging=True)
self.assertEqual(["data"], result)
def test_get_object_override_paging(self):
"""Check if paging argument will override Connector.paging attribute
when calling get_object method"""
self.connector._get_object = mock.MagicMock()
self.connector._build_query_params = mock.MagicMock()
self.connector.paging = True
self.connector.get_object(
'network',
paging=False,
)
self.connector._build_query_params.assert_called_once_with(
payload=None,
return_fields=None,
max_results=None,
paging=False,
)
def test__handle_get_object_with_pagination_with_no_record(self):
query_params = {"_paging": 1,
"_return_as_object": 1,
"_max_results": 100}
self.connector._get_object = mock.MagicMock(return_value=None)
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual(None, result)
def test__handle_get_object_with_max_results_negative(self):
query_params = {"_paging": 1,
"_return_as_object": 1,
"_max_results": -100}
self.connector._get_object = mock.MagicMock(return_value=None)
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual(None, result)
def test__handle_get_object_with_pagination_with_record(self):
query_params = {"_paging": 1,
"_return_as_object": 1,
"_max_results": 100}
self.connector._get_object = mock.MagicMock(return_value={"result": ["data"]}) # noqa: E501
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual(["data"], result)
def _get_object(self, url, **opts):
resp = requests.Response
resp.status_code = 200
if "_page_id" in url:
resp.content = jsonutils.dumps({"result": [6, 7, 8, 9, 10]})
else:
resp.content = jsonutils.dumps({"result": [1, 2, 3, 4, 5], "next_page_id": 1}) # noqa: E501
return resp
def test__handle_get_object_with_record_more_than_max_results_paging(self):
query_params = {"_paging": 1,
"_return_as_object": 1,
"_max_results": 5}
with patch.object(requests.Session, 'get') as patched_get:
patched_get.side_effect = self._get_object
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], result)
def test__handle_get_object_without_pagination(self):
query_params = {"_max_results": 100}
self.connector._get_object = mock.MagicMock(return_value=None)
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual(None, result)
def test__handle_get_object_without_pagination_with_record(self):
query_params = {"_max_results": 100}
self.connector._get_object = mock.MagicMock(return_value=["data"])
result = self.connector._handle_get_object("network", query_params,
None, False)
self.assertEqual(["data"], result)
def test_call_func(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_call_func:
patched_call_func.return_value.status_code = 201
patched_call_func.return_value.content = '{}'
self.connector.call_func(objtype, "_ref", payload)
patched_call_func.assert_called_once_with(
'https://infoblox.example.org/wapi/v1.1/_ref?_function=network', # noqa: E501
data=jsonutils.dumps(payload),
headers=self.connector.DEFAULT_HEADER,
timeout=self.default_opts.http_request_timeout,
verify=self.default_opts.ssl_verify,
)
def test_call_upload_file(self):
upload_file_path = '/http_direct_file_io/req_id-UPLOAD-0302163936014609/ibx_networks.csv' # noqa: E501
upload_url = 'https://infoblox.example.org' + upload_file_path
self._create_infoblox_csv()
with open('tests/ibx_networks.csv', 'r') as fh:
data = fh.read()
fh.close()
payload = dict(file=data)
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_post:
cookie_jar = RequestsCookieJar()
cookie_jar.set(
'ibapauth',
'ctime={},user=admin,group=admin-group,auth=LOCAL,client=API,mtime=1731651582,su=1,ip=localhost,timeout=60,mjuHjy8l1tY0GhSf+aRcxI7rybaIONUIpjc'.format(
int(time.time())), domain='infoblox.localhost')
self.connector.session.cookies = cookie_jar
patched_post.return_value.status_code = 200
patched_post.return_value.content = '{}'
self.connector.upload_file(upload_url, payload)
self.assertEqual(None, self.connector.session.auth)
self._delete_infoblox_csv()
def test_call_upload_file_with_error_403(self):
upload_file_path = '/http_direct_file_io/req_id-UPLOAD-0302163936014609/ibx_networks.csv' # noqa: E501
upload_url = 'https://infoblox.example.org' + upload_file_path
self._create_infoblox_csv()
with open('tests/ibx_networks.csv', 'r') as fh:
data = fh.read()
fh.close()
payload = dict(file=data)
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_post:
patched_post.return_value.status_code = 403
patched_post.return_value.content = '{}'
self.assertRaises(exceptions.InfobloxFileUploadFailed,
self.connector.upload_file,
upload_url,
payload)
self._delete_infoblox_csv()
def test_call_download_file(self):
download_file_path = '/http_direct_file_io/req_id-DOWNLOAD-0302163936014609/ibx_networks.csv' # noqa: E501
download_url = 'https://infoblox.example.org' + download_file_path
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
cookie_jar = RequestsCookieJar()
cookie_jar.set(
'ibapauth',
'ctime={},user=admin,group=admin-group,auth=LOCAL,client=API,mtime=1731651582,su=1,ip=localhost,timeout=60,mjuHjy8l1tY0GhSf+aRcxI7rybaIONUIpjc'.format(
int(time.time())), domain='infoblox.localhost')
self.connector.session.cookies = cookie_jar
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.download_file(download_url)
self.assertEqual(None, self.connector.session.auth)
def test_call_download_file_with_error_403(self):
download_file_path = '/http_direct_file_io/req_id-DOWNLOAD-0302163936014609/ibx_networks.csv' # noqa: E501
download_url = 'https://infoblox.example.org' + download_file_path
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
patched_get.return_value.status_code = 403
patched_get.return_value.content = '{}'
self.assertRaises(exceptions.InfobloxFileDownloadFailed,
self.connector.download_file,
download_url)
def test_call_func_with_http_error(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_call_func:
patched_call_func.return_value.status_code = 400
patched_call_func.return_value.content = '{}'
self.assertRaises(exceptions.InfobloxFuncException,
self.connector.call_func,
objtype,
"_ref",
payload)
def test_call_func_with_http_error_503(self):
objtype = 'network'
payload = {'ip': '0.0.0.0'}
with patch.object(requests.Session, 'post',
return_value=mock.Mock()) as patched_call_func:
patched_call_func.return_value.status_code = 503
patched_call_func.return_value.content = 'Temporary Unavailable'
self.assertRaises(exceptions.InfobloxGridTemporaryUnavailable,
self.connector.call_func,
objtype,
"_ref",
payload)
def test__check_service_availability(self):
resp = requests.Response
resp.status_code = 503
resp.content = 'Temporary Unavailable'
self.assertRaises(exceptions.InfobloxGridTemporaryUnavailable,
self.connector._check_service_availability,
"delete",
resp,
'_ref')
def test_get_object_with_cookies(self):
objtype = 'network'
with patch.object(requests.Session, 'get',
return_value=mock.Mock()) as patched_get:
cookie_jar = RequestsCookieJar()
cookie_jar.set('ibapauth',
'ctime={},user=admin,group=admin-group,auth=LOCAL,client=API,mtime=1731651582,su=1,ip=localhost,timeout=60,mjuHjy8l1tY0GhSf+aRcxI7rybaIONUIpjc'.format(
int(time.time())), domain='infoblox.localhost')
self.connector.session.cookies = cookie_jar
patched_get.return_value.status_code = 200
patched_get.return_value.content = '{}'
self.connector.get_object(objtype, {})
self.assertEqual(None, self.connector.session.auth)
class TestInfobloxConnectorStaticMethods(unittest.TestCase):
def test_neutron_exception_is_raised_on_any_request_error(self):
# timeout exception raises InfobloxTimeoutError
f = mock.Mock()
f.__name__ = 'mock'
f.side_effect = req_exc.Timeout
self.assertRaises(exceptions.InfobloxTimeoutError,
connector.reraise_neutron_exception(f))
# all other request exception raises InfobloxConnectionError
supported_exceptions = [req_exc.HTTPError,
req_exc.ConnectionError,
req_exc.ProxyError,
req_exc.SSLError,
req_exc.TooManyRedirects,
req_exc.InvalidURL]
for ex in supported_exceptions:
f.side_effect = ex
self.assertRaises(exceptions.InfobloxConnectionError,
connector.reraise_neutron_exception(f))
def test_exception_raised_for_non_authorized(self):
response = mock.Mock()
response.status_code = requests.codes.UNAUTHORIZED
response.content = (b''
b'\n\n401 Unauthorized\n'
b'\nUnauthorized
'
b'\nThis server could not verify that you'
b'\nare authorized to access the document'
b'\nrequested. Either you supplied the wrong'
b'\ncredentials (e.g., bad password), or your'
b'\nbrowser doesn\'t understand how to supply'
b'\nthe credentials required.
\n\n')
self.assertRaises(exceptions.InfobloxBadWAPICredential,
connector.Connector._validate_authorized,
response)
def test_no_exceptions_for_ok_statuses(self):
response = mock.Mock()
ok_statuses = (requests.codes.OK,
requests.codes.CREATED,
requests.codes.ACCEPTED)
for status_code in ok_statuses:
response.status_code = status_code
connector.Connector._validate_authorized(response)
def test_non_cloud_api_detection(self):
wapi_not_cloud = ('1.4.1', '1.9/', '1.99', 'asd', 'v1.4')
for url in wapi_not_cloud:
self.assertFalse(
connector.Connector.is_cloud_wapi(url))
def test_cloud_api_detection(self):
wapi_cloud = ('2.1/', '/2.0/', '2.0.1',
'3.0/', '11.0.1/', 'v2.1', 'v2.0')
for url in wapi_cloud:
self.assertTrue(
connector.Connector.is_cloud_wapi(url))
def test_allow_options_as_dict(self):
opts = dict(host='infoblox.example.org',
wapi_version='1.1',
username='admin',
password='password',
ssl_verify=False,
silent_ssl_warnings=True,
max_results=50,
http_pool_connections=10,
http_pool_maxsize=10,
http_request_timeout=10)
conn = connector.Connector(opts)
self.assertEqual(opts['host'], conn.host)
self.assertEqual(opts['wapi_version'], conn.wapi_version)
self.assertEqual(opts['username'], conn.username)
self.assertEqual(opts['password'], conn.password)
self.assertEqual(opts['ssl_verify'], conn.ssl_verify)
self.assertEqual(opts['silent_ssl_warnings'],
conn.silent_ssl_warnings)
self.assertEqual(opts['max_results'],
conn.max_results)
self.assertEqual(opts['http_pool_connections'],
conn.http_pool_connections)
self.assertEqual(opts['http_pool_maxsize'], conn.http_pool_maxsize)
self.assertEqual(opts['http_request_timeout'],
conn.http_request_timeout)
def test_incomplete_options_raises_exception(self):
opts = dict(host='infoblox.example.org',
wapi_version='1.1')
self.assertRaises(exceptions.InfobloxConfigException,
connector.Connector, opts)
def test_default_options(self):
opts = dict(host='infoblox.example.org',
username='admin',
password='password')
conn = connector.Connector(opts)
self.assertEqual(False, conn.ssl_verify)
self.assertEqual(False, conn.silent_ssl_warnings)
self.assertEqual(10, conn.http_request_timeout)
self.assertEqual(10, conn.http_pool_connections)
self.assertEqual(10, conn.http_pool_maxsize)
self.assertEqual(3, conn.max_retries)
self.assertEqual('2.10', conn.wapi_version)
self.assertEqual(None, conn.max_results)
def test_blank_values_not_allowed(self):
base_dict = {'host': '192.168.1.15',
'username': 'admin',
'password': 'pass'}
for field in base_dict:
test_dict = base_dict.copy()
test_dict[field] = None
self.assertRaises(exceptions.InfobloxConfigException,
connector.Connector, test_dict)
def test_blank_values_not_allowed_cert_auth(self):
"""
Checks if connector's _parse_options method raises
exception if one of the host/cert/key is not provided
"""
base_dict = {'host': '192.168.1.15',
'cert': 'cert',
'key': 'key'}
for field in base_dict:
test_dict = base_dict.copy()
test_dict[field] = None
self.assertRaises(exceptions.InfobloxConfigException,
connector.Connector, test_dict)
def test_is_cloud_wapi_raises_exception(self):
for value in (None, '', 0, 1, self, 1.2):
self.assertRaises(ValueError,
connector.Connector.is_cloud_wapi,
value)
def test__parse_reply_raises_connection_error(self):
request = mock.Mock()
request.content = (''
'Some error reply\n')
self.assertRaises(exceptions.InfobloxConnectionError,
connector.Connector._parse_reply,
request)
def test__parse_reply(self):
request = mock.Mock()
request.content = (
'[{"_ref": "network/ZG5zLm5ldHdvcmskMTAuNDAuMjUuMC8yNC8w:'
'10.40.25.0/24/default","network": "10.40.25.0/24",'
'"network_view": "default"}]')
expected_reply = [
{'_ref': "network/ZG5zLm5ldHdvcmskMTAuNDAuMjUuMC8yNC8w"
":10.40.25.0/24/default",
'network': "10.40.25.0/24",
'network_view': "default"}]
parsed_reply = connector.Connector._parse_reply(request)
self.assertEqual(expected_reply, parsed_reply)
def test_session_auth(self):
"""
Checks if connector's session is configured,
when username and password are provided
"""
# Case 1: Only username and password are provided
options = {'host': '192.168.1.15',
'username': 'admin',
'password': 'pass'}
conn = connector.Connector(options)
self.assertEqual(conn.session.auth, ('admin', 'pass'))
self.assertEqual(conn.session.cert, None)
# Case 2: Username, password, cert and key are
# provided. Connector should use username and password.
options = {'host': '192.168.1.15',
'username': 'admin',
'password': 'pass',
'cert': 'cert',
'key': 'key'}
conn = connector.Connector(options)
self.assertEqual(conn.session.auth, ('admin', 'pass'))
self.assertEqual(conn.session.cert, None)
def test_session_cert(self):
"""
Checks if connector's session is configured,
when cert and key are provided
"""
options = {'host': '192.168.1.15',
'cert': 'cert',
'key': 'key'}
conn = connector.Connector(options)
self.assertEqual(conn.session.auth, None)
self.assertEqual(conn.session.cert, ('cert', 'key'))