Loading src/tests/tools/mock_ietf_network_slice_sdn_ctrl/Dockerfile 0 → 100644 +37 −0 Original line number Diff line number Diff line # 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_ietf_network_slice_sdn_ctrl WORKDIR /var/teraflow/mock_ietf_network_slice_sdn_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", "MockIetfNetworkSliceSdnCtrl.py"] src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py 0 → 100644 +88 −0 Original line number Diff line number Diff line # 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 ResourceNetworkSlices import NetworkSliceService, NetworkSliceServices from ResourceConnectionGroups import ConnectionGroup BIND_ADDRESS = "0.0.0.0" BIND_PORT = 8443 BASE_URL = "/restconf/data/ietf-network-slice-service:network-slice-services" 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(NetworkSliceServices, BASE_URL) api.add_resource(NetworkSliceService, BASE_URL + "/slice-service=<string:slice_id>") api.add_resource( ConnectionGroup, BASE_URL + "/slice-service=<string:slice_id>/connection-groups/connection-group=<string:connection_group_id>", ) 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()) src/tests/tools/mock_ietf_network_slice_sdn_ctrl/README.md 0 → 100644 +31 −0 Original line number Diff line number Diff line # Mock IETF ACTN SDN Controller This REST server implements very basic support for the following YANG data models: - IETF YANG Data Model for Transport Network Client Signals (draft-ietf-ccamp-client-signal-yang-10) - Ref: https://datatracker.ietf.org/doc/draft-ietf-teas-ietf-network-slice-nbi-yang/ The aim of this server is to enable testing ietf netowrk slice service driver and ietf slice service service handler ## 1. Install requirements for the Mock IETF Network Slice SDN controller __NOTE__: if you run the Mock IETF Network Slice SDN 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_ietf_network_slice_sdn_ctrl/requirements.in ``` Run the Mock IETF Network Slice SDN Controller as follows: ```bash python src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py ``` ## 2. Run the Mock IETF Network Slice SDN controller Run the Mock IETF Network Slice SDN Controller as follows: ```bash python src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py ``` src/tests/tools/mock_ietf_network_slice_sdn_ctrl/ResourceConnectionGroups.py 0 → 100644 +31 −0 Original line number Diff line number Diff line # 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 CONNECTION_GROUPS = [] class ConnectionGroup(Resource): def get(self, slice_id: str, connection_group_id: str): return make_response(jsonify(CONNECTION_GROUPS), 200) def put(self, slice_id: str, connection_group_id: str): json_request = request.get_json() CONNECTION_GROUPS.append(json_request) return make_response(jsonify({}), 200) src/tests/tools/mock_ietf_network_slice_sdn_ctrl/ResourceNetworkSlices.py 0 → 100644 +39 −0 Original line number Diff line number Diff line # 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 NETWORK_SLICES = {} class NetworkSliceServices(Resource): def get(self): return make_response(jsonify(NETWORK_SLICES), 200) def post(self): json_request = request.get_json() name = json_request["network-slice-services"]["slice-service"][0]["id"] NETWORK_SLICES[name] = json_request return make_response(jsonify({}), 201) class NetworkSliceService(Resource): def delete(self, slice_id: str): slice = NETWORK_SLICES.pop(slice_id, None) data, status = ({}, 404) if slice is None else (slice, 204) return make_response(jsonify(data), status) Loading
src/tests/tools/mock_ietf_network_slice_sdn_ctrl/Dockerfile 0 → 100644 +37 −0 Original line number Diff line number Diff line # 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_ietf_network_slice_sdn_ctrl WORKDIR /var/teraflow/mock_ietf_network_slice_sdn_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", "MockIetfNetworkSliceSdnCtrl.py"]
src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py 0 → 100644 +88 −0 Original line number Diff line number Diff line # 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 ResourceNetworkSlices import NetworkSliceService, NetworkSliceServices from ResourceConnectionGroups import ConnectionGroup BIND_ADDRESS = "0.0.0.0" BIND_PORT = 8443 BASE_URL = "/restconf/data/ietf-network-slice-service:network-slice-services" 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(NetworkSliceServices, BASE_URL) api.add_resource(NetworkSliceService, BASE_URL + "/slice-service=<string:slice_id>") api.add_resource( ConnectionGroup, BASE_URL + "/slice-service=<string:slice_id>/connection-groups/connection-group=<string:connection_group_id>", ) 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())
src/tests/tools/mock_ietf_network_slice_sdn_ctrl/README.md 0 → 100644 +31 −0 Original line number Diff line number Diff line # Mock IETF ACTN SDN Controller This REST server implements very basic support for the following YANG data models: - IETF YANG Data Model for Transport Network Client Signals (draft-ietf-ccamp-client-signal-yang-10) - Ref: https://datatracker.ietf.org/doc/draft-ietf-teas-ietf-network-slice-nbi-yang/ The aim of this server is to enable testing ietf netowrk slice service driver and ietf slice service service handler ## 1. Install requirements for the Mock IETF Network Slice SDN controller __NOTE__: if you run the Mock IETF Network Slice SDN 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_ietf_network_slice_sdn_ctrl/requirements.in ``` Run the Mock IETF Network Slice SDN Controller as follows: ```bash python src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py ``` ## 2. Run the Mock IETF Network Slice SDN controller Run the Mock IETF Network Slice SDN Controller as follows: ```bash python src/tests/tools/mock_ietf_network_slice_sdn_ctrl/MockIetfNetworkSliceSdnCtrl.py ```
src/tests/tools/mock_ietf_network_slice_sdn_ctrl/ResourceConnectionGroups.py 0 → 100644 +31 −0 Original line number Diff line number Diff line # 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 CONNECTION_GROUPS = [] class ConnectionGroup(Resource): def get(self, slice_id: str, connection_group_id: str): return make_response(jsonify(CONNECTION_GROUPS), 200) def put(self, slice_id: str, connection_group_id: str): json_request = request.get_json() CONNECTION_GROUPS.append(json_request) return make_response(jsonify({}), 200)
src/tests/tools/mock_ietf_network_slice_sdn_ctrl/ResourceNetworkSlices.py 0 → 100644 +39 −0 Original line number Diff line number Diff line # 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 NETWORK_SLICES = {} class NetworkSliceServices(Resource): def get(self): return make_response(jsonify(NETWORK_SLICES), 200) def post(self): json_request = request.get_json() name = json_request["network-slice-services"]["slice-service"][0]["id"] NETWORK_SLICES[name] = json_request return make_response(jsonify({}), 201) class NetworkSliceService(Resource): def delete(self, slice_id: str): slice = NETWORK_SLICES.pop(slice_id, None) data, status = ({}, 404) if slice is None else (slice, 204) return make_response(jsonify(data), status)
