smo-install/test/pythonsdk/src/orantests/test_dmaap.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: Apache-2.0
   3
   4 import logging
   5 from pathlib import Path
   6 from onapsdk.configuration import settings
   7 from oransdk.dmaap.dmaap import OranDmaap
   8 from oransdk.a1sim.a1sim import A1sim
   9 from oransdk.utils.jinja import jinja_env
  10
  11 BASIC_AUTH = {}
  12
  13 logging.config.dictConfig(settings.LOG_CONFIG)
  14 logger = logging.getLogger("test DMAAP")
  15
  16 #dmaap = OranDmaap()
  17 #logger.info("Get all the topics")
  18 #topiclist = dmaap.get_all_topics(BASIC_AUTH)
  19 #logger.info("response is: %s", topiclist)
  20
  21
  22 #logger.info("Create new topic")
  23 #topic = '{  "topicName": "unauthenticated.SEC_FAULT_OUTPUT",  "topicDescription": "test topic",  "partitionCount": 1,  "replicationCnCount": 1,  "transactionEnabled": "false"}'
  24 #response = dmaap.create_topic(topic, BASIC_AUTH)
  25 #logger.info("response is: %s", response)
  26
  27
  28 #logger.info("Get topics again")
  29 #topiclist = dmaap.get_all_topics(BASIC_AUTH)
  30 #logger.info("response is: %s", topiclist)
  31
  32
  33 def test_dmaap():
  34     logger.info("Get ric version for ost")
  35     a1sim = A1sim()
  36     version1 = a1sim.check_version(settings.A1SIM_OSC_URL)
  37
  38     status = a1sim.check_status(settings.A1SIM_OSC_URL)
  39
  40     number = a1sim.get_policy_number(settings.A1SIM_OSC_URL)
  41
  42     data = jinja_env().get_template("OSC/policy_type.json.j2").render()
  43     a1sim.create_policy_type(settings.A1SIM_OSC_URL, 1, data)
  44     assert(True)