Loading src/context/tests/test_unitary.py +29 −26 Original line number Diff line number Diff line import logging, pytest, sys #import logging, pytest, sys from pathlib import Path sys.path.append(__file__.split('src')[0] + 'src') print(sys.path) #from pathlib import Path #sys.path.append(__file__.split('src')[0] + 'src') #print(sys.path) from common.database.Factory import get_database, DatabaseEngineEnum from context.client.ContextClient import ContextClient from context.proto.context_pb2 import Empty from context.service.ContextService import ContextService from context.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD from context.tests.tools.ValidateTopology import validate_topology_dict #from common.database.Factory import get_database, DatabaseEngineEnum #from context.client.ContextClient import ContextClient #from context.proto.context_pb2 import Empty #from context.service.ContextService import ContextService #from context.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD #from context.tests.tools.ValidateTopology import validate_topology_dict LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) #LOGGER = logging.getLogger(__name__) #LOGGER.setLevel(logging.DEBUG) @pytest.fixture(scope='session') def local_context_service(): database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json') _service = ContextService( database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD) _service.start() yield _service _service.stop() #@pytest.fixture(scope='session') #def local_context_service(): # database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json') # _service = ContextService( # database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD) # _service.start() # yield _service # _service.stop() @pytest.fixture(scope='session') def local_context_client(local_context_service): return ContextClient(address='127.0.0.1', port=GRPC_SERVICE_PORT) #@pytest.fixture(scope='session') #def local_context_client(local_context_service): # return ContextClient(address='127.0.0.1', port=GRPC_SERVICE_PORT) def test_local_get_topology(local_context_client): response = local_context_client.GetTopology(Empty()) validate_topology_dict(response) #def test_local_get_topology(local_context_client): # response = local_context_client.GetTopology(Empty()) # validate_topology_dict(response) def test_nothing(): pass Loading
src/context/tests/test_unitary.py +29 −26 Original line number Diff line number Diff line import logging, pytest, sys #import logging, pytest, sys from pathlib import Path sys.path.append(__file__.split('src')[0] + 'src') print(sys.path) #from pathlib import Path #sys.path.append(__file__.split('src')[0] + 'src') #print(sys.path) from common.database.Factory import get_database, DatabaseEngineEnum from context.client.ContextClient import ContextClient from context.proto.context_pb2 import Empty from context.service.ContextService import ContextService from context.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD from context.tests.tools.ValidateTopology import validate_topology_dict #from common.database.Factory import get_database, DatabaseEngineEnum #from context.client.ContextClient import ContextClient #from context.proto.context_pb2 import Empty #from context.service.ContextService import ContextService #from context.Config import GRPC_SERVICE_PORT, GRPC_MAX_WORKERS, GRPC_GRACE_PERIOD #from context.tests.tools.ValidateTopology import validate_topology_dict LOGGER = logging.getLogger(__name__) LOGGER.setLevel(logging.DEBUG) #LOGGER = logging.getLogger(__name__) #LOGGER.setLevel(logging.DEBUG) @pytest.fixture(scope='session') def local_context_service(): database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json') _service = ContextService( database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD) _service.start() yield _service _service.stop() #@pytest.fixture(scope='session') #def local_context_service(): # database = get_database(engine=DatabaseEngineEnum.INMEMORY, filepath='data/topo_nsfnet.json') # _service = ContextService( # database, port=GRPC_SERVICE_PORT, max_workers=GRPC_MAX_WORKERS, grace_period=GRPC_GRACE_PERIOD) # _service.start() # yield _service # _service.stop() @pytest.fixture(scope='session') def local_context_client(local_context_service): return ContextClient(address='127.0.0.1', port=GRPC_SERVICE_PORT) #@pytest.fixture(scope='session') #def local_context_client(local_context_service): # return ContextClient(address='127.0.0.1', port=GRPC_SERVICE_PORT) def test_local_get_topology(local_context_client): response = local_context_client.GetTopology(Empty()) validate_topology_dict(response) #def test_local_get_topology(local_context_client): # response = local_context_client.GetTopology(Empty()) # validate_topology_dict(response) def test_nothing(): pass