Loading src/kpi_manager/tests/test_unitary.py +1 −45 Original line number Diff line number Diff line Loading @@ -191,50 +191,6 @@ def kpi_manager_client(kpi_manager_service : KpiManagerService): # pylint: disab LOGGER.info('Closed KpiManagerClient...') # @pytest.fixture(scope='session') # def management_db(): # _management_db = ManagementDB('monitoring.db') # return _management_db # @pytest.fixture(scope='session') # def metrics_db(kpi_manager_service : KpiManagerService): # pylint: disable=redefined-outer-name # return monitoring_service.monitoring_servicer.metrics_db # # This function os not clear to me (Changes should me made before execution) # @pytest.fixture(scope='session') # def metrics_db(monitoring_service : MonitoringService): # pylint: disable=redefined-outer-name # return monitoring_service.monitoring_servicer.metrics_db # #_metrics_db = MetricsDBTools.MetricsDB( # # METRICSDB_HOSTNAME, METRICSDB_ILP_PORT, METRICSDB_REST_PORT, METRICSDB_TABLE_MONITORING_KPIS) # #return _metrics_db # @pytest.fixture(scope='session') # def subs_scheduler(): # _scheduler = BackgroundScheduler(executors={'processpool': ProcessPoolExecutor(max_workers=20)}) # _scheduler.start() # return _scheduler # def ingestion_data(kpi_id_int): # # pylint: disable=redefined-outer-name,unused-argument # metrics_db = MetricsDB('localhost', '9009', '9000', 'monitoring') # kpiSampleType = KpiSampleType.KPISAMPLETYPE_PACKETS_RECEIVED # kpiSampleType_name = KpiSampleType.Name(kpiSampleType).upper().replace('KPISAMPLETYPE_', '') # for _ in range(50): # kpiSampleType = kpiSampleType_name # kpiId = kpi_id_int # deviceId = 'DEV'+ str(kpi_id_int) # endpointId = 'END' + str(kpi_id_int) # serviceId = 'SERV' + str(kpi_id_int) # sliceId = 'SLC' + str(kpi_id_int) # connectionId = 'CON' + str(kpi_id_int) # time_stamp = timestamp_utcnow_to_float() # kpi_value = 500*random() # metrics_db.write_KPI(time_stamp, kpiId, kpiSampleType, deviceId, endpointId, serviceId, sliceId, connectionId, # kpi_value) # sleep(0.1) ################################################## # Prepare Environment, should be the first test ################################################## Loading @@ -248,7 +204,7 @@ def test_prepare_environment( context_client.SetTopology(Topology(**json_topology(DEFAULT_TOPOLOGY_NAME, context_id=context_id))) ########################### # Tests Implementation # Tests Implementation of Kpi Manager ########################### # Test case that makes use of client fixture to test server's CreateKpi method Loading Loading
src/kpi_manager/tests/test_unitary.py +1 −45 Original line number Diff line number Diff line Loading @@ -191,50 +191,6 @@ def kpi_manager_client(kpi_manager_service : KpiManagerService): # pylint: disab LOGGER.info('Closed KpiManagerClient...') # @pytest.fixture(scope='session') # def management_db(): # _management_db = ManagementDB('monitoring.db') # return _management_db # @pytest.fixture(scope='session') # def metrics_db(kpi_manager_service : KpiManagerService): # pylint: disable=redefined-outer-name # return monitoring_service.monitoring_servicer.metrics_db # # This function os not clear to me (Changes should me made before execution) # @pytest.fixture(scope='session') # def metrics_db(monitoring_service : MonitoringService): # pylint: disable=redefined-outer-name # return monitoring_service.monitoring_servicer.metrics_db # #_metrics_db = MetricsDBTools.MetricsDB( # # METRICSDB_HOSTNAME, METRICSDB_ILP_PORT, METRICSDB_REST_PORT, METRICSDB_TABLE_MONITORING_KPIS) # #return _metrics_db # @pytest.fixture(scope='session') # def subs_scheduler(): # _scheduler = BackgroundScheduler(executors={'processpool': ProcessPoolExecutor(max_workers=20)}) # _scheduler.start() # return _scheduler # def ingestion_data(kpi_id_int): # # pylint: disable=redefined-outer-name,unused-argument # metrics_db = MetricsDB('localhost', '9009', '9000', 'monitoring') # kpiSampleType = KpiSampleType.KPISAMPLETYPE_PACKETS_RECEIVED # kpiSampleType_name = KpiSampleType.Name(kpiSampleType).upper().replace('KPISAMPLETYPE_', '') # for _ in range(50): # kpiSampleType = kpiSampleType_name # kpiId = kpi_id_int # deviceId = 'DEV'+ str(kpi_id_int) # endpointId = 'END' + str(kpi_id_int) # serviceId = 'SERV' + str(kpi_id_int) # sliceId = 'SLC' + str(kpi_id_int) # connectionId = 'CON' + str(kpi_id_int) # time_stamp = timestamp_utcnow_to_float() # kpi_value = 500*random() # metrics_db.write_KPI(time_stamp, kpiId, kpiSampleType, deviceId, endpointId, serviceId, sliceId, connectionId, # kpi_value) # sleep(0.1) ################################################## # Prepare Environment, should be the first test ################################################## Loading @@ -248,7 +204,7 @@ def test_prepare_environment( context_client.SetTopology(Topology(**json_topology(DEFAULT_TOPOLOGY_NAME, context_id=context_id))) ########################### # Tests Implementation # Tests Implementation of Kpi Manager ########################### # Test case that makes use of client fixture to test server's CreateKpi method Loading