csit/libraries/GbpSxp.py

   1 import isodate
   2 from robot.api import logger
   3 from robot.errors import ExecutionFailed
   4 import json
   5
   6
   7 def check_iso8601_datetime_younger_then_limit(status_timestamp_raw, limit):
   8     """
   9     Compare timestamp of given status to given limit, raise exception if status is older or equal to limit
  10
  11     :param status_timestamp_raw: json from DS/operational (ise-source status)
  12     :param limit: datetime value - status must be younger than this
  13     """
  14
  15     logger.debug("limit:{0}".format(limit))
  16
  17     timestamp_raw = status_timestamp_raw
  18     # 2016-11-23T13:25:00.733+01:00
  19     status_tstamp = isodate.parse_datetime(timestamp_raw)
  20     limit_tstamp = isodate.parse_datetime(limit)
  21     limit_tstamp = limit_tstamp.replace(tzinfo=status_tstamp.tzinfo)
  22
  23     if status_tstamp <= limit_tstamp:
  24         logger.info("status stamp --> {0}".format(status_tstamp))
  25         logger.info("limit        --> {0}".format(limit_tstamp))
  26         raise ExecutionFailed(
  27             "received status is not up-to-date: {0}".format(status_tstamp)
  28         )
  29
  30
  31 def replace_ise_source_address(ise_source_json, new_target):
  32     """
  33     Replace ise server url with given target
  34
  35     :param ise_source_json: ise source configuration as json
  36     :param new_target: current ise server url
  37     """
  38     ise_source_json["ise-source-config"]["connection-config"][
  39         "ise-rest-url"
  40     ] = new_target
  41
  42
  43 def remove_endpoint_timestamp(endpoint_json):
  44     """
  45     Remove timestamp from given endpoint node and return plain text (for simple comparison)
  46     :param endpoint_json: endpoint node from DS/operational
  47     :return: plain text without timestamp
  48     """
  49     try:
  50         for address_endpoint in endpoint_json["endpoints"]["address-endpoints"][
  51             "address-endpoint"
  52         ]:
  53             del address_endpoint["timestamp"]
  54     except KeyError:
  55         msg = "No endpoint present - can not wipe timestamp"
  56         logger.debug(msg)
  57         raise ExecutionFailed(msg)
  58
  59     return json.dumps(endpoint_json)
  60
  61
  62 def resolve_sxp_node_is_enabled(sxp_node_json):
  63     """
  64     Try to get value of leaf enabled
  65     :param sxp_node_json: sxp node operational state
  66     :return: enabled value
  67     """
  68     enabled = None
  69     try:
  70         for node in sxp_node_json["node"]:
  71             enabled = node["sxp-node:enabled"]
  72     except KeyError:
  73         msg = "No sxp node content present - can not read value of enabled"
  74         logger.debug(msg)
  75         raise ExecutionFailed(msg)
  76
  77     return enabled
  78
  79
  80 def replace_netconf_node_host(netconf_node_json, node_name, host_value):
  81     """
  82     Replace host value in netconf node configuration
  83     :param netconf_node_json: netconf node configuration
  84     :param node_name:  required node-name value
  85     :param host_value: required host value
  86     :return: plain text with replaced host value
  87     """
  88     try:
  89         for node in netconf_node_json["node"]:
  90             node["netconf-node-topology:host"] = host_value
  91             node["node-id"] = node_name
  92     except KeyError:
  93         msg = "No host found in given netconf node config"
  94         logger.debug(msg)
  95         raise ExecutionFailed(msg)
  96
  97     return json.dumps(netconf_node_json)
  98
  99
 100 def replace_ip_mgmt_address_in_forwarder(sf_forwarders_json, ip_mgmt_map):
 101     """
 102     Find and replace ip-mgmt-address values for corresponding forwarders names
 103     :param sf_forwarders_json: sfc forwarders json
 104     :param ip_mgmt_map: key=forwarder-name, value=ip-mgmt-address
 105     :return: plain sfc forwarders with replaced ip-mgmt-addresses
 106     """
 107     try:
 108         for sff in sf_forwarders_json["service-function-forwarders"][
 109             "service-function-forwarder"
 110         ]:
 111             sff_name = sff["name"]
 112             if sff_name in ip_mgmt_map:
 113                 sff["ip-mgmt-address"] = ip_mgmt_map[sff_name]
 114
 115     except KeyError:
 116         msg = "Expected sff not found in given config"
 117         logger.debug(msg)
 118         raise ExecutionFailed(msg)
 119
 120     return json.dumps(sf_forwarders_json)
 121
 122
 123 def replace_renderer_policy_version(renderer_policy_json, next_version):
 124     """
 125     Find and replace version of given renderer-policy json
 126     :param renderer_policy_json:    renderer policy
 127     :param next_version:    version to be written into given policy
 128     :return: plain renderer policy with replaced version
 129     """
 130     try:
 131         renderer_policy_json["renderer-policy"]["version"] = next_version
 132     except KeyError:
 133         msg = "Expected version element not found in given renderer-policy"
 134         logger.debug(msg)
 135         raise ExecutionFailed(msg)
 136
 137     return json.dumps(renderer_policy_json)