X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Ftest_o1.py;fp=smo-install%2Ftest%2Fpythonsdk%2Fsrc%2Forantests%2Ftest_o1.py;h=85771e83ec37a64f6b5390e4633cfcae8328d45c;hb=0eb37372a837a16225f5788895638b00da52ce43;hp=0000000000000000000000000000000000000000;hpb=5592b6267f4801163873f8fd734c9ab9adb2f368;p=it%2Fdep.git

diff --git a/smo-install/test/pythonsdk/src/orantests/test_o1.py b/smo-install/test/pythonsdk/src/orantests/test_o1.py
new file mode 100644
index 00000000..85771e83
--- /dev/null
+++ b/smo-install/test/pythonsdk/src/orantests/test_o1.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+from onapsdk.configuration import settings
+from oransdk.dmaap.dmaap import OranDmaap
+import json
+
+BASIC_AUTH = {}
+
+logging.config.dictConfig(settings.LOG_CONFIG)
+logger = logging.getLogger("test O1")
+
+def test_network_devices_registration():
+	logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP")
+	dmaap = OranDmaap()
+	# As the user has been registered in DMAAP during test session init, that call should return all sims registered by SDNR
+	events = dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()
+	# events should be a list of messages
+	logger.info("Verify if the number of events is well equals to the number of devices")
+	assert (len(events) >= 6)
+
+	for event in events:
+		logger.info(f"Checking event: {event}")
+		eventjson = json.loads(event)
+		logger.info("json ?"+str(eventjson))
+		assert(eventjson["event"]["commonEventHeader"]["sourceName"] in settings.NETWORK_SIMULATOR_DEVICES_LIST)
+
+		if "o-ru" in eventjson["event"]["commonEventHeader"]["sourceName"]:
+			logger.info(f"RU event detected checking SDNR has well registered it")
+			assert(eventjson["event"]["commonEventHeader"]["reportingEntityName"] == "ONAP SDN-R")
+		elif "o-du" in eventjson["event"]["commonEventHeader"]["sourceName"]:
+			logger.info(f"DU detected checking it has well registered itself")
+			assert("o-du" in eventjson["event"]["commonEventHeader"]["reportingEntityName"])