2 from robot.api import logger
3 from robot.errors import ExecutionFailed
7 def check_iso8601_datetime_younger_then_limit(status_timestamp_raw, limit):
9 Compare timestamp of given status to given limit, raise exception if status is older or equal to limit
11 :param status_timestamp_raw: json from DS/operational (ise-source status)
12 :param limit: datetime value - status must be younger than this
15 logger.debug('limit:{0}'.format(limit))
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)
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))
29 def replace_ise_source_address(ise_source_json, new_target):
31 Replace ise server url with given target
33 :param ise_source_json: ise source configuration as json
34 :param new_target: current ise server url
36 ise_source_json['ise-source-config']['connection-config']['ise-rest-url'] = new_target
39 def remove_endpoint_timestamp(endpoint_json):
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
46 for address_endpoint in endpoint_json['endpoints']['address-endpoints']['address-endpoint']:
47 del address_endpoint['timestamp']
49 msg = 'No endpoint present - can not wipe timestamp'
51 raise ExecutionFailed(msg)
53 return json.dumps(endpoint_json)
56 def resolve_sxp_node_is_enabled(sxp_node_json):
58 Try to get value of leaf enabled
59 :param sxp_node_json: sxp node operational state
60 :return: enabled value
64 for node in sxp_node_json['node']:
65 enabled = node['sxp-node:enabled']
67 msg = 'No sxp node content present - can not read value of enabled'
69 raise ExecutionFailed(msg)
74 def replace_netconf_node_host(netconf_node_json, node_name, host_value):
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
83 for node in netconf_node_json['node']:
84 node['netconf-node-topology:host'] = host_value
85 node['node-id'] = node_name
87 msg = 'No host found in given netconf node config'
89 raise ExecutionFailed(msg)
91 return json.dumps(netconf_node_json)
94 def replace_ip_mgmt_address_in_forwarder(sf_forwarders_json, ip_mgmt_map):
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
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]
108 msg = 'Expected sff not found in given config'
110 raise ExecutionFailed(msg)
112 return json.dumps(sf_forwarders_json)
115 def replace_renderer_policy_version(renderer_policy_json, next_version):
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
123 renderer_policy_json['renderer-policy']['version'] = next_version
125 msg = 'Expected version element not found in given renderer-policy'
127 raise ExecutionFailed(msg)
129 return json.dumps(renderer_policy_json)