Remove Dmaap configurations from CSITs

Issue-ID: POLICY-4880
Change-Id: I4f973ef1f7a173eaf5f196d9991d67d584e235fe
Signed-off-by: rameshiyer27 <ramesh.murugan.iyer@est.tech>

5 files changed, 19 insertions, 53 deletions

diff --git a/csit/resources/tests/apex-pdp-test.robot b/csit/resources/tests/apex-pdp-test.robot
index 5e4ea34b..37bcff6f 100644
--- a/csit/resources/tests/apex-pdp-test.robot
+++ b/csit/resources/tests/apex-pdp-test.robot
@@ -77,7 +77,7 @@ Metrics
 *** Keywords ***
 
 TriggerAndVerifyTestPnfPolicy
-    [Documentation]    Send TestPnf policy trigger event to DMaaP and read notifications to verify policy execution
+    [Documentation]    Send TestPnf policy trigger event to Kafka and read notifications to verify policy execution
     [Arguments]    ${topic}
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForPnfPolicy.json
     ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_cl_output    ${data}
@@ -86,13 +86,11 @@ TriggerAndVerifyTestPnfPolicy
     Run Keyword    CheckLogMessage    ${topic}    FINAL_SUCCESS    Successfully processed the VES event. Hostname is updated.
 
 TriggerAndVerifyTestVnfPolicy
-    [Documentation]    Send TestVnf policy trigger event to DMaaP and read notifications to verify policy execution
-    Create Session   apexSession  http://${KAFKA_IP}   max_retries=1
+    [Documentation]    Send TestVnf policy trigger event to Kafka and read notifications to verify policy execution
+    [Arguments]    ${topic}
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForVnfPolicy.json
-    &{headers}=  Create Dictionary    Content-Type=application/json    Accept=application/json
-    ${resp}=    POST On Session    apexSession    /events/unauthenticated.DCAE_POLICY_EXAMPLE_OUTPUT    data=${data}   headers=${headers}
-    Should Be Equal As Strings    ${resp.status_code}   200
-    Run Keyword    CheckLogMessage    ACTIVE    VES event has been received. Going to fetch VNF details from AAI.
-    Run Keyword    CheckLogMessage    SUCCESS    VNF details are received from AAI successfully. Sending ConfigModify request to CDS.
-    Run Keyword    CheckLogMessage    SUCCESS    ConfigModify request is successful. Sending restart request to CDS.
-    Run Keyword    CheckLogMessage    FINAL_SUCCESS    Successfully processed the VES Event. Restart is complete.
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_policy_example_output    ${data}
+    Run Keyword    CheckLogMessage    ${topic}    ACTIVE    VES event has been received. Going to fetch VNF details from AAI.
+    Run Keyword    CheckLogMessage    ${topic}    SUCCESS    VNF details are received from AAI successfully. Sending ConfigModify request to CDS.
+    Run Keyword    CheckLogMessage    ${topic}    SUCCESS    ConfigModify request is successful. Sending restart request to CDS.
+    Run Keyword    CheckLogMessage    ${topic}    FINAL_SUCCESS    Successfully processed the VES Event. Restart is complete.
diff --git a/csit/resources/tests/apex-slas-10.robot b/csit/resources/tests/apex-slas-10.robot
index 7639ce95..833bb837 100644
--- a/csit/resources/tests/apex-slas-10.robot
+++ b/csit/resources/tests/apex-slas-10.robot
@@ -25,14 +25,11 @@ ValidatePolicyExecutionAndEventRateLowComplexity
     CreateNodeTemplate  /policy/api/v1/nodetemplates  200  ${postjson}  1
     DeployPolicy
     Wait Until Keyword Succeeds    2 min    5 sec    QueryPolicyStatus  ${policyName}  defaultGroup  apex  ${pdpName}  onap.policies.native.Apex
-    GetTopic     APEX-CL-MGT2
-    Create Session   apexSession  http://${DMAAP_IP}   max_retries=1
+    GetKafkaTopic     apex-cl-mgt2
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForVnfPolicy.json
-    &{headers}=  Create Dictionary    Content-Type=application/json    Accept=application/json
     ${eventStartTime}=  Get Current Date
-    ${resp}=    POST On Session    apexSession    /events/unauthenticated.DCAE_POLICY_EXAMPLE_OUTPUT    data=${data}   headers=${headers}
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_policy_example_output    ${data}
     ${eventEndTime}=  Get Current Date
-    Should Be Equal As Strings    ${resp.status_code}   200
     ValidateEventExecution    ${eventStartTime}  ${eventEndTime}  100
 
 ValidatePolicyExecutionAndEventRateHighComplexity
@@ -58,14 +55,11 @@ ValidatePolicyExecutionAndEventRateModerateComplexity
     CreatePolicy  /policy/api/v1/policytypes/onap.policies.native.Apex/versions/1.0.0/policies  200  ${postjson}  ${policyName}  1.0.0
     DeployPolicy
     Wait Until Keyword Succeeds    2 min    5 sec    QueryPolicyStatus  ${policyName}  defaultGroup  apex  ${pdpName}  onap.policies.native.Apex
-    GetTopic     APEX-CL-MGT
-    Create Session   apexSession  http://${DMAAP_IP}   max_retries=1
+    GetKafkaTopic     apex-cl-mgt
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForPnfPolicy.json
-    &{headers}=  Create Dictionary    Content-Type=application/json    Accept=application/json
     ${eventStartTime}=  Get Current Date
-    ${resp}=    POST On Session    apexSession    /events/unauthenticated.DCAE_CL_OUTPUT    data=${data}   headers=${headers}
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_cl_output    ${data}
     ${eventEndTime}=  Get Current Date
-    Should Be Equal As Strings    ${resp.status_code}   200
     ValidateEventExecution    ${eventStartTime}  ${eventEndTime}  10
 
 WaitForPrometheusServer
diff --git a/csit/resources/tests/apex-slas-3.robot b/csit/resources/tests/apex-slas-3.robot
index 13d58957..c37c1cda 100644
--- a/csit/resources/tests/apex-slas-3.robot
+++ b/csit/resources/tests/apex-slas-3.robot
@@ -23,13 +23,10 @@ ValidatePolicyExecutionAndEventRateLowComplexity
     CreatePolicy  /policy/api/v1/policytypes/onap.policies.native.Apex/versions/1.0.0/policies  200  ${postjson}  ${policyName}  1.0.0
     DeployPolicy
     Wait Until Keyword Succeeds    2 min    5 sec    QueryPolicyStatus  ${policyName}  defaultGroup  apex  ${pdpName}  onap.policies.native.Apex
-    GetTopic     APEX-CL-MGT
-    Create Session   apexSession  http://${DMAAP_IP}   max_retries=1
+    GetKafkaTopic     apex-cl-mgt
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForPnfPolicy.json
-    &{headers}=  Create Dictionary    Content-Type=application/json    Accept=application/json
     ${eventStartTime}=  Get Current Date
-    ${resp}=    POST On Session    apexSession    /events/unauthenticated.DCAE_CL_OUTPUT    data=${data}   headers=${headers}
-    Should Be Equal As Strings    ${resp.status_code}   200
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_cl_output    ${data}
     ${eventEndTime}=  Get Current Date
     ValidateEventExecution    ${eventStartTime}  ${eventEndTime}  30
 
@@ -42,13 +39,10 @@ ValidatePolicyExecutionAndEventRateHighComplexity
     CreateNodeTemplate  /policy/api/v1/nodetemplates  200  ${postjson}  1
     DeployPolicy
     Wait Until Keyword Succeeds    2 min    5 sec    QueryPolicyStatus  ${policyName}  defaultGroup  apex  ${pdpName}  onap.policies.native.Apex
-    GetTopic     APEX-CL-MGT2
-    Create Session   apexSession  http://${DMAAP_IP}   max_retries=1
+    GetKafkaTopic     apex-cl-mgt2
     ${data}=    Get Binary File     ${CURDIR}/data/VesEventForVnfPolicy.json
-    &{headers}=  Create Dictionary    Content-Type=application/json    Accept=application/json
     ${eventStartTime}=  Get Current Date
-    ${resp}=    POST On Session    apexSession    /events/unauthenticated.DCAE_POLICY_EXAMPLE_OUTPUT    data=${data}   headers=${headers}
-    Should Be Equal As Strings    ${resp.status_code}   200
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_policy_example_output    ${data}
     ${eventEndTime}=  Get Current Date
     ValidateEventExecution    ${eventStartTime}  ${eventEndTime}  0.6
 
diff --git a/csit/resources/tests/common-library.robot b/csit/resources/tests/common-library.robot
index 531572d2..8693be14 100644
--- a/csit/resources/tests/common-library.robot
+++ b/csit/resources/tests/common-library.robot
@@ -131,23 +131,6 @@ ValidateResponseTime
     ${actualTime}=   Set Variable  ${rawNumber * ${1000}}
     Should Be True   ${actualTime} <= ${timeLimit}
 
-GetTopic
-    [Arguments]    ${topic}
-    Create Session   session  http://${DMAAP_IP}   max_retries=1
-    ${params}=  Create Dictionary    limit    1    timeout    0
-    ${resp}=    GET On Session    session    /events/${topic}/script/1    ${params}
-    Status Should Be    OK    ${resp}
-
-CheckTopic
-    [Arguments]    ${topic}    ${expected_status}
-    Create Session   session  http://${DMAAP_IP}   max_retries=1
-    ${params}=  Create Dictionary    limit    1
-    ${resp}=    GET On Session    session    /events/${topic}/script/1    ${params}
-    Log  Received response from dmaap ${resp.text}
-    Status Should Be    OK    ${resp}
-    Should Contain    ${resp.text}    ${expected_status}
-    [Return]    ${resp.text}
-
 CheckKafkaTopic
     [Arguments]    ${topic}    ${expected_status}
     ${resp}=    Run Process    ${CURDIR}/kafka_consumer.py    ${topic}    60    ${expected_status}
diff --git a/csit/resources/tests/drools-applications-test.robot b/csit/resources/tests/drools-applications-test.robot
index 5c6a1b5c..40497008 100644
--- a/csit/resources/tests/drools-applications-test.robot
+++ b/csit/resources/tests/drools-applications-test.robot
@@ -170,9 +170,6 @@ PerformPostRequest
 OnSet
     [Arguments]    ${file}
     ${data}=    Get File    ${file}
-    Create Session   session  http://${DMAAP_IP}   max_retries=1
-    ${headers}=  Create Dictionary  Content-Type=application/json
-    ${resp}=  POST On Session    session    /events/unauthenticateddcae_cl_output    headers=${headers}    data=${data}
-    Log    Response from dmaap ${resp.text}
-    Status Should Be    OK
-    [Return]    ${resp.text}
+    ${resp}=    Run Process    ${CURDIR}/kafka_producer.py    unauthenticated.dcae_cl_output    ${data}
+    Log    Response from kafka ${resp.stdout}
+    [Return]    ${resp.stdout}