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(
27 "received status is not up-to-date: {0}".format(status_tstamp)
31 def replace_ise_source_address(ise_source_json, new_target):
33 Replace ise server url with given target
35 :param ise_source_json: ise source configuration as json
36 :param new_target: current ise server url
38 ise_source_json["ise-source-config"]["connection-config"][
43 def remove_endpoint_timestamp(endpoint_json):
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
50 for address_endpoint in endpoint_json["endpoints"]["address-endpoints"][
53 del address_endpoint["timestamp"]
55 msg = "No endpoint present - can not wipe timestamp"
57 raise ExecutionFailed(msg)
59 return json.dumps(endpoint_json)
62 def resolve_sxp_node_is_enabled(sxp_node_json):
64 Try to get value of leaf enabled
65 :param sxp_node_json: sxp node operational state
66 :return: enabled value
70 for node in sxp_node_json["node"]:
71 enabled = node["sxp-node:enabled"]
73 msg = "No sxp node content present - can not read value of enabled"
75 raise ExecutionFailed(msg)
80 def replace_netconf_node_host(netconf_node_json, node_name, host_value):
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
89 for node in netconf_node_json["node"]:
90 node["netconf-node-topology:host"] = host_value
91 node["node-id"] = node_name
93 msg = "No host found in given netconf node config"
95 raise ExecutionFailed(msg)
97 return json.dumps(netconf_node_json)
100 def replace_ip_mgmt_address_in_forwarder(sf_forwarders_json, ip_mgmt_map):
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
108 for sff in sf_forwarders_json["service-function-forwarders"][
109 "service-function-forwarder"
111 sff_name = sff["name"]
112 if sff_name in ip_mgmt_map:
113 sff["ip-mgmt-address"] = ip_mgmt_map[sff_name]
116 msg = "Expected sff not found in given config"
118 raise ExecutionFailed(msg)
120 return json.dumps(sf_forwarders_json)
123 def replace_renderer_policy_version(renderer_policy_json, next_version):
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
131 renderer_policy_json["renderer-policy"]["version"] = next_version
133 msg = "Expected version element not found in given renderer-policy"
135 raise ExecutionFailed(msg)
137 return json.dumps(renderer_policy_json)