Source code for ceilometer.tests.unit.network.statistics.opendaylight.test_client
#
# Copyright 2013 NEC Corporation. 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 mock
from oslo_config import fixture as config_fixture
from oslotest import base
from requests import auth as req_auth
import six
from six.moves.urllib import parse as urlparse
from ceilometer.i18n import _
from ceilometer.network.statistics.opendaylight import client
from ceilometer import service as ceilometer_service
[docs]class TestClientHTTPBasicAuth(base.BaseTestCase):
auth_way = 'basic'
scheme = 'http'
[docs] def setUp(self):
super(TestClientHTTPBasicAuth, self).setUp()
self.conf = self.useFixture(config_fixture.Config())
ceilometer_service.prepare_service(argv=[], config_files=[])
self.parsed_url = urlparse.urlparse(
'http://127.0.0.1:8080/controller/nb/v2?container_name=default&'
'container_name=egg&auth=%s&user=admin&password=admin_pass&'
'scheme=%s' % (self.auth_way, self.scheme))
self.params = urlparse.parse_qs(self.parsed_url.query)
self.endpoint = urlparse.urlunparse(
urlparse.ParseResult(self.scheme,
self.parsed_url.netloc,
self.parsed_url.path,
None, None, None))
odl_params = {'auth': self.params.get('auth')[0],
'user': self.params.get('user')[0],
'password': self.params.get('password')[0]}
self.client = client.Client(self.endpoint, odl_params)
self.resp = mock.MagicMock()
self.get = mock.patch('requests.get',
return_value=self.resp).start()
self.resp.raw.version = 1.1
self.resp.status_code = 200
self.resp.reason = 'OK'
self.resp.headers = {}
self.resp.content = 'dummy'
def _test_request(self, method, url):
data = method('default')
call_args = self.get.call_args_list[0][0]
call_kwargs = self.get.call_args_list[0][1]
# check url
real_url = url % {'container_name': 'default',
'scheme': self.scheme}
self.assertEqual(real_url, call_args[0])
# check auth parameters
auth = call_kwargs.get('auth')
if self.auth_way == 'digest':
self.assertIsInstance(auth, req_auth.HTTPDigestAuth)
else:
self.assertIsInstance(auth, req_auth.HTTPBasicAuth)
self.assertEqual('admin', auth.username)
self.assertEqual('admin_pass', auth.password)
# check header
self.assertEqual(
{'Accept': 'application/json'},
call_kwargs['headers'])
# check return value
self.assertEqual(self.get().json(), data)
[docs] def test_flow_statistics(self):
self._test_request(
self.client.statistics.get_flow_statistics,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/statistics/%(container_name)s/flow')
[docs] def test_port_statistics(self):
self._test_request(
self.client.statistics.get_port_statistics,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/statistics/%(container_name)s/port')
[docs] def test_table_statistics(self):
self._test_request(
self.client.statistics.get_table_statistics,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/statistics/%(container_name)s/table')
[docs] def test_topology(self):
self._test_request(
self.client.topology.get_topology,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/topology/%(container_name)s')
[docs] def test_user_links(self):
self._test_request(
self.client.topology.get_user_links,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/topology/%(container_name)s/userLinks')
[docs] def test_switch(self):
self._test_request(
self.client.switch_manager.get_nodes,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/switchmanager/%(container_name)s/nodes')
[docs] def test_active_hosts(self):
self._test_request(
self.client.host_tracker.get_active_hosts,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/hosttracker/%(container_name)s/hosts/active')
[docs] def test_inactive_hosts(self):
self._test_request(
self.client.host_tracker.get_inactive_hosts,
'%(scheme)s://127.0.0.1:8080/controller/nb/v2'
'/hosttracker/%(container_name)s/hosts/inactive')
[docs] def test_http_error(self):
self.resp.status_code = 404
self.resp.reason = 'Not Found'
try:
self.client.statistics.get_flow_statistics('default')
self.fail('')
except client.OpenDaylightRESTAPIFailed as e:
self.assertEqual(
_('OpenDaylitght API returned %(status)s %(reason)s') %
{'status': self.resp.status_code,
'reason': self.resp.reason},
six.text_type(e))
[docs] def test_other_error(self):
class _Exception(Exception):
pass
self.get = mock.patch('requests.get',
side_effect=_Exception).start()
self.assertRaises(_Exception,
self.client.statistics.get_flow_statistics,
'default')
[docs]class TestClientHTTPDigestAuth(TestClientHTTPBasicAuth):
auth_way = 'digest'
[docs]class TestClientHTTPSBasicAuth(TestClientHTTPBasicAuth):
scheme = 'https'
[docs]class TestClientHTTPSDigestAuth(TestClientHTTPDigestAuth):
scheme = 'https'