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('received status is not up-to-date: {0}'.format(status_tstamp))
  27
  28
  29 def replace_ise_source_address(ise_source_json, new_target):
  30     """
  31     Replace ise server url with given target
  32
  33     :param ise_source_json: ise source configuration as json
  34     :param new_target: current ise server url
  35     """
  36     ise_source_json['ise-source-config']['connection-config']['ise-rest-url'] = new_target
  37
  38
  39 def remove_endpoint_timestamp(endpoint_json):
  40     """
  41     Remove timestamp from given endpoint node and return plain text (for simple comparison)
  42     :param endpoint_json: endpoint node from DS/operational
  43     :return: plain text without timestamp
  44     """
  45     try:
  46         for address_endpoint in endpoint_json['endpoints']['address-endpoints']['address-endpoint']:
  47             del address_endpoint['timestamp']
  48     except KeyError:
  49         msg = 'No endpoint present - can not wipe timestamp'
  50         logger.debug(msg)
  51         raise ExecutionFailed(msg)
  52
  53     return json.dumps(endpoint_json)
  54
  55
  56 def resolve_sxp_node_is_enabled(sxp_node_json):
  57     """
  58     Try to get value of leaf enabled
  59     :param sxp_node_json: sxp node operational state
  60     :return: enabled value
  61     """
  62     enabled = None
  63     try:
  64         for node in sxp_node_json['node']:
  65             enabled = node['sxp-node:enabled']
  66     except KeyError:
  67         msg = 'No sxp node content present - can not read value of enabled'
  68         logger.debug(msg)
  69         raise ExecutionFailed(msg)
  70
  71     return enabled
  72
  73
  74 def replace_netconf_node_host(netconf_node_json, node_name, host_value):
  75     """
  76     Replace host value in netconf node configuration
  77     :param netconf_node_json: netconf node configuration
  78     :param node_name:  required node-name value
  79     :param host_value: required host value
  80     :return: plain text with replaced host value
  81     """
  82     try:
  83         for node in netconf_node_json['node']:
  84             node['netconf-node-topology:host'] = host_value
  85             node['node-id'] = node_name
  86     except KeyError:
  87         msg = 'No host found in given netconf node config'
  88         logger.debug(msg)
  89         raise ExecutionFailed(msg)
  90
  91     return json.dumps(netconf_node_json)
  92
  93
  94 def replace_ip_mgmt_address_in_forwarder(sf_forwarders_json, ip_mgmt_map):
  95     """
  96     Find and replace ip-mgmt-address values for corresponding forwarders names
  97     :param sf_forwarders_json: sfc forwarders json
  98     :param ip_mgmt_map: key=forwarder-name, value=ip-mgmt-address
  99     :return: plain sfc forwarders with replaced ip-mgmt-addresses
 100     """
 101     try:
 102         for sff in sf_forwarders_json['service-function-forwarders']['service-function-forwarder']:
 103             sff_name = sff['name']
 104             if sff_name in ip_mgmt_map:
 105                 sff['ip-mgmt-address'] = ip_mgmt_map[sff_name]
 106
 107     except KeyError:
 108         msg = 'Expected sff not found in given config'
 109         logger.debug(msg)
 110         raise ExecutionFailed(msg)
 111
 112     return json.dumps(sf_forwarders_json)
 113
 114
 115 def replace_renderer_policy_version(renderer_policy_json, next_version):
 116     """
 117     Find and replace version of given renderer-policy json
 118     :param renderer_policy_json:    renderer policy
 119     :param next_version:    version to be written into given policy
 120     :return: plain renderer policy with replaced version
 121     """
 122     try:
 123         renderer_policy_json['renderer-policy']['version'] = next_version
 124     except KeyError:
 125         msg = 'Expected version element not found in given renderer-policy'
 126         logger.debug(msg)
 127         raise ExecutionFailed(msg)
 128
 129     return json.dumps(renderer_policy_json)