import grpc, logging, pytest, time from common.metrics.Metrics import create_metrics, safe_and_metered_rpc_method logging.basicConfig(level=logging.DEBUG) LOGGER = logging.getLogger(__name__) def test_database_instantiation(): SERVICE_NAME = 'Context' METHOD_NAMES = [ 'ListContextIds', 'ListContexts', 'GetContext', 'SetContext', 'RemoveContext', 'GetContextEvents', 'ListTopologyIds', 'ListTopologies', 'GetTopology', 'SetTopology', 'RemoveTopology', 'GetTopologyEvents', 'ListDeviceIds', 'ListDevices', 'GetDevice', 'SetDevice', 'RemoveDevice', 'GetDeviceEvents', 'ListLinkIds', 'ListLinks', 'GetLink', 'SetLink', 'RemoveLink', 'GetLinkEvents', 'ListServiceIds', 'ListServices', 'GetService', 'SetService', 'RemoveService', 'GetServiceEvents', ] METRICS = create_metrics(SERVICE_NAME, METHOD_NAMES) class TestServiceServicerImpl: @safe_and_metered_rpc_method(METRICS, LOGGER) def GetTopology(self, request, grpc_context : grpc.ServicerContext): print('doing funny things') time.sleep(0.1) return 'done' tssi = TestServiceServicerImpl() tssi.GetTopology(1, 2) for metric_name,metric in METRICS.items(): if 'GETTOPOLOGY_' not in metric_name: continue print(metric_name, metric._child_samples())