143 lines
4.8 KiB
Python
143 lines
4.8 KiB
Python
# Copyright (c) 2016 Cisco Systems
|
|
#
|
|
# 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.
|
|
|
|
from unittest import mock
|
|
import uuid
|
|
|
|
from oslotest import base as test_base
|
|
import statsd
|
|
import webob.dec
|
|
import webob.exc
|
|
|
|
from oslo_middleware import stats
|
|
|
|
|
|
class TestStaticMethods(test_base.BaseTestCase):
|
|
|
|
def test_removes_uuid(self):
|
|
# Generate a long-format UUID (standard form).
|
|
id = str(uuid.uuid4())
|
|
path = "foo.{uuid}.bar".format(uuid=id)
|
|
stat = stats.StatsMiddleware.strip_uuid(path)
|
|
self.assertEqual("foo.bar", stat)
|
|
|
|
def test_removes_short_uuid(self):
|
|
id = uuid.uuid4().hex
|
|
path = "foo.{uuid}.bar".format(uuid=id)
|
|
stat = stats.StatsMiddleware.strip_short_uuid(path)
|
|
self.assertEqual("foo.bar", stat)
|
|
|
|
def test_strips_dots_from_version(self):
|
|
path = "/v1.2/foo.bar/bar.foo"
|
|
stat = stats.StatsMiddleware.strip_dot_from_version(path)
|
|
self.assertEqual("/v12/foo.bar/bar.foo", stat)
|
|
|
|
|
|
class TestStatsMiddleware(test_base.BaseTestCase):
|
|
|
|
def setUp(self):
|
|
super(TestStatsMiddleware, self).setUp()
|
|
self.patch(statsd, 'StatsClient', mock.MagicMock())
|
|
|
|
def make_stats_middleware(self, stat_name=None, stats_host=None,
|
|
remove_uuid=False, remove_short_uuid=False):
|
|
if stat_name is None:
|
|
stat_name = uuid.uuid4().hex
|
|
if stats_host is None:
|
|
stats_host = uuid.uuid4().hex
|
|
|
|
conf = dict(
|
|
name=stat_name,
|
|
stats_host=stats_host,
|
|
remove_uuid=remove_uuid,
|
|
remove_short_uuid=remove_short_uuid,
|
|
)
|
|
|
|
@webob.dec.wsgify
|
|
def fake_application(req):
|
|
return 'Hello, World'
|
|
|
|
return stats.StatsMiddleware(fake_application, conf)
|
|
|
|
def perform_request(self, app, path, method):
|
|
req = webob.Request.blank(path, method=method)
|
|
return req.get_response(app)
|
|
|
|
def test_sends_counter_to_statsd(self):
|
|
app = self.make_stats_middleware()
|
|
path = '/test/foo/bar'
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}.{path}".format(
|
|
name=app.stat_name, method='GET',
|
|
path=path.lstrip('/').replace('/', '.'))
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|
|
|
|
def test_strips_uuid_if_configured(self):
|
|
app = self.make_stats_middleware(remove_uuid=True)
|
|
random_uuid = str(uuid.uuid4())
|
|
path = '/foo/{uuid}/bar'.format(uuid=random_uuid)
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}.foo.bar".format(
|
|
name=app.stat_name, method='GET')
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|
|
|
|
def test_strips_short_uuid_if_configured(self):
|
|
app = self.make_stats_middleware(remove_short_uuid=True)
|
|
random_uuid = uuid.uuid4().hex
|
|
path = '/foo/{uuid}/bar'.format(uuid=random_uuid)
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}.foo.bar".format(
|
|
name=app.stat_name, method='GET')
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|
|
|
|
def test_strips_both_uuid_types_if_configured(self):
|
|
app = self.make_stats_middleware(
|
|
remove_uuid=True, remove_short_uuid=True)
|
|
random_short_uuid = uuid.uuid4().hex
|
|
random_uuid = str(uuid.uuid4())
|
|
path = '/foo/{uuid}/bar/{short_uuid}'.format(
|
|
uuid=random_uuid, short_uuid=random_short_uuid)
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}.foo.bar".format(
|
|
name=app.stat_name, method='GET')
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|
|
|
|
def test_always_mutates_version_id(self):
|
|
app = self.make_stats_middleware()
|
|
path = '/v2.1/foo/bar'
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}.v21.foo.bar".format(
|
|
name=app.stat_name, method='GET')
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|
|
|
|
def test_empty_path_has_sane_stat_name(self):
|
|
app = self.make_stats_middleware()
|
|
path = '/'
|
|
|
|
self.perform_request(app, path, 'GET')
|
|
|
|
expected_stat = "{name}.{method}".format(
|
|
name=app.stat_name, method='GET')
|
|
app.statsd.timer.assert_called_once_with(expected_stat)
|