308 lines
11 KiB
Python
308 lines
11 KiB
Python
# Copyright 2014 Mirantis 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 mock
|
|
from webob import response as webob_response
|
|
|
|
from osprofiler import _utils as utils
|
|
from osprofiler import profiler
|
|
from osprofiler.tests import test
|
|
from osprofiler import web
|
|
|
|
|
|
def dummy_app(environ, response):
|
|
res = webob_response.Response()
|
|
return res(environ, response)
|
|
|
|
|
|
class WebTestCase(test.TestCase):
|
|
|
|
def setUp(self):
|
|
super(WebTestCase, self).setUp()
|
|
profiler.clean()
|
|
self.addCleanup(profiler.clean)
|
|
|
|
def test_get_trace_id_headers_no_hmac(self):
|
|
profiler.init(None, base_id="y", parent_id="z")
|
|
headers = web.get_trace_id_headers()
|
|
self.assertEqual(headers, {})
|
|
|
|
def test_get_trace_id_headers(self):
|
|
profiler.init("key", base_id="y", parent_id="z")
|
|
headers = web.get_trace_id_headers()
|
|
self.assertEqual(sorted(headers.keys()),
|
|
sorted(["X-Trace-Info", "X-Trace-HMAC"]))
|
|
|
|
trace_info = utils.signed_unpack(headers["X-Trace-Info"],
|
|
headers["X-Trace-HMAC"], ["key"])
|
|
self.assertIn("hmac_key", trace_info)
|
|
self.assertEqual("key", trace_info.pop("hmac_key"))
|
|
self.assertEqual({"parent_id": "z", "base_id": "y"}, trace_info)
|
|
|
|
@mock.patch("osprofiler.profiler.get")
|
|
def test_get_trace_id_headers_no_profiler(self, mock_get_profiler):
|
|
mock_get_profiler.return_value = False
|
|
headers = web.get_trace_id_headers()
|
|
self.assertEqual(headers, {})
|
|
|
|
|
|
class WebMiddlewareTestCase(test.TestCase):
|
|
def setUp(self):
|
|
super(WebMiddlewareTestCase, self).setUp()
|
|
profiler.clean()
|
|
# it's default state of _ENABLED param, so let's set it here
|
|
web._ENABLED = None
|
|
self.addCleanup(profiler.clean)
|
|
|
|
def tearDown(self):
|
|
web.enable()
|
|
super(WebMiddlewareTestCase, self).tearDown()
|
|
|
|
def test_factory(self):
|
|
mock_app = mock.MagicMock()
|
|
local_conf = {"enabled": True, "hmac_keys": "123"}
|
|
|
|
factory = web.WsgiMiddleware.factory(None, **local_conf)
|
|
wsgi = factory(mock_app)
|
|
|
|
self.assertEqual(wsgi.application, mock_app)
|
|
self.assertEqual(wsgi.name, "wsgi")
|
|
self.assertTrue(wsgi.enabled)
|
|
self.assertEqual(wsgi.hmac_keys, [local_conf["hmac_keys"]])
|
|
|
|
def _test_wsgi_middleware_with_invalid_trace(self, headers, hmac_key,
|
|
mock_profiler_init,
|
|
enabled=True):
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
request.headers = headers
|
|
|
|
middleware = web.WsgiMiddleware("app", hmac_key, enabled=enabled)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
request.get_response.assert_called_once_with("app")
|
|
self.assertEqual(0, mock_profiler_init.call_count)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_disabled(self, mock_profiler_init):
|
|
hmac_key = "secret"
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, hmac_key,
|
|
mock_profiler_init,
|
|
enabled=False)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_no_trace(self, mock_profiler_init):
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2"
|
|
}
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, "secret",
|
|
mock_profiler_init)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_invalid_trace_headers(self, mock_profiler_init):
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": "abbababababa",
|
|
"X-Trace-HMAC": "abbababababa"
|
|
}
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, "secret",
|
|
mock_profiler_init)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_no_trace_hmac(self, mock_profiler_init):
|
|
hmac_key = "secret"
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0]
|
|
}
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, hmac_key,
|
|
mock_profiler_init)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_invalid_hmac(self, mock_profiler_init):
|
|
hmac_key = "secret"
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": "not valid hmac"
|
|
}
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, hmac_key,
|
|
mock_profiler_init)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_invalid_trace_info(self, mock_profiler_init):
|
|
hmac_key = "secret"
|
|
pack = utils.signed_pack([{"base_id": "1"}, {"parent_id": "2"}],
|
|
hmac_key)
|
|
headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
self._test_wsgi_middleware_with_invalid_trace(headers, hmac_key,
|
|
mock_profiler_init)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_key_passthrough(self, mock_profiler_init):
|
|
hmac_key = "secret2"
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
request.url = "someurl"
|
|
request.host_url = "someurl"
|
|
request.path = "path"
|
|
request.query_string = "query"
|
|
request.method = "method"
|
|
request.scheme = "scheme"
|
|
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
|
|
request.headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
|
|
middleware = web.WsgiMiddleware("app", "secret1,%s" % hmac_key,
|
|
enabled=True)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
mock_profiler_init.assert_called_once_with(hmac_key=hmac_key,
|
|
base_id="1",
|
|
parent_id="2")
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_key_passthrough2(self, mock_profiler_init):
|
|
hmac_key = "secret1"
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
request.url = "someurl"
|
|
request.host_url = "someurl"
|
|
request.path = "path"
|
|
request.query_string = "query"
|
|
request.method = "method"
|
|
request.scheme = "scheme"
|
|
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
|
|
request.headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
|
|
middleware = web.WsgiMiddleware("app", "%s,secret2" % hmac_key,
|
|
enabled=True)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
mock_profiler_init.assert_called_once_with(hmac_key=hmac_key,
|
|
base_id="1",
|
|
parent_id="2")
|
|
|
|
@mock.patch("osprofiler.web.profiler.Trace")
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware(self, mock_profiler_init, mock_profiler_trace):
|
|
hmac_key = "secret"
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
request.url = "someurl"
|
|
request.host_url = "someurl"
|
|
request.path = "path"
|
|
request.query_string = "query"
|
|
request.method = "method"
|
|
request.scheme = "scheme"
|
|
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
|
|
request.headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
|
|
middleware = web.WsgiMiddleware("app", hmac_key, enabled=True)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
mock_profiler_init.assert_called_once_with(hmac_key=hmac_key,
|
|
base_id="1",
|
|
parent_id="2")
|
|
expected_info = {
|
|
"request": {
|
|
"path": request.path,
|
|
"query": request.query_string,
|
|
"method": request.method,
|
|
"scheme": request.scheme
|
|
}
|
|
}
|
|
mock_profiler_trace.assert_called_once_with("wsgi", info=expected_info)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_disable_via_python(self, mock_profiler_init):
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
web.disable()
|
|
middleware = web.WsgiMiddleware("app", "hmac_key", enabled=True)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
self.assertEqual(mock_profiler_init.call_count, 0)
|
|
|
|
@mock.patch("osprofiler.web.profiler.init")
|
|
def test_wsgi_middleware_enable_via_python(self, mock_profiler_init):
|
|
request = mock.MagicMock()
|
|
request.get_response.return_value = "yeah!"
|
|
request.url = "someurl"
|
|
request.host_url = "someurl"
|
|
request.path = "path"
|
|
request.query_string = "query"
|
|
request.method = "method"
|
|
request.scheme = "scheme"
|
|
hmac_key = "super_secret_key2"
|
|
|
|
pack = utils.signed_pack({"base_id": "1", "parent_id": "2"}, hmac_key)
|
|
request.headers = {
|
|
"a": "1",
|
|
"b": "2",
|
|
"X-Trace-Info": pack[0],
|
|
"X-Trace-HMAC": pack[1]
|
|
}
|
|
|
|
web.enable("super_secret_key1,super_secret_key2")
|
|
middleware = web.WsgiMiddleware("app", enabled=True)
|
|
self.assertEqual("yeah!", middleware(request))
|
|
mock_profiler_init.assert_called_once_with(hmac_key=hmac_key,
|
|
base_id="1",
|
|
parent_id="2")
|
|
|
|
def test_disable(self):
|
|
web.disable()
|
|
self.assertFalse(web._ENABLED)
|
|
|
|
def test_enabled(self):
|
|
web.disable()
|
|
web.enable()
|
|
self.assertTrue(web._ENABLED)
|