Resolve "(CTTC) Enhanced RESTCONF-based OpenConfig NBI for DSCM Pluggables"

#!/bin/bash

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

PROJECTDIR=`pwd`

cd $PROJECTDIR/src

# test DSCM NBI functions

python3 -m pytest --log-level=INFO --log-cli-level=INFO --verbose \

    nbi/tests/test_dscm_restconf.py::test_post_get_delete_leaf_optical_channel_frequency

# # test JSON to Proto conversion functions

# python3 -m pytest --log-level=INFO --log-cli-level=INFO --verbose \

#     nbi/tests/test_json_to_proto.py::test_create_pluggable_request_hub_format \

#     nbi/tests/test_json_to_proto.py::test_create_pluggable_request_leaf_format \

#     nbi/tests/test_json_to_proto.py::test_configure_pluggable_request_hub_format \

#     nbi/tests/test_json_to_proto.py::test_configure_pluggable_request_leaf_format \

#     nbi/tests/test_json_to_proto.py::test_empty_payload

    OPEN_ROADM                      = 'openroadm'

    MORPHEUS                        = 'morpheus'

    OPENFLOW_RYU_CONTROLLER         = 'openflow-ryu-controller'

    DSCM_NODE                       = 'dscm'

    # ETSI TeraFlowSDN controller

    TERAFLOWSDN_CONTROLLER          = 'teraflowsdn'

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from .routes import blueprint

from nbi.service.NbiApplication import NbiApplication

# NBI service calls this in main.py file to register blueprints.

def register_dscm_oc(nbi_application : NbiApplication):

    nbi_flask_app = nbi_application.get_flask_app()

    nbi_flask_app.register_blueprint(blueprint, url_prefix='/restconf/data')

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

from functools import wraps

from flask import request

from .error import yang_error

def require_accept(allowed):

    def deco(fn):

        @wraps(fn)

        def _wrap(*args, **kwargs):

            accept = request.headers.get("Accept", "")

            if not any(a in accept or accept == "*/*" for a in allowed):

                return yang_error({"error-message": f"Accept not supported. Use one of {allowed}"}, status=406)

            return fn(*args, **kwargs)

        return _wrap

    return deco

def require_content_type(allowed):

    def deco(fn):

        @wraps(fn)

        def _wrap(*args, **kwargs):

            ctype = request.headers.get("Content-Type", "")

            if not any(a in ctype for a in allowed):

                return yang_error({"error-message": f"Content-Type not supported. Use one of {allowed}"}, status=415)

            return fn(*args, **kwargs)

        return _wrap

    return deco

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

#      http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import json

from flask import Response

YANG_JSON = "application/yang-data+json"

ERR_JSON  = "application/yang-errors+json"

def yang_json(data, status=200):

    return Response(json.dumps(data, ensure_ascii=False), status=status, mimetype=YANG_JSON)

def yang_error(err_dict, status=400):

    body = {"errors": {"error": [err_dict]}}

    return Response(json.dumps(body, ensure_ascii=False), status=status, mimetype=ERR_JSON)

def _bad_request(msg, path=None):

    return yang_error({"error-type": "protocol", "error-tag": "operation-failed",

                       "error-message": msg, **({"error-path": path} if path else {})}, status=400)

def _not_found(msg, path=None):

    return yang_error({"error-type": "application", "error-tag": "data-missing",

                       "error-message": msg, **({"error-path": path} if path else {})}, status=404)