scottk/wlan-testing
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-testing.git synced 2025-12-24 22:47:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

healthcheck method modified to ap event test

Browse Source 
Signed-off-by: anil-tegala <anil.tegala@candelatech.com>
This commit is contained in:
anil-tegala
2023-04-24 17:11:20 +05:30
parent d384c02ae3
commit 01da87bd5f
1 changed files with 11 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/kafka_tests/ap_events/test_kafka_ap_events.py
View File

@@ -8,20 +8,20 @@ import pytest
@allure.feature("Test Kafka Messages")
@allure.title("Real Time AP Events")
@pytest.mark.ap_events
class TestKafkaEvents(object):
class TestKafkaApEvents(object):
# Pytest unit test for validating Kafka healthcheck messages
@allure.title("Test HealthCheck Messages")
@pytest.mark.health_check
def test_kafka_healthcheck(self, kafka_consumer_healthcheck):
@allure.title("Test AP events")
@pytest.mark.ap_events
def test_kafka_ap_event(self, kafka_consumer_deq):
# Consume messages and validate them
for message in kafka_consumer_healthcheck:
for event in kafka_consumer_deq:
# Apply validation logic on message value
print(message)
print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
message.offset, message.key,
message.value))
if message.value is not None:
allure.attach(name="Kafka Health Check Message Info", body=str(message.value))
print(event)
print("%s:%d:%d: key=%s value=%s" % (event.topic, event.partition,
event.offset, event.key,
event.value))
if event.value is not None:
allure.attach(name="Kafka AP Event Info", body=str(event.value))
break
# Assert that the message is valid