Changes to refine code for monitoring E2E test.

scripts/run_tests_locally-telemetry-backend.sh

@@ -22,6 +22,7 @@ cd $PROJECTDIR/src
 #     kpi_manager/tests/test_unitary.py
 
 # python3 kpi_manager/tests/test_unitary.py
+export KFK_SERVER_ADDRESS='127.0.0.1:9092'
 CRDB_SQL_ADDRESS=$(kubectl get service cockroachdb-public --namespace crdb -o jsonpath='{.spec.clusterIP}')
 export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_kpi_mgmt?sslmode=require"
 RCFILE=$PROJECTDIR/coverage/.coveragerc

scripts/run_tests_locally-telemetry-frontend.sh

@@ -19,6 +19,7 @@ PROJECTDIR=`pwd`
 cd $PROJECTDIR/src
 
 # python3 kpi_manager/tests/test_unitary.py
+export KFK_SERVER_ADDRESS='127.0.0.1:9092'
 CRDB_SQL_ADDRESS=$(kubectl get service cockroachdb-public --namespace crdb -o jsonpath='{.spec.clusterIP}')
 export CRDB_URI="cockroachdb://tfs:tfs123@${CRDB_SQL_ADDRESS}:26257/tfs_kpi_mgmt?sslmode=require"
 RCFILE=$PROJECTDIR/coverage/.coveragerc

src/analytics/backend/tests/test_backend.py

@@ -119,20 +119,20 @@ LOGGER = logging.getLogger(__name__)
 #     assert isinstance(response, bool)
 
 # --- To TEST StartRequestListenerFunctionality
-def test_StartRequestListener():
-    LOGGER.info('test_RunRequestListener')
-    AnalyticsBackendServiceObj = AnalyticsBackendService()
-    AnalyticsBackendServiceObj.stop_event = Event()
-    listener_thread = Thread(target=AnalyticsBackendServiceObj.RequestListener, args=())
-    listener_thread.start()
+# def test_StartRequestListener():
+#     LOGGER.info('test_RunRequestListener')
+#     AnalyticsBackendServiceObj = AnalyticsBackendService()
+#     AnalyticsBackendServiceObj.stop_event = Event()
+#     listener_thread = Thread(target=AnalyticsBackendServiceObj.RequestListener, args=())
+#     listener_thread.start()
 
-    time.sleep(100)
+#     time.sleep(100)
 
     # AnalyticsBackendServiceObj.stop_event.set()
     # LOGGER.info('Backend termination initiated. waiting for termination... 10 seconds')
     # listener_thread.join(timeout=10)
     # assert not listener_thread.is_alive(), "RequestListener thread did not terminate as expected."
-    LOGGER.info('Completed test_RunRequestListener')
+    # LOGGER.info('Completed test_RunRequestListener')
 
 # To test START and STOP communication together
 # def test_StopRequestListener():

src/analytics/frontend/service/AnalyticsFrontendServiceServicerImpl.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging, grpc, json
+import logging, grpc, json, queue
 
 from typing          import Dict
 from confluent_kafka import Producer as KafkaProducer
@@ -24,6 +24,8 @@ from common.proto.analytics_frontend_pb2      import Analyzer, AnalyzerId, Analy
 from common.proto.analytics_frontend_pb2_grpc import AnalyticsFrontendServiceServicer
 from analytics.database.Analyzer_DB           import AnalyzerDB
 from analytics.database.AnalyzerModel         import Analyzer as AnalyzerModel
+from apscheduler.schedulers.background        import BackgroundScheduler
+from apscheduler.triggers.interval            import IntervalTrigger
 
 
 LOGGER           = logging.getLogger(__name__)

src/analytics/frontend/tests/test_frontend.py

@@ -84,10 +84,10 @@ def analyticsFrontend_client(analyticsFrontend_service : AnalyticsFrontendServic
 ###########################
 
 # --- "test_validate_kafka_topics" should be executed before the functionality tests ---
-# def test_validate_kafka_topics():
-#     LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
-#     response = KafkaTopic.create_all_topics()
-#     assert isinstance(response, bool)
+def test_validate_kafka_topics():
+    LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+    response = KafkaTopic.create_all_topics()
+    assert isinstance(response, bool)
 
 # ----- core funtionality test -----
 # def test_StartAnalytics(analyticsFrontend_client):

src/analytics/tests/test_analytics_db.py

@@ -15,12 +15,13 @@
 
 import logging
 from analytics.database.Analyzer_DB import AnalyzerDB
+from analytics.database.AnalyzerModel import Analyzer
 
 LOGGER = logging.getLogger(__name__)
 
 def test_verify_databases_and_tables():
     LOGGER.info('>>> test_verify_databases_and_tables : START <<< ')
-    AnalyzerDBobj = AnalyzerDB()
+    AnalyzerDBobj = AnalyzerDB(Analyzer)
     # AnalyzerDBobj.drop_database()
     # AnalyzerDBobj.verify_tables()
     AnalyzerDBobj.create_database()

src/telemetry/backend/tests/test_backend.py

@@ -26,10 +26,10 @@ LOGGER = logging.getLogger(__name__)
 ###########################
 
 # --- "test_validate_kafka_topics" should be run before the functionality tests ---
-# def test_validate_kafka_topics():
-#     LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
-#     response = KafkaTopic.create_all_topics()
-#     assert isinstance(response, bool)
+def test_validate_kafka_topics():
+    LOGGER.debug(" >>> test_validate_kafka_topics: START <<< ")
+    response = KafkaTopic.create_all_topics()
+    assert isinstance(response, bool)
 
 def test_RunRequestListener():
     LOGGER.info('test_RunRequestListener')

src/telemetry/tests/test_telemetryDB.py

@@ -15,12 +15,13 @@
 
 import logging
 from telemetry.database.Telemetry_DB import TelemetryDB
+from telemetry.database.TelemetryModel import Collector as CollectorModel
 
 LOGGER = logging.getLogger(__name__)
 
 def test_verify_databases_and_tables():
     LOGGER.info('>>> test_verify_databases_and_tables : START <<< ')
-    TelemetryDBobj = TelemetryDB()
+    TelemetryDBobj = TelemetryDB(CollectorModel)
     # TelemetryDBobj.drop_database()
     # TelemetryDBobj.verify_tables()
     TelemetryDBobj.create_database()