1 # -*- coding: utf-8 -*-
3 #######################################################################################
4 # This file is part of OSM RO module
6 # Copyright ETSI Contributors and Others.
8 # Licensed under the Apache License, Version 2.0 (the "License"); you may
9 # not use this file except in compliance with the License. You may obtain
10 # a copy of the License at
12 # http://www.apache.org/licenses/LICENSE-2.0
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
16 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17 # License for the specific language governing permissions and limitations
19 #######################################################################################
20 # This work has been performed in the context of the TeraFlow Project -
21 # funded by the European Commission under Grant number 101015857 through the
22 # Horizon 2020 program.
24 # - Lluis Gifre <lluis.gifre@cttc.es>
25 # - Ricard Vilalta <ricard.vilalta@cttc.es>
26 #######################################################################################
28 """This file contains a minimalistic Mock Transport API (TAPI) WIM server."""
35 PHOTONIC_PROTOCOL_QUALIFIER
= "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC"
36 DSR_PROTOCOL_QUALIFIER
= "tapi-dsr:DIGITAL_SIGNAL_TYPE"
40 uuid
, layer_protocol_name
, supported_layer_protocol_qualifier
, direction
44 "layer-protocol-name": layer_protocol_name
,
45 "supported-layer-protocol-qualifier": [supported_layer_protocol_qualifier
],
46 "administrative-state": "UNLOCKED",
47 "operational-state": "ENABLED",
48 "direction": direction
,
52 def compose_sip_dsr(uuid
):
53 return compose_sip(uuid
, "DSR", DSR_PROTOCOL_QUALIFIER
, "BIDIRECTIONAL")
56 def compose_sip_photonic_input(uuid
):
57 return compose_sip(uuid
, "PHOTONIC_MEDIA", PHOTONIC_PROTOCOL_QUALIFIER
, "INPUT")
60 def compose_sip_photonic_output(uuid
):
61 return compose_sip(uuid
, "PHOTONIC_MEDIA", PHOTONIC_PROTOCOL_QUALIFIER
, "OUTPUT")
65 "uuid": str(uuid
.uuid4()),
66 "service-interface-point": [
67 compose_sip_dsr("R1-eth0"),
68 compose_sip_dsr("R2-eth0"),
69 compose_sip_photonic_input("R3-opt1-rx"),
70 compose_sip_photonic_output("R3-opt1-tx"),
71 compose_sip_photonic_input("R4-opt1-rx"),
72 compose_sip_photonic_output("R4-opt1-tx"),
74 # topology details not used by the WIM connector
75 "topology-context": {},
76 "connectivity-context": {"connectivity-service": [], "connection": []},
80 class MockTapiRequestHandler(http
.server
.BaseHTTPRequestHandler
):
81 """Mock TAPI Request Handler for the unit tests"""
83 def do_GET(self
): # pylint: disable=invalid-name
84 """Handle GET requests"""
85 path
= self
.path
.replace("tapi-common:", "").replace("tapi-connectivity:", "")
87 if path
== "/restconf/data/context":
89 headers
= {"Content-Type": "application/json"}
91 elif path
== "/restconf/data/context/service-interface-point":
93 headers
= {"Content-Type": "application/json"}
94 data
= CONTEXT
["service-interface-point"]
95 data
= {"tapi-common:service-interface-point": data
}
96 elif path
== "/restconf/data/context/connectivity-context/connectivity-service":
98 headers
= {"Content-Type": "application/json"}
99 data
= CONTEXT
["connectivity-context"]["connectivity-service"]
100 data
= {"tapi-connectivity:connectivity-service": data
}
102 status
= 404 # not found
104 data
= {"error": "Not found"}
106 self
.send_response(status
)
107 for header_name
, header_value
in headers
.items():
108 self
.send_header(header_name
, header_value
)
110 data
= json
.dumps(data
)
111 self
.wfile
.write(data
.encode("UTF-8"))
113 def do_POST(self
): # pylint: disable=invalid-name
114 """Handle POST requests"""
115 path
= self
.path
.replace("tapi-common:", "").replace("tapi-connectivity:", "")
116 length
= int(self
.headers
["content-length"])
117 data
= json
.loads(self
.rfile
.read(length
))
119 if path
== "/restconf/data/context/connectivity-context":
120 if "tapi-connectivity:connectivity-service" in data
:
121 data
["connectivity-service"] = data
.pop(
122 "tapi-connectivity:connectivity-service"
126 isinstance(data
["connectivity-service"], list)
127 and len(data
["connectivity-service"]) > 0
129 data
["connectivity-service"] = data
["connectivity-service"][0]
131 conn_svc
= data
["connectivity-service"]
132 if "connectivity-constraint" in conn_svc
:
133 conn_constr
= conn_svc
.pop("connectivity-constraint")
134 if "requested-capacity" in conn_constr
:
135 req_cap
= conn_constr
.pop("requested-capacity")
136 conn_svc
["requested-capacity"] = req_cap
137 if "connectivity-direction" in conn_constr
:
138 conn_dir
= conn_constr
.pop("connectivity-direction")
139 conn_svc
["connectivity-direction"] = conn_dir
141 connection
= {"uuid": conn_svc
["uuid"], "connection-end-point": []}
142 conn_svc
["connection"] = [{"connection_uuid": conn_svc
["uuid"]}]
144 CONTEXT
["connectivity-context"]["connection"].append(connection
)
145 CONTEXT
["connectivity-context"]["connectivity-service"].append(conn_svc
)
147 status
= 201 # created
149 elif path
== "/restconf/operations/delete-connectivity-service":
150 if "tapi-connectivity:input" in data
:
151 data
["input"] = data
.pop("tapi-connectivity:input")
152 conn_svc_uuid
= data
["input"]["uuid"]
153 conn_ctx
= CONTEXT
["connectivity-context"]
155 # keep connectivity services and connections with different uuid
156 conn_ctx
["connection"] = [
157 conn
for conn
in conn_ctx
["connection"] if conn
["uuid"] != conn_svc_uuid
159 conn_ctx
["connectivity-service"] = [
161 for conn_svc
in conn_ctx
["connectivity-service"]
162 if conn_svc
["uuid"] != conn_svc_uuid
165 status
= 204 # ok, no content
168 status
= 404 # not found
171 self
.send_response(status
)
172 for header_name
, header_value
in headers
.items():
173 self
.send_header(header_name
, header_value
)