scripts/run_tests_locally_kpi_manager.sh

@@ -17,8 +17,12 @@
 PROJECTDIR=`pwd`
 
 cd $PROJECTDIR/src
-RCFILE=$PROJECTDIR/coverage/.coveragerc
-coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
-    kpi_manager/tests/test_unitary.py
+# RCFILE=$PROJECTDIR/coverage/.coveragerc
+# coverage run --rcfile=$RCFILE --append -m pytest --log-level=INFO --verbose \
+#     kpi_manager/tests/test_unitary.py
+
+# python3 kpi_manager/tests/test_unitary.py
 
-# python3 kpi_manager/tests/test_unitary.py
\ No newline at end of file
+RCFILE=$PROJECTDIR/coverage/.coveragerc
+python3 -m pytest --log-level=INFO --verbose \
+    kpi_manager/tests/test_unitary.py
\ No newline at end of file

src/kpi_manager/client/KpiManagerClient.py

@@ -19,7 +19,7 @@ from common.Settings import get_service_host, get_service_port_grpc
 from common.tools.client.RetryDecorator import retry, delay_exponential
 from common.tools.grpc.Tools import grpc_message_to_json_string
 from common.proto.context_pb2 import Empty
-from common.proto.kpi_manager_pb2 import KpiId, KpiDescriptor, KpiDescriptorList,
+from common.proto.kpi_manager_pb2 import KpiId, KpiDescriptor, KpiDescriptorList
 from common.proto.kpi_manager_pb2_grpc import KpiManagerServiceStub
 
 LOGGER = logging.getLogger(__name__)

src/kpi_manager/client/__init__.py

+# Copyright 2022-2023 ETSI TeraFlowSDN - TFS OSG (https://tfs.etsi.org/)
+#
+# 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.
+

src/kpi_manager/tests/test_unitary.py

@@ -17,6 +17,7 @@
 # sys.path.append('.')
 import os, pytest
 import logging, json
+from typing import Union
 
 from apscheduler.schedulers.background import BackgroundScheduler
 
@@ -33,6 +34,8 @@ from common.tools.object_factory.Topology import json_topology, json_topology_id
 # from common.proto.monitoring_pb2 import KpiId, KpiDescriptor, SubsDescriptor, SubsList, AlarmID, \
 #     AlarmDescriptor, AlarmList, KpiDescriptorList, SubsResponse, AlarmResponse, RawKpiTable #, Kpi, KpiList
 from common.proto.kpi_manager_pb2 import KpiId, KpiDescriptor, KpiDescriptorList
+from common.tools.service.GenericGrpcService import GenericGrpcService
+from context.client.ContextClient import ContextClient
 
 
 from device.service.driver_api.DriverFactory import DriverFactory
@@ -50,6 +53,9 @@ from monitoring.service.ManagementDBTools import ManagementDB
 from monitoring.service.MetricsDBTools import MetricsDB
 from monitoring.service.NameMapping import NameMapping
 
+os.environ['DEVICE_EMULATED_ONLY'] = 'TRUE'
+from device.service.drivers import DRIVERS
+
 ###########################
 # Tests Setup
 ###########################
@@ -185,49 +191,49 @@ 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)
+# @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