Add kafka support in Policy CSIT

Issue-ID: POLICY-4402
Signed-off-by: zrrmmua <ramesh.murugan.iyer@est.tech>
Change-Id: I802c19a3c9817d304164eba634adb8c119aa4ced

8 files changed, 12 insertions, 164 deletions

diff --git a/csit/resources/scripts/kafka_consumer.py b/csit/resources/scripts/kafka_consumer.py
deleted file mode 100755
index 80b6167a..00000000
--- a/csit/resources/scripts/kafka_consumer.py
+++ /dev/null
@@ -1,71 +0,0 @@
-#!/usr/bin/env python3
-#
-# ============LICENSE_START====================================================
-#  Copyright (C) 2023 Nordix Foundation.
-# =============================================================================
-# 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.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END======================================================
-
-# Python utility to fetch kafka topic and look for required messages.
-# Accepts the arguments {topic_name} and {list of expected values} and {timeout} to verify the kafka topic.
-
-
-from confluent_kafka import Consumer, KafkaException
-import sys
-import time
-
-def consume_kafka_topic(topic, expected_values, timeout):
-    config = {
-            'bootstrap.servers': 'localhost:29092',
-            'group.id': 'testgrp',
-            'auto.offset.reset': 'earliest'
-    }
-    consumer = Consumer(config)
-    consumer.subscribe([topic])
-    try:
-        start_time = time.time()
-        while time.time() - start_time < timeout:
-                msg = consumer.poll(1.0)
-                if msg is None:
-                    continue
-                if msg.error():
-                    if msg.error().code() == KafkaException._PARTITION_EOF:
-                        sys.stderr.write(f"Reached end of topic {msg.topic()} / partition {msg.partition()}\n")
-                        print('ERROR')
-                        sys.exit(404)
-                    else:
-                        # Error
-                        raise KafkaException(msg.error())
-                else:
-                    # Message received
-                    message = msg.value().decode('utf-8')
-                    if verify_msg(expected_values, message):
-                        print(message)
-                        sys.exit(200)
-    finally:
-        consumer.close()
-
-def verify_msg(expected_values, message):
-    for item in expected_values:
-        if item not in message:
-            return False
-    return True
-
-
-if __name__ == '__main__':
-    topic_name = sys.argv[1]
-    timeout = sys.argv[2]  # timeout in seconds for verifying the kafka topic
-    expected_values = sys.argv[3:]
-    consume_kafka_topic(topic_name, expected_values, timeout)
diff --git a/csit/resources/scripts/kafka_producer.py b/csit/resources/scripts/kafka_producer.py
deleted file mode 100755
index ff129872..00000000
--- a/csit/resources/scripts/kafka_producer.py
+++ /dev/null
@@ -1,41 +0,0 @@
-#!/usr/bin/env python3
-#
-# ============LICENSE_START====================================================
-#  Copyright (C) 2023 Nordix Foundation.
-# =============================================================================
-# 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.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END======================================================
-
-# Python utility to produce a message on a kafka topic
-# Accepts the arguments {topic_name} and {message}
-
-from confluent_kafka import Producer
-import sys
-
-def post_to_kafka(topic, message):
-    conf = {'bootstrap.servers': 'localhost:29092'}
-
-    producer = Producer(conf)
-    try:
-        producer.produce(topic, value=message.encode('utf-8'))
-        producer.flush()
-        print('Message posted to Kafka topic: {}'.format(topic))
-    except Exception as e:
-        print('Failed to post message: {}'.format(str(e)))
-    finally:
-        producer.flush()
-
-if __name__ == '__main__':
-    post_to_kafka(sys.argv[1], sys.argv[2])
diff --git a/csit/resources/scripts/make_topics.py b/csit/resources/scripts/make_topics.py
deleted file mode 100755
index daee4341..00000000
--- a/csit/resources/scripts/make_topics.py
+++ /dev/null
@@ -1,41 +0,0 @@
-#!/usr/bin/env python3
-#
-# ============LICENSE_START====================================================
-#  Copyright (C) 2023 Nordix Foundation.
-# =============================================================================
-# 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.
-#
-# SPDX-License-Identifier: Apache-2.0
-# ============LICENSE_END======================================================
-
-# Python utility to create a new kafka topic
-# Accepts the argument {topic_name}
-
-from confluent_kafka.admin import AdminClient, NewTopic
-import sys
-
-def create_topic(bootstrap_servers, topic_name, num_partitions=2, replication_factor=2):
-    admin_client = AdminClient({'bootstrap.servers': bootstrap_servers})
-
-    # Define the topic configuration
-    topic = NewTopic(topic_name, num_partitions=num_partitions, replication_factor=replication_factor)
-
-    # Create the topic
-    admin_client.create_topics([topic])
-
-
-if __name__ == '__main__':
-    topic_name = sys.argv[1]
-    bootstrap_servers = 'localhost:29092'
-
-    create_topic(bootstrap_servers, topic_name)
diff --git a/csit/resources/scripts/run-test.sh b/csit/resources/scripts/run-test.sh
index 02f06ff4..6bd7c07b 100755
--- a/csit/resources/scripts/run-test.sh
+++ b/csit/resources/scripts/run-test.sh
@@ -1,7 +1,7 @@
 #!/bin/bash
 #
 # ============LICENSE_START====================================================
-#  Copyright (C) 2023 Nordix Foundation.
+#  Copyright (C) 2023-2024 Nordix Foundation.
 # =============================================================================
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@ POLICY_PDPX_IP=policy-xacml-pdp:${DEFAULT_PORT}
 POLICY_DROOLS_IP=policy-drools-pdp:9696
 DISTRIBUTION_IP=policy-distribution:6969
 DMAAP_IP=message-router:3904
+KAFKA_IP=kafka:9092
 APEX_EVENTS_IP=policy-apex-pdp:23324
 PROMETHEUS_IP=prometheus:9090
 CLAMP_K8S_TEST=true
@@ -43,7 +44,7 @@ DIST_TEMP_FOLDER=/tmp/distribution
 export ROBOT_VARIABLES=
 ROBOT_VARIABLES="-v DATA:$DATA -v NODETEMPLATES:$NODETEMPLATES -v POLICY_API_IP:$POLICY_API_IP
 -v POLICY_RUNTIME_ACM_IP:$POLICY_RUNTIME_ACM_IP -v POLICY_PAP_IP:$POLICY_PAP_IP -v APEX_IP:$APEX_IP
--v APEX_EVENTS_IP:$APEX_EVENTS_IP -v DMAAP_IP:$DMAAP_IP -v PROMETHEUS_IP:${PROMETHEUS_IP}
+-v APEX_EVENTS_IP:$APEX_EVENTS_IP -v DMAAP_IP:$DMAAP_IP  -v KAFKA_IP:$KAFKA_IP -v PROMETHEUS_IP:${PROMETHEUS_IP}
 -v POLICY_PDPX_IP:$POLICY_PDPX_IP -v POLICY_DROOLS_IP:$POLICY_DROOLS_IP -v TEMP_FOLDER:${DIST_TEMP_FOLDER}
 -v DISTRIBUTION_IP:$DISTRIBUTION_IP -v CLAMP_K8S_TEST:$CLAMP_K8S_TEST"
 
diff --git a/csit/resources/scripts/setup-apex-pdp-postgres.sh b/csit/resources/scripts/setup-apex-pdp-postgres.sh
index f088da3c..150ec8f6 100755
--- a/csit/resources/scripts/setup-apex-pdp-postgres.sh
+++ b/csit/resources/scripts/setup-apex-pdp-postgres.sh
@@ -44,10 +44,10 @@ do
     sleep 10s
 done
 
-export DMAAP_IP="localhost:${DMAAP_PORT}"
+export KAFKA_IP="localhost:${KAFKA_PORT}"
 export SUITES="apex-pdp-test.robot"
 
 ROBOT_VARIABLES="-v POLICY_PAP_IP:localhost:${PAP_PORT} -v POLICY_API_IP:localhost:${API_PORT}
 -v PROMETHEUS_IP:localhost:${PROMETHEUS_PORT} -v DATA:${DATA} -v NODETEMPLATES:${NODETEMPLATES}
--v APEX_IP:localhost:${APEX_PORT} -v DMAAP_IP:${DMAAP_IP}
+-v APEX_IP:localhost:${APEX_PORT} -v KAFKA_IP:${KAFKA_IP}
 -v APEX_EVENTS_IP:localhost:${APEX_EVENTS_PORT}"
diff --git a/csit/resources/scripts/setup-apex-pdp.sh b/csit/resources/scripts/setup-apex-pdp.sh
index b9b1a78a..198a6017a 100755
--- a/csit/resources/scripts/setup-apex-pdp.sh
+++ b/csit/resources/scripts/setup-apex-pdp.sh
@@ -2,7 +2,7 @@
 # ============LICENSE_START=======================================================
 #  Copyright (C) 2018 Ericsson. All rights reserved.
 #
-#  Modifications Copyright (c) 2019-2023 Nordix Foundation.
+#  Modifications Copyright (c) 2019-2024 Nordix Foundation.
 #  Modifications Copyright (C) 2020-2021 AT&T Intellectual Property.
 #  Modifications Copyright (C) 2021 Bell Canada. All rights reserved.
 # ================================================================================
@@ -26,9 +26,9 @@ source "${SCRIPTS}"/setup-pap.sh
 # wait for the app to start up
 bash "${SCRIPTS}"/wait_for_rest.sh localhost ${APEX_PORT}
 
-export DMAAP_IP="localhost:${DMAAP_PORT}"
+export KAFKA_IP="kafka:${KAFKA_PORT}"
 export SUITES="apex-pdp-test.robot
 apex-slas.robot"
 
-ROBOT_VARIABLES="${ROBOT_VARIABLES} -v APEX_IP:localhost:${APEX_PORT} -v DMAAP_IP:${DMAAP_IP}
+ROBOT_VARIABLES="${ROBOT_VARIABLES} -v APEX_IP:localhost:${APEX_PORT} -v KAFKA_IP:${KAFKA_IP}
 -v APEX_EVENTS_IP:localhost:${APEX_EVENTS_PORT}"
diff --git a/csit/resources/scripts/setup-drools-applications.sh b/csit/resources/scripts/setup-drools-applications.sh
index d8542bd1..369874b6 100755
--- a/csit/resources/scripts/setup-drools-applications.sh
+++ b/csit/resources/scripts/setup-drools-applications.sh
@@ -2,7 +2,7 @@
 #
 # ===========LICENSE_START====================================================
 #  Copyright (C) 2019-2021 AT&T Intellectual Property. All rights reserved.
-#  Modifications Copyright 2021-2023 Nordix Foundation.
+#  Modifications Copyright 2021-2024 Nordix Foundation.
 # ============================================================================
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -35,4 +35,4 @@ sleep 15
 
 ROBOT_VARIABLES="-v DATA:${DATA} -v DROOLS_IP:localhost:${DROOLS_APPS_PORT}
 -v DROOLS_IP_2:localhost:${DROOLS_APPS_TELEMETRY_PORT} -v POLICY_API_IP:localhost:${API_PORT}
--v POLICY_PAP_IP:localhost:${PAP_PORT} -v DMAAP_IP:localhost:${DMAAP_PORT}"
+-v POLICY_PAP_IP:localhost:${PAP_PORT} -v KAFKA_IP:localhost:${KAFKA_PORT}"
diff --git a/csit/resources/scripts/setup-xacml-pdp.sh b/csit/resources/scripts/setup-xacml-pdp.sh
index 251cb29c..4511d91e 100755
--- a/csit/resources/scripts/setup-xacml-pdp.sh
+++ b/csit/resources/scripts/setup-xacml-pdp.sh
@@ -1,7 +1,7 @@
 #!/bin/bash
 # ============LICENSE_START=======================================================
 #  Copyright (C) 2020-2021 AT&T Intellectual Property. All rights reserved.
-#  Modifications Copyright 2021-2023 Nordix Foundation.
+#  Modifications Copyright 2021-2024 Nordix Foundation.
 # ================================================================================
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -32,4 +32,4 @@ export SUITES="xacml-pdp-test.robot"
 
 ROBOT_VARIABLES="-v DATA:${DATA} -v POLICY_PDPX_IP:localhost:${XACML_PORT}
 -v POLICY_API_IP:localhost:${API_PORT} -v POLICY_PAP_IP:localhost:${PAP_PORT}
--v DMAAP_IP:localhost:${DMAAP_PORT}"
+-v KAFKA_IP:localhost:${KAFKA_PORT}"