RO-SDN-tapi/osm_rosdn_tapi/tests/mock_tapi_handler.py

   1 # -*- coding: utf-8 -*-
   2
   3 #######################################################################################
   4 # This file is part of OSM RO module
   5 #
   6 # Copyright ETSI Contributors and Others.
   7 #
   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
  11 #
  12 #         http://www.apache.org/licenses/LICENSE-2.0
  13 #
  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
  18 # under the License.
  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.
  23 # Contributors:
  24 # - Lluis Gifre <lluis.gifre@cttc.es>
  25 # - Ricard Vilalta <ricard.vilalta@cttc.es>
  26 #######################################################################################
  27
  28 """This file contains a minimalistic Mock Transport API (TAPI) WIM server."""
  29
  30 import http.server
  31 import json
  32 import uuid
  33
  34
  35 PHOTONIC_PROTOCOL_QUALIFIER = "tapi-photonic-media:PHOTONIC_LAYER_QUALIFIER_NMC"
  36 DSR_PROTOCOL_QUALIFIER = "tapi-dsr:DIGITAL_SIGNAL_TYPE"
  37
  38
  39 def compose_sip(
  40     uuid, layer_protocol_name, supported_layer_protocol_qualifier, direction
  41 ):
  42     return {
  43         "uuid": uuid,
  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,
  49     }
  50
  51
  52 def compose_sip_dsr(uuid):
  53     return compose_sip(uuid, "DSR", DSR_PROTOCOL_QUALIFIER, "BIDIRECTIONAL")
  54
  55
  56 def compose_sip_photonic_input(uuid):
  57     return compose_sip(uuid, "PHOTONIC_MEDIA", PHOTONIC_PROTOCOL_QUALIFIER, "INPUT")
  58
  59
  60 def compose_sip_photonic_output(uuid):
  61     return compose_sip(uuid, "PHOTONIC_MEDIA", PHOTONIC_PROTOCOL_QUALIFIER, "OUTPUT")
  62
  63
  64 CONTEXT = {
  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"),
  73     ],
  74     # topology details not used by the WIM connector
  75     "topology-context": {},
  76     "connectivity-context": {"connectivity-service": [], "connection": []},
  77 }
  78
  79
  80 class MockTapiRequestHandler(http.server.BaseHTTPRequestHandler):
  81     """Mock TAPI Request Handler for the unit tests"""
  82
  83     def do_GET(self):  # pylint: disable=invalid-name
  84         """Handle GET requests"""
  85         path = self.path.replace("tapi-common:", "").replace("tapi-connectivity:", "")
  86
  87         if path == "/restconf/data/context":
  88             status = 200  # ok
  89             headers = {"Content-Type": "application/json"}
  90             data = CONTEXT
  91         elif path == "/restconf/data/context/service-interface-point":
  92             status = 200  # ok
  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":
  97             status = 200  # ok
  98             headers = {"Content-Type": "application/json"}
  99             data = CONTEXT["connectivity-context"]["connectivity-service"]
 100             data = {"tapi-connectivity:connectivity-service": data}
 101         else:
 102             status = 404  # not found
 103             headers = {}
 104             data = {"error": "Not found"}
 105
 106         self.send_response(status)
 107         for header_name, header_value in headers.items():
 108             self.send_header(header_name, header_value)
 109         self.end_headers()
 110         data = json.dumps(data)
 111         self.wfile.write(data.encode("UTF-8"))
 112
 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))
 118
 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"
 123                 )
 124
 125             if (
 126                 isinstance(data["connectivity-service"], list)
 127                 and len(data["connectivity-service"]) > 0
 128             ):
 129                 data["connectivity-service"] = data["connectivity-service"][0]
 130
 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
 140
 141             connection = {"uuid": conn_svc["uuid"], "connection-end-point": []}
 142             conn_svc["connection"] = [{"connection_uuid": conn_svc["uuid"]}]
 143
 144             CONTEXT["connectivity-context"]["connection"].append(connection)
 145             CONTEXT["connectivity-context"]["connectivity-service"].append(conn_svc)
 146
 147             status = 201  # created
 148             headers = {}
 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"]
 154
 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
 158             ]
 159             conn_ctx["connectivity-service"] = [
 160                 conn_svc
 161                 for conn_svc in conn_ctx["connectivity-service"]
 162                 if conn_svc["uuid"] != conn_svc_uuid
 163             ]
 164
 165             status = 204  # ok, no content
 166             headers = {}
 167         else:
 168             status = 404  # not found
 169             headers = {}
 170
 171         self.send_response(status)
 172         for header_name, header_value in headers.items():
 173             self.send_header(header_name, header_value)
 174         self.end_headers()