Test Tools - Mock QKD Node:

FROM python:3.9-slim

# Install dependencies

RUN apt-get --yes --quiet --quiet update && \

    apt-get --yes --quiet --quiet install g++ git build-essential cmake libpcre2-dev python3-dev python3-cffi && \

    rm -rf /var/lib/apt/lists/*

# Download, build and install libyang. Note that APT package is outdated

# - Ref: https://github.com/CESNET/libyang

# - Ref: https://github.com/CESNET/libyang-python/

RUN mkdir -p /var/libyang

RUN git clone https://github.com/CESNET/libyang.git /var/libyang

WORKDIR /var/libyang

RUN git fetch

RUN git checkout v2.1.148

RUN mkdir -p /var/libyang/build

WORKDIR /var/libyang/build

RUN cmake -D CMAKE_BUILD_TYPE:String="Release" ..

RUN make

RUN make install

RUN ldconfig

# Set Python to show logs as they occur

ENV PYTHONUNBUFFERED=0

# Get generic Python packages

# Get Python dependencies

RUN python3 -m pip install --upgrade pip

RUN python3 -m pip install --upgrade setuptools wheel

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

RUN python3 -m pip install https://github.com/freeconf/lang/releases/download/v0.1.0-alpha/freeconf-0.1.0-py3-none-any.whl

RUN fc-lang-install -v

# Create component sub-folders, and copy content

RUN mkdir -p /var/teraflow/mock_qkd_node/{data,yang}

WORKDIR /var/teraflow/mock_qkd_node

RUN mkdir -p /var/mock_qkd_node/

WORKDIR /var/mock_qkd_node

COPY yang/. yang/

COPY requirements.in requirements.in

COPY wsgi.py wsgi.py

COPY YangValidator.py YangValidator.py

# Get specific Python packages

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

RUN python3 -m pip install -r requirements.txt

# Copy code and data

COPY ./yang ./yang

COPY ./startup.json ./startup.json

COPY ./qkd_node.py ./qkd_node.py

# Start the service

ENTRYPOINT ["gunicorn", "--workers", "1", "--bind", "0.0.0.0:8080", "wsgi:app"]

ENTRYPOINT ["python", "qkd_node.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

import json, os

from flask import jsonify, make_response, request

from flask_restful import Resource

DATA_FILE_PATH = os.environ.get('DATA_FILE_PATH')

if DATA_FILE_PATH is None:

    raise Exception('DataFile({:s}) is not defined'.format(str(DATA_FILE_PATH)))

if not os.path.isfile(DATA_FILE_PATH):

    raise Exception('DataFile({:s}) not found'.format(str(DATA_FILE_PATH)))

with open(DATA_FILE_PATH, mode='r', encoding='UTF-8') as fp:

    QKD_NODE = json.load(fp)

class QkdNode(Resource):

    def get(self):

        return make_response(jsonify(QKD_NODE), 200)

    def post(self):

        json_request = request.get_json()

        slice_id = json_request["network-slice-services"]["slice-service"][0]["id"]

        QKD_NODE[slice_id] = json_request

        return make_response(jsonify({}), 201)

    def delete(self, slice_id : str):

        slice = QKD_NODE.pop(slice_id, None)

        data, status = ({}, 404) if slice is None else (slice, 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.

import libyang, os

from typing import Dict, Optional

YANG_DIR = os.path.join(os.path.dirname(__file__), 'yang')

class YangValidator:

    def __init__(self, main_module : str, dependency_modules : [str]) -> None:

        self._yang_context = libyang.Context(YANG_DIR)

        self._yang_module = self._yang_context.load_module(main_module)

        mods = [self._yang_context.load_module(mod) for mod in dependency_modules] + [self._yang_module]

        for mod in mods:

            mod.feature_enable_all()

    def parse_to_dict(self, message : Dict) -> Dict:

        dnode : Optional[libyang.DNode] = self._yang_module.parse_data_dict(

            message, validate_present=True, validate=True, strict=True

        )

        if dnode is None: raise Exception('Unable to parse Message({:s})'.format(str(message)))

        message = dnode.print_dict()

        dnode.free()

        return message

    def destroy(self) -> None:

        self._yang_context.destroy()

cd $(dirname $0)

docker build -t mock-qkd-node:test -f Dockerfile .

docker tag mock-qkd-node:test localhost:32000/tfs/mock-qkd-node:test

docker push localhost:32000/tfs/mock-qkd-node:test

#docker tag mock-qkd-node:test localhost:32000/tfs/mock-qkd-node:test

#docker push localhost:32000/tfs/mock-qkd-node:test

{

    "node": {"qkdn_id": "00000001-0000-0000-0000-000000000000"},

    "qkdn_capabilities": {},

    "qkd_applications": {"qkd_app": []},

    "qkd_interfaces": {

        "qkd_interface": [

            {

                "qkdi_id": "100",

                "qkdi_att_point": {},

                "qkdi_capabilities": {}

            },

            {

                "qkdi_id": "101",

                "qkdi_att_point": {"device": current_ip, "port": "1001"},

                "qkdi_capabilities": {}

            }

        ]

    },

    "qkd_links": {"qkd_link": []}

}