NCE controller mock added

# 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 python:3.9-slim

# Set Python to show logs as they occur

ENV PYTHONUNBUFFERED=0

# Get generic Python packages

RUN python3 -m pip install --upgrade pip

RUN python3 -m pip install --upgrade setuptools wheel

RUN python3 -m pip install --upgrade pip-tools

# Create component sub-folders, and copy content

RUN mkdir -p /var/teraflow/mock_nce_ctrl

WORKDIR /var/teraflow/mock_nce_ctrl

COPY . .

# Get specific Python packages

RUN pip-compile --quiet --output-file=requirements.txt requirements.in

RUN python3 -m pip install -r requirements.txt

RUN python3 -m pip list

# Start the service

ENTRYPOINT ["python", "MockNCECtrl.py"]

# 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.

# Mock IETF ACTN SDN controller

# -----------------------------

# REST server implementing minimal support for:

# - IETF YANG Data Model for Transport Network Client Signals

#       Ref: https://www.ietf.org/archive/id/draft-ietf-ccamp-client-signal-yang-10.html

# - IETF YANG Data Model for Traffic Engineering Tunnels, Label Switched Paths and Interfaces

#       Ref: https://www.ietf.org/archive/id/draft-ietf-teas-yang-te-34.html

import functools, logging, sys, time

from flask import Flask, jsonify, make_response, request

from flask_restful import Api, Resource

from ResourceApps import Apps, App

from ResourceAppFlows import AppFlows, AppFlow

BIND_ADDRESS = "0.0.0.0"

BIND_PORT = 8443

BASE_URL = "/restconf/v1/data/app-flows"

STR_ENDPOINT = "http://{:s}:{:s}{:s}".format(

    str(BIND_ADDRESS), str(BIND_PORT), str(BASE_URL)

)

LOG_LEVEL = logging.DEBUG

logging.basicConfig(

    level=LOG_LEVEL, format="[%(asctime)s] %(levelname)s:%(name)s:%(message)s"

)

LOGGER = logging.getLogger(__name__)

logging.getLogger("werkzeug").setLevel(logging.WARNING)

def log_request(logger: logging.Logger, response):

    timestamp = time.strftime("[%Y-%b-%d %H:%M]")

    logger.info(

        "%s %s %s %s %s",

        timestamp,

        request.remote_addr,

        request.method,

        request.full_path,

        response.status,

    )

    return response

class Health(Resource):

    def get(self):

        return make_response(jsonify({}), 200)

def main():

    LOGGER.info("Starting...")

    app = Flask(__name__)

    app.after_request(functools.partial(log_request, LOGGER))

    api = Api(app, prefix=BASE_URL)

    api.add_resource(Health, "/")

    api.add_resource(Apps, BASE_URL + "/apps")

    api.add_resource(App, BASE_URL + "/application=<string:app_name>")

    api.add_resource(AppFlows, BASE_URL)

    api.add_resource(AppFlow, BASE_URL + "/app-flow=<string:app_name>")

    LOGGER.info("Listening on {:s}...".format(str(STR_ENDPOINT)))

    app.run(debug=True, host=BIND_ADDRESS, port=BIND_PORT, ssl_context="adhoc")

    LOGGER.info("Bye")

    return 0

if __name__ == "__main__":

    sys.exit(main())

# Mock NCE Controller

This REST server implements very basic support for the NCE access controller.

The aim of this server is to enable testing IETF Network Slice NBI, NCE driver and NCE service handler.

## 1. Install requirements for the Mock NCE controller

__NOTE__: if you run the Mock NCE controller from the PyEnv used for developing on the TeraFlowSDN

framework and you followed the official steps in

[Development Guide > Configure Environment > Python](https://labs.etsi.org/rep/tfs/controller/-/wikis/2.-Development-Guide/2.1.-Configure-Environment/2.1.1.-Python),

all the requirements are already in place. Install them only if you execute it in a separate/standalone environment.

Install the required dependencies as follows:

```bash

pip install -r src/tests/tools/mock_nce_ctrl/requirements.in

```

Run the Mock NCE Controller as follows:

```bash

python src/tests/tools/mock_nce_ctrl/MockNCECtrl.py

```

## 2. Run the Mock NCE controller

Run the Mock NCE Controller as follows:

```bash

python src/tests/tools/mock_nce_ctrl/MockNCECtrl.py

```

# 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.

# REST-API resource implementing minimal support for "IETF YANG Data Model for Transport Network Client Signals".

# Ref: https://www.ietf.org/archive/id/draft-ietf-ccamp-client-signal-yang-10.html

from flask import jsonify, make_response, request

from flask_restful import Resource

APP_FLOWS = {}

class AppFlows(Resource):

    def get(self):

        return make_response(jsonify(APP_FLOWS), 200)

    def post(self):

        json_request = request.get_json()

        name = json_request["app-flow"][0]["app-name"]

        APP_FLOWS[name] = json_request

        return make_response(jsonify({}), 201)

class AppFlow(Resource):

    def delete(self, app_name: str):

        app_flow = APP_FLOWS.pop(app_name, None)

        data, status = ({}, 404) if app_flow is None else (app_flow, 204)

        return make_response(jsonify(data), status)

# 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.

# REST-API resource implementing minimal support for "IETF YANG Data Model for Transport Network Client Signals".

# Ref: https://www.ietf.org/archive/id/draft-ietf-ccamp-client-signal-yang-10.html

from flask import jsonify, make_response, request

from flask_restful import Resource

APPS = {}

class Apps(Resource):

    def get(self):

        return make_response(jsonify(APPS), 200)

    def post(self):

        json_request = request.get_json()

        name = json_request["application"][0]["name"]

        APPS[name] = json_request

        return make_response(jsonify({}), 201)

class App(Resource):

    def delete(self, app_name: str):

        app_flow = APPS.pop(app_name, None)

        data, status = ({}, 404) if app_flow is None else (app_flow, 204)

        return make_response(jsonify(data), status)