85771e83ec37a64f6b5390e4633cfcae8328d45c
[it/dep.git] / smo-install / test / pythonsdk / src / orantests / test_o1.py
1 #!/usr/bin/env python3
2 # SPDX-License-Identifier: Apache-2.0
3
4 import logging
5 from onapsdk.configuration import settings
6 from oransdk.dmaap.dmaap import OranDmaap
7 import json
8
9 BASIC_AUTH = {}
10
11 logging.config.dictConfig(settings.LOG_CONFIG)
12 logger = logging.getLogger("test O1")
13
14 def test_network_devices_registration():
15         logger.info("Verify if SDNR sends well the RU registration to VES by checking in DMAAP")
16         dmaap = OranDmaap()
17         # As the user has been registered in DMAAP during test session init, that call should return all sims registered by SDNR
18         events = dmaap.get_message_from_topic("unauthenticated.VES_PNFREG_OUTPUT", 10000, settings.DMAAP_GROUP, settings.DMAAP_USER).json()
19         # events should be a list of messages
20         logger.info("Verify if the number of events is well equals to the number of devices")
21         assert (len(events) >= 6)
22
23         for event in events:
24                 logger.info(f"Checking event: {event}")
25                 eventjson = json.loads(event)
26                 logger.info("json ?"+str(eventjson))
27                 assert(eventjson["event"]["commonEventHeader"]["sourceName"] in settings.NETWORK_SIMULATOR_DEVICES_LIST)
28
29                 if "o-ru" in eventjson["event"]["commonEventHeader"]["sourceName"]:
30                         logger.info(f"RU event detected checking SDNR has well registered it")
31                         assert(eventjson["event"]["commonEventHeader"]["reportingEntityName"] == "ONAP SDN-R")
32                 elif "o-du" in eventjson["event"]["commonEventHeader"]["sourceName"]:
33                         logger.info(f"DU detected checking it has well registered itself")
34                         assert("o-du" in eventjson["event"]["commonEventHeader"]["reportingEntityName"])