tests/dcaegen2-services-bbs-event-processor/bbs-testcases/resources/BbsLibrary.py

   1 import json
   2
   3 import docker
   4 import time
   5 from docker.utils.json_stream import json_stream
   6 from collections import OrderedDict
   7
   8
   9 class BbsLibrary(object):
  10
  11     def __init__(self):
  12         pass
  13
  14     @staticmethod
  15     def check_for_log(search_for):
  16         client = docker.from_env()
  17         container = client.containers.get('bbs')
  18
  19         alog = container.logs(stream=False, tail=1000)
  20         try:
  21             alog = alog.decode()
  22         except AttributeError:
  23             pass
  24
  25         found = alog.find(search_for)
  26         if found != -1:
  27             return True
  28         else:
  29             return False
  30
  31     @staticmethod
  32     def create_pnf_name_from_auth(json_file):
  33         json_to_python = json.loads(json_file)
  34         correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName")
  35         return correlation_id
  36
  37     @staticmethod
  38     def get_invalid_auth_elements(json_file):
  39         """
  40         Get the correlationId, oldState, newState, stateInterface, macAddress, swVersion elements
  41         from the invalid message and place the elements into a JSON object (string) as fields for comparision
  42         """
  43         eventString = json.loads(json_file)[0]
  44         json_to_python = json.loads(eventString.replace("\\", ""))
  45         correlation_id = json_to_python.get("event").get("commonEventHeader").get("sourceName")
  46         oldState = json_to_python.get("event").get("stateChangeFields").get("oldState")
  47         newState = json_to_python.get("event").get("stateChangeFields").get("newState")
  48         stateInterface = json_to_python.get("event").get("stateChangeFields").get("stateInterface")
  49         macAddress = json_to_python.get("event").get("stateChangeFields").get("additionalFields").get("macAddress")
  50         swVersion = json_to_python.get("event").get("stateChangeFields").get("additionalFields").get("swVersion")
  51         if swVersion is None:
  52             swVersion = ""
  53
  54         inv_fields = OrderedDict()
  55
  56         #inv_fields = dict()
  57         inv_fields['correlationId'] = correlation_id
  58         inv_fields['oldState'] = oldState
  59         inv_fields['newState'] = newState
  60         inv_fields['stateInterface'] = stateInterface
  61         inv_fields['macAddress'] = macAddress
  62         inv_fields['swVersion'] = swVersion
  63
  64         # Transform the dictionary to JSON string
  65         json_str = json.dumps(inv_fields)
  66
  67         # Need to remove spaces between elements
  68         json_str = json_str.replace(', ', ',')
  69         return json_str
  70
  71     @staticmethod
  72     def get_invalid_update_elements(json_file):
  73         """
  74         Get the correlationId, attachment-point, remote-id, cvlan, svlan, elements
  75         from the invalid message and place the elements into a JSON object (string) as fields for comparision
  76         """
  77         eventString = json.loads(json_file)[0]
  78         json_to_python = json.loads(eventString.replace("\\", ""))
  79         correlation_id = json_to_python.get("correlationId")
  80         attachmentPoint = json_to_python.get("additionalFields").get("attachment-point")
  81         remoteId = json_to_python.get("additionalFields").get("remote-id")
  82         cvlan = json_to_python.get("additionalFields").get("cvlan")
  83         svlan = json_to_python.get("additionalFields").get("svlan")
  84
  85         inv_fields = OrderedDict()
  86         #inv_fields = dict()
  87         inv_fields['correlationId'] = correlation_id
  88         inv_fields['attachment-point'] = attachmentPoint
  89         inv_fields['remote-id'] = remoteId
  90         inv_fields['cvlan'] = cvlan
  91         inv_fields['svlan'] = svlan
  92
  93         # Transform the dictionary to JSON string
  94         json_str = json.dumps(inv_fields)
  95
  96         # Need to remove spaces between elements
  97         json_str = json_str.replace(', ', ',')
  98         return json_str
  99
 100     @staticmethod
 101     def compare_policy(dmaap_policy, json_policy):
 102         resp = False
 103         try:
 104             python_policy = json.loads(json_policy).pop()
 105         except:
 106             python_policy = ""
 107
 108         try:
 109             python_dmaap_policy = json.loads(dmaap_policy)
 110         except:
 111             python_dmaap_policy = ""
 112
 113         try:
 114             d_policy = python_dmaap_policy[0].get("policyName")
 115         except:
 116             d_policy = ""
 117
 118         try:
 119             j_policy = python_policy.get("policyName")
 120         except:
 121             return "False"
 122
 123         resp = "False"
 124         if (d_policy == j_policy):
 125             resp = "True"
 126         return resp
 127
 128     @staticmethod
 129     def create_pnf_name_from_update(json_file):
 130         json_to_python = json.loads(json_file)
 131         correlation_id = json_to_python.get("correlationId")
 132         return correlation_id
 133
 134     @staticmethod
 135     def ensure_container_is_running(name):
 136
 137         client = docker.from_env()
 138
 139         if not BbsLibrary.is_in_status(client, name, "running"):
 140             print ("starting container", name)
 141             container = client.containers.get(name)
 142             container.start()
 143             BbsLibrary.wait_for_status(client, name, "running")
 144
 145         BbsLibrary.print_status(client)
 146
 147     @staticmethod
 148     def ensure_container_is_exited(name):
 149
 150         client = docker.from_env()
 151
 152         if not BbsLibrary.is_in_status(client, name, "exited"):
 153             print ("stopping container", name)
 154             container = client.containers.get(name)
 155             container.stop()
 156             BbsLibrary.wait_for_status(client, name, "exited")
 157
 158         BbsLibrary.print_status(client)
 159
 160     @staticmethod
 161     def print_status(client):
 162         print("containers status")
 163         for c in client.containers.list(all=True):
 164             print(c.name, "   ", c.status)
 165
 166     @staticmethod
 167     def wait_for_status(client, name, status):
 168         while not BbsLibrary.is_in_status(client, name, status):
 169             print ("waiting for container: ", name, "to be in status: ", status)
 170             time.sleep(3)
 171
 172     @staticmethod
 173     def is_in_status(client, name, status):
 174         return len(client.containers.list(all=True, filters={"name": "^/"+name+"$", "status": status})) == 1
 175