blob: f5bfcc852f1c9e0d0269eff8b0d6a6671d9d31af [file] [log] [blame]
#!/usr/bin/python
import requests
import json
#
# Creates client-side connectivity json
#
def tapi_client_input(sip_uuids):
create_input = {
"tapi-connectivity:input": {
"end-point" : [
{
"local-id": sip_uuids[0],
"service-interface-point": {
"service-interface-point-uuid" : sip_uuids[0]
}
}
,
{
"local-id": sip_uuids[1],
"service-interface-point": {
"service-interface-point-uuid" : sip_uuids[1]
}
}
]
}
}
return create_input
#
# Creates line-side connectivity json
#
def tapi_line_input(sip_uuids):
create_input = {
"tapi-connectivity:input" : {
"end-point" : [
{
"layer-protocol-qualifier" : "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC",
"role" : "UNKNOWN",
"local-id" : "Src_end_point",
"direction" : "BIDIRECTIONAL",
"service-interface-point" : {
"service-interface-point-uuid" : sip_uuids[0]
},
"protection-role" : "WORK",
"layer-protocol-name" : "PHOTONIC_MEDIA"
},
{
"direction" : "BIDIRECTIONAL",
"service-interface-point" : {
"service-interface-point-uuid" : sip_uuids[1]
},
"protection-role" : "WORK",
"layer-protocol-name" : "PHOTONIC_MEDIA",
"layer-protocol-qualifier" : "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC",
"role" : "UNKNOWN",
"local-id" : "Dst_end_point"
}
]
}
}
return create_input
#
# Obtains TAPI context through restconf
#
def get_context(url_context):
resp = requests.get(url_context, auth=('onos', 'rocks'))
if resp.status_code != 200:
raise Exception('GET {}'.format(resp.status_code))
return resp.json()
#
# Requests a connectivity service
#
def request_connection(url_connectivity, context):
# All Context SIPs
sips = context["tapi-common:context"]["service-interface-point"]
# Sorted Photonic Media SIPs. filter is an iterable
esips = list(filter(is_dsr_media, sorted(sips, key=lambda sip: sip["name"][0]["value"])))
endpoints = [esips[0], esips[-1]]
sip_uuids = []
for sip in endpoints:
sip_uuids.append(sip["uuid"])
for uuid in sip_uuids:
print(uuid)
create_input_json = json.dumps(tapi_client_input(sip_uuids))
print (create_input_json)
headers = {'Content-type': 'application/json'}
resp = requests.post(url_connectivity, data=create_input_json, headers=headers, auth=('onos', 'rocks'))
if resp.status_code != 200:
raise Exception('POST {}'.format(resp.status_code))
return resp
#
# Filter method used to keep only SIPs that are photonic_media
#
def is_photonic_media(sip):
return sip["layer-protocol-name"] == "PHOTONIC_MEDIA"
#
# Filter method used to keep only SIPs that are DSR
#
def is_dsr_media(sip):
return sip["layer-protocol-name"] == "DSR"
#
# Processes the topology to verify the correctness
#
def process_topology():
# TODO use method to parse topology
# Getting the Topology
# topology = context["tapi-common:context"]["tapi-topology:topology-context"]["topology"][0]
# nodes = topology["node"];
# links = topology["link"];
noop
#
# Create a client-side connection. Firstly, get the context, parsing for SIPs that connect
# with each other in line-side; Secondly, issue the request
#
def create_client_connection(url_context, url_connectivity):
context = get_context(url_context)
# select the first topo from all topologies
topo = context["tapi-common:context"]["tapi-topology:topology-context"]["topology"][0]
# select the first link from all links of topo
nep_pair = topo["link"][0]["node-edge-point"]
assert topo["uuid"] == nep_pair[0]["topology-uuid"]
assert topo["uuid"] == nep_pair[1]["topology-uuid"]
(src_onep, dst_onep) = (find_client_onep(nep_pair[0], topo["node"]),
find_client_onep(nep_pair[1], topo["node"]))
print "client side neps", (src_onep, dst_onep)
src_sip_uuid, dst_sip_uuid = \
(src_onep["mapped-service-interface-point"][0]["service-interface-point-uuid"],
dst_onep["mapped-service-interface-point"][0]["service-interface-point-uuid"])
print "\nBuild client-side connectivity:\n|Item|SRC|DST|\n|:--|:--|:--|\n|onos-cp|%s|%s|\n|connection id|%s|%s|\n|sip uuid|%s|%s|" % \
(src_onep["name"][2]["value"], dst_onep["name"][2]["value"],
src_onep["name"][1]["value"], dst_onep["name"][1]["value"],
src_sip_uuid, dst_sip_uuid)
create_input_json = json.dumps(tapi_client_input((src_sip_uuid, dst_sip_uuid)))
print "\nThe json content of creation operation for client-side connectivity service is \n\t\t%s." % \
create_input_json
headers = {'Content-type': 'application/json'}
resp = requests.post(url_connectivity, data=create_input_json, headers=headers, auth=('onos', 'rocks'))
if resp.status_code != 200:
raise Exception('POST {}'.format(resp.status_code))
return resp
#
# Parse array structure "name" under structure "owned node edge point"
#
def parse_value(arr):
rtn = {}
for item in arr:
rtn[item["value-name"]] = item["value"]
return rtn
#
# Find node edge point of node structure in topology with client-side port, by using nep with line-side port.
# The odtn-connection-id should be the same in both line-side nep and client-side nep
#
def find_client_onep(line_nep_in_link, nodes):
for node in nodes:
if node["uuid"] == line_nep_in_link["node-uuid"]:
conn_id = None
for onep in node["owned-node-edge-point"]:
if onep["uuid"] == line_nep_in_link["node-edge-point-uuid"]:
name = parse_value(onep["name"])
if name["odtn-port-type"] == "line":
conn_id = name["odtn-connection-id"]
break
if conn_id is None:
raise AssertionError("Cannot find owned node edge point with node id %s and nep id %s."
% (line_nep_in_link["node-uuid"], line_nep_in_link["node-edge-point-uuid"], ))
for onep in node["owned-node-edge-point"]:
name = parse_value(onep["name"])
if name["odtn-port-type"] == "client" and name["odtn-connection-id"] == conn_id:
return onep
return None
#
# Create a line-side connection. Firstly, get the context, parsing for SIPs with photonic_media type,
# and select one pair of them; Secondly, issue the request
#
def create_line_connection(url_context, url_connectivity):
context = get_context(url_context)
# select the first topo from all topologies
topo = context["tapi-common:context"]["tapi-topology:topology-context"]["topology"][0]
# select the first link from all links of topo
nep_pair = topo["link"][0]["node-edge-point"]
assert topo["uuid"] == nep_pair[0]["topology-uuid"]
assert topo["uuid"] == nep_pair[1]["topology-uuid"]
src_onep, dst_onep = (find_line_onep(nep_pair[0], topo["node"]),
find_line_onep(nep_pair[1], topo["node"]))
src_sip_uuid, dst_sip_uuid = \
(src_onep["mapped-service-interface-point"][0]["service-interface-point-uuid"],
dst_onep["mapped-service-interface-point"][0]["service-interface-point-uuid"])
print "\nBuild line-side connectivity:\n|Item|SRC|DST|\n|:--|:--|:--|\n|onos-cp|%s|%s|\n|connection id|%s|%s|\n|sip uuid|%s|%s|" % \
(src_onep["name"][2]["value"], dst_onep["name"][2]["value"],
src_onep["name"][1]["value"], dst_onep["name"][1]["value"],
src_sip_uuid, dst_sip_uuid)
create_input_json = json.dumps(tapi_line_input((src_sip_uuid, dst_sip_uuid)))
print "\nThe json content of creation operation for line-side connectivity service is \n\t\t%s." % \
create_input_json
headers = {'Content-type': 'application/json'}
resp = requests.post(url_connectivity, data=create_input_json, headers=headers, auth=('onos', 'rocks'))
if resp.status_code != 200:
raise Exception('POST {}'.format(resp.status_code))
return resp
def find_line_onep(line_nep_in_link, nodes):
for node in nodes:
if node["uuid"] == line_nep_in_link["node-uuid"]:
for onep in node["owned-node-edge-point"]:
if onep["uuid"] == line_nep_in_link["node-edge-point-uuid"]:
# check the length equals 1 to verify the 1-to-1 mapping relationship
assert len(onep["mapped-service-interface-point"]) == 1
return onep
return None
#
# Obtains existing connectivity services
#
def get_connection(url_connectivity, uuid):
# uuid is useless for this method
json = '{}'
headers = {'Content-type': 'application/json'}
resp = requests.post(url_connectivity, data=json, headers=headers, auth=('onos', 'rocks'))
if resp.status_code != 200:
raise Exception('POST {}'.format(resp.status_code))
return resp