# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/) # # 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 grpc, logging, time from common.rpc_method_wrapper.Decorator 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()) # pylint: disable=protected-access