import copy, grpc, logging, os, pytest from common.tests.MockService import MockService from common.tests.MockServicerImpl_Context import MockServicerImpl_Context from common.tests.MockServicerImpl_Device import MockServicerImpl_Device from common.tests.PytestGenerateTests import pytest_generate_tests # (required) pylint: disable=unused-import from common.tools.grpc.Tools import grpc_message_to_json_string from context.client.ContextClient import ContextClient from context.proto.context_pb2 import Context, ContextId, DeviceId, Link, LinkId, Topology, Device, TopologyId from context.proto.context_pb2_grpc import add_ContextServiceServicer_to_server from device.client.DeviceClient import DeviceClient from device.proto.device_pb2_grpc import add_DeviceServiceServicer_to_server from service.Config import ( GRPC_SERVICE_PORT as SERVICE_GRPC_SERVICE_PORT, GRPC_MAX_WORKERS as SERVICE_GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD as SERVICE_GRPC_GRACE_PERIOD) from service.client.ServiceClient import ServiceClient from service.proto.context_pb2 import Service, ServiceId from service.service.ServiceService import ServiceService from service.service.service_handler_api.ServiceHandlerFactory import ServiceHandlerFactory from service.service.service_handlers import SERVICE_HANDLERS LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) LOCALHOST = '127.0.0.1' MOCKSERVER_GRPC_PORT = 10000 class MockService_Combined(MockService): # Mock Server implementing Context and Service to simplify unitary tests of Compute def __init__(self, cls_name='MockService_Service'): super().__init__(LOCALHOST, MOCKSERVER_GRPC_PORT, cls_name=cls_name) # pylint: disable=attribute-defined-outside-init def install_servicers(self): self.context_servicer = MockServicerImpl_Context() add_ContextServiceServicer_to_server(self.context_servicer, self.server) self.device_servicer = MockServicerImpl_Device() add_DeviceServiceServicer_to_server(self.device_servicer, self.server) os.environ['CONTEXTSERVICE_SERVICE_HOST'] = LOCALHOST os.environ['CONTEXTSERVICE_SERVICE_PORT_GRPC'] = str(MOCKSERVER_GRPC_PORT) os.environ['DEVICESERVICE_SERVICE_HOST'] = LOCALHOST os.environ['DEVICESERVICE_SERVICE_PORT_GRPC'] = str(MOCKSERVER_GRPC_PORT) @pytest.fixture(scope='session') def mockservice(): _service = MockService_Combined() _service.start() yield _service _service.stop() @pytest.fixture(scope='session') def context_client(mockservice : MockService_Combined): # pylint: disable=redefined-outer-name _client = ContextClient(address=LOCALHOST, port=MOCKSERVER_GRPC_PORT) yield _client _client.close() @pytest.fixture(scope='session') def device_client(mockservice : MockService_Combined): # pylint: disable=redefined-outer-name _client = DeviceClient(address=LOCALHOST, port=MOCKSERVER_GRPC_PORT) yield _client _client.close() @pytest.fixture(scope='session') def service_service( context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient): # pylint: disable=redefined-outer-name _service_handler_factory = ServiceHandlerFactory(SERVICE_HANDLERS) _service = ServiceService( context_client, device_client, _service_handler_factory, port=SERVICE_GRPC_SERVICE_PORT, max_workers=SERVICE_GRPC_MAX_WORKERS, grace_period=SERVICE_GRPC_GRACE_PERIOD) _service.start() yield _service _service.stop() @pytest.fixture(scope='session') def service_client(service_service : ServiceService): # pylint: disable=redefined-outer-name _client = ServiceClient(address=LOCALHOST, port=SERVICE_GRPC_SERVICE_PORT) yield _client _client.close() try: from .ServiceHandlersToTest import SERVICE_HANDLERS_TO_TEST except ImportError: LOGGER.exception('Unable to load service handlers, nothing will be tested.') SERVICE_HANDLERS_TO_TEST = [] class TestServiceHandlers: scenarios = SERVICE_HANDLERS_TO_TEST def test_prepare_environment( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient): # pylint: disable=redefined-outer-name for context in contexts: context_client.SetContext(Context(**context)) for topology in topologies: context_client.SetTopology(Topology(**topology)) for device in devices: device_client.AddDevice(Device(**device)) for link in links: context_client.SetLink(Link(**link)) def test_service_create_error_cases( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, service_client : ServiceClient): # pylint: disable=redefined-outer-name with pytest.raises(grpc.RpcError) as e: service_with_endpoints = copy.deepcopy(service_descriptor) service_with_endpoints['service_endpoint_ids'].extend(service_endpoint_ids) service_client.CreateService(Service(**service_with_endpoints)) assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT msg_head = 'service.service_endpoint_ids([' msg_tail = ']) is invalid; RPC method CreateService does not accept Endpoints. '\ 'Endpoints should be configured after creating the service.' except_msg = str(e.value.details()) assert except_msg.startswith(msg_head) and except_msg.endswith(msg_tail) with pytest.raises(grpc.RpcError) as e: service_with_config_rules = copy.deepcopy(service_descriptor) service_with_config_rules['service_config']['config_rules'].extend(service_config_rules) service_client.CreateService(Service(**service_with_config_rules)) assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT msg_head = 'service.service_config.config_rules([' msg_tail = ']) is invalid; RPC method CreateService does not accept Config Rules. '\ 'Config Rules should be configured after creating the service.' except_msg = str(e.value.details()) assert except_msg.startswith(msg_head) and except_msg.endswith(msg_tail) with pytest.raises(grpc.RpcError) as e: service_with_constraints = copy.deepcopy(service_descriptor) service_with_constraints['service_constraints'].extend(service_constraints) service_client.CreateService(Service(**service_with_constraints)) assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT msg_head = 'service.service_constraints([' msg_tail = ']) is invalid; RPC method CreateService does not accept Constraints. '\ 'Constraints should be configured after creating the service.' except_msg = str(e.value.details()) assert except_msg.startswith(msg_head) and except_msg.endswith(msg_tail) def test_service_create_correct( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, service_client : ServiceClient): # pylint: disable=redefined-outer-name service_client.CreateService(Service(**service_descriptor)) def test_service_get_created( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient): # pylint: disable=redefined-outer-name service_data = context_client.GetService(ServiceId(**service_id)) LOGGER.info('service_data = {:s}'.format(grpc_message_to_json_string(service_data))) def test_service_update_configure( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient, # pylint: disable=redefined-outer-name service_client : ServiceClient): # pylint: disable=redefined-outer-name service_with_settings = copy.deepcopy(service_descriptor) service_with_settings['service_endpoint_ids'].extend(service_endpoint_ids) service_with_settings['service_config']['config_rules'].extend(service_config_rules) service_with_settings['service_constraints'].extend(service_constraints) service_client.UpdateService(Service(**service_with_settings)) for endpoint_id in service_endpoint_ids: device_id = endpoint_id['device_id'] device_data = context_client.GetDevice(DeviceId(**device_id)) for i,config_rule in enumerate(device_data.device_config.config_rules): LOGGER.info('device_data[{:s}][#{:d}] => {:s}'.format( str(device_id), i, grpc_message_to_json_string(config_rule))) def test_service_update_deconfigure( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient, # pylint: disable=redefined-outer-name service_client : ServiceClient): # pylint: disable=redefined-outer-name service_with_settings = copy.deepcopy(service_descriptor) service_with_settings['service_endpoint_ids'].extend([]) # remove endpoints service_client.UpdateService(Service(**service_with_settings)) for endpoint_id in service_endpoint_ids: device_id = endpoint_id['device_id'] device_data = context_client.GetDevice(DeviceId(**device_id)) for i,config_rule in enumerate(device_data.device_config.config_rules): LOGGER.info('device_data[{:s}][#{:d}] => {:s}'.format( str(device_id), i, grpc_message_to_json_string(config_rule))) def test_service_get_updated( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient): # pylint: disable=redefined-outer-name service_data = context_client.GetService(ServiceId(**service_id)) LOGGER.info('service_data = {:s}'.format(grpc_message_to_json_string(service_data))) def test_service_delete( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, service_client : ServiceClient): # pylint: disable=redefined-outer-name service_client.DeleteService(ServiceId(**service_id)) def test_cleanup_environment( self, service_id, service_descriptor, service_endpoint_ids, service_config_rules, service_constraints, contexts, topologies, devices, links, context_client : ContextClient, # pylint: disable=redefined-outer-name device_client : DeviceClient): # pylint: disable=redefined-outer-name for link in links: context_client.RemoveLink(LinkId(**link['link_id'])) for device in devices: device_client.DeleteDevice(DeviceId(**device['device_id'])) for topology in topologies: context_client.RemoveTopology(TopologyId(**topology['topology_id'])) for context in contexts: context_client.RemoveContext(ContextId(**context['context_id']))