/usr/lib/python2.7/dist-packages/osprofiler/tests/functional/test_driver.py is in python-osprofiler 1.15.2-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | # Copyright (c) 2016 VMware, 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 logging
import os
from oslo_config import cfg
from osprofiler.drivers import base
from osprofiler import initializer
from osprofiler import opts
from osprofiler import profiler
from osprofiler.tests import test
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
@profiler.trace_cls("rpc", hide_args=True)
class Foo(object):
def bar(self, x):
return self.baz(x, x)
def baz(self, x, y):
return x * y
class DriverTestCase(test.FunctionalTestCase):
SERVICE = "service"
PROJECT = "project"
def setUp(self):
super(DriverTestCase, self).setUp()
CONF(["--config-file", os.path.dirname(__file__) + "/config.cfg"])
opts.set_defaults(CONF,
enabled=True,
trace_sqlalchemy=False,
hmac_keys="SECRET_KEY")
def _assert_dict(self, info, **kwargs):
for key in kwargs:
self.assertEqual(kwargs[key], info[key])
def _assert_child_dict(self, child, base_id, parent_id, name, fn_name):
self.assertEqual(parent_id, child["parent_id"])
exp_info = {"name": "rpc",
"service": self.SERVICE,
"project": self.PROJECT}
self._assert_dict(child["info"], **exp_info)
raw_start = child["info"]["meta.raw_payload.%s-start" % name]
self.assertEqual(fn_name, raw_start["info"]["function"]["name"])
exp_raw = {"name": "%s-start" % name,
"service": self.SERVICE,
"trace_id": child["trace_id"],
"project": self.PROJECT,
"base_id": base_id}
self._assert_dict(raw_start, **exp_raw)
raw_stop = child["info"]["meta.raw_payload.%s-stop" % name]
exp_raw["name"] = "%s-stop" % name
self._assert_dict(raw_stop, **exp_raw)
def test_get_report(self):
# initialize profiler notifier (the same way as in services)
initializer.init_from_conf(
CONF, {}, self.PROJECT, self.SERVICE, "host")
profiler.init("SECRET_KEY")
# grab base_id
base_id = profiler.get().get_base_id()
# execute profiled code
foo = Foo()
foo.bar(1)
# instantiate report engine (the same way as in osprofiler CLI)
engine = base.get_driver(CONF.profiler.connection_string,
project=self.PROJECT,
service=self.SERVICE,
host="host",
conf=CONF)
# generate the report
report = engine.get_report(base_id)
LOG.debug("OSProfiler report: %s", report)
# verify the report
self.assertEqual("total", report["info"]["name"])
self.assertEqual(2, report["stats"]["rpc"]["count"])
self.assertEqual(1, len(report["children"]))
cbar = report["children"][0]
self._assert_child_dict(
cbar, base_id, base_id, "rpc",
"osprofiler.tests.functional.test_driver.Foo.bar")
self.assertEqual(1, len(cbar["children"]))
cbaz = cbar["children"][0]
self._assert_child_dict(
cbaz, base_id, cbar["trace_id"], "rpc",
"osprofiler.tests.functional.test_driver.Foo.baz")
class RedisDriverTestCase(DriverTestCase):
def setUp(self):
super(DriverTestCase, self).setUp()
CONF([])
opts.set_defaults(CONF,
connection_string="redis://localhost:6379",
enabled=True,
trace_sqlalchemy=False,
hmac_keys="SECRET_KEY")
|