CI/CD pipeline - QKD E2E test:

#!/bin/bash

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

#

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

# See the License for the specific language governing permissions and

# limitations under the License.

cd "$(dirname "$0")"

# 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

# Ensure the local bin directory is in the PATH

export PATH=$PATH:/home/gitlab-runner/.local/bin

import json, os

from flask import jsonify, make_response, request

from flask_restful import Resource

# Function to kill all background processes

killbg() {

    for p in "${pids[@]}" ; do

        kill "$p" 2>/dev/null;

    done

}

trap killbg EXIT

pids=()

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)))

# Set FLASK_APP and run the Flask instances on different ports

export FLASK_APP=wsgi

if not os.path.isfile(DATA_FILE_PATH):

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

# Starting Flask instances on different ports

for port in 11111 22222 33333; do

    flask run --host 0.0.0.0 --port "$port" &

    pids+=($!)

    sleep 2 # To avoid conflicts during startup, giving each Flask instance time to initialize

done

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

    QKD_NODE = json.load(fp)

# Wait for all background processes to finish

wait

class QkdNode(Resource):

    def get(self):

        return make_response(jsonify(QKD_NODE), 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)

    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)

# Mock IETF ACTN SDN Controller

# Mock QKD Node

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-ccamp-client-signal-yang/

- IETF YANG Data Model for Traffic Engineering Tunnels, Label Switched Paths and Interfaces (draft-ietf-teas-yang-te-34)

  - Ref: https://datatracker.ietf.org/doc/draft-ietf-teas-yang-te/

This Mock implements very basic support for the software-defined QKD node information models specified in ETSI GS QKD 015 V2.1.1.

The aim of this server is to enable testing the IetfActnDeviceDriver and the IetfActnServiceHandler.

The aim of this mock is to enable testing the TFS QKD Framework with an emulated data plane.

## 1. Install requirements for the Mock IETF ACTN SDN controller

__NOTE__: if you run the Mock IETF ACTN 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:

## Build the Mock QKD Node Docker image

```bash

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

./build.sh

```

Run the Mock IETF ACTN SDN Controller as follows:

```bash

python src/tests/tools/mock_ietf_actn_sdn_ctrl/MockIetfActnSdnCtrl.py

```

## 2. Run the Mock IETF ACTN SDN controller

Run the Mock IETF ACTN SDN Controller as follows:

## Run the Mock QKD Node as a container:

```bash

python src/tests/tools/mock_ietf_actn_sdn_ctrl/MockIetfActnSdnCtrl.py

docker network create --driver bridge --subnet=172.254.252.0/24 --gateway=172.254.252.254 tfs-qkd-net-mgmt

docker run --name qkd-node-01 --detach --publish 80:80 \

  --network=tfs-qkd-net-mgmt --ip=172.254.252.101 \

  --env "DATA_FILE_PATH=/var/teraflow/mock-qkd-node/data/database.json" \

  --volume "$PWD/src/tests/mock-qkd-node/data/database-01.json:/var/teraflow/mock-qkd-node/data/database.json" \

  mock-qkd-node:test

docker run --name qkd-node-02 --detach --publish 80:80 \

  --network=tfs-qkd-net-mgmt --ip=172.254.252.102 \

  --env "DATA_FILE_PATH=/var/teraflow/mock-qkd-node/data/database.json" \

  --volume "$PWD/src/tests/mock-qkd-node/data/database-02.json:/var/teraflow/mock-qkd-node/data/database.json" \

  mock-qkd-node:test

docker run --name qkd-node-03 --detach --publish 80:80 \

  --network=tfs-qkd-net-mgmt --ip=172.254.252.103 \

  --env "DATA_FILE_PATH=/var/teraflow/mock-qkd-node/data/database.json" \

  --volume "$PWD/src/tests/mock-qkd-node/data/database-03.json:/var/teraflow/mock-qkd-node/data/database.json" \

  mock-qkd-node:test

```

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

kubectl delete namespace mocks

kubectl --namespace mocks apply -f mock-qkd-node.yaml

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

kind: Namespace

apiVersion: v1

metadata:

  name: mocks

---

apiVersion: apps/v1

kind: Deployment

metadata:

  name: mock-qkd-node-01

spec:

  selector:

    matchLabels:

      app: mock-qkd-node-01

  replicas: 1

  template:

    metadata:

      annotations:

        config.linkerd.io/skip-inbound-ports: "8443"

      labels:

        app: mock-qkd-node-01

    spec:

      terminationGracePeriodSeconds: 5

      containers:

      - name: server

        image: localhost:32000/tfs/mock-qkd-node:test

        imagePullPolicy: IfNotPresent

        ports:

        - containerPort: 8443

        resources:

          requests:

            cpu: 250m

            memory: 512Mi

          limits:

            cpu: 700m

            memory: 1024Mi

---

apiVersion: v1

kind: Service

metadata:

  name: mock-qkd-node-01

  labels:

    app: mock-qkd-node-01

spec:

  type: ClusterIP

  selector:

    app: mock-qkd-node-01

  ports:

  - name: https

    port: 8443

    targetPort: 8443

# See the License for the specific language governing permissions and

# limitations under the License.

import json, os

import functools, json, logging, os, time

from flask import Flask, request

from flask_restful import Api

from ResourceNetworkSlices import NetworkSliceService, NetworkSliceServices

from ResourceConnectionGroups import ConnectionGroup

from YangValidator import YangValidator

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

if DATA_FILE_PATH is None:

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

LOG_LEVEL = logging.DEBUG

# TODO: if file not found, raise exception

logging.basicConfig(

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

)

LOGGER = logging.getLogger(__name__)

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

    data = json.load(fp)

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

BASE_URL = '/restconf/data/etsi-qkd-sdn-node:qkd_node'

yang_validator = YangValidator('etsi-qkd-sdn-node', ['etsi-qkd-node-types'])

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

app = Flask(__name__)

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

api = Api(app, prefix=BASE_URL)

api.add_resource(NetworkSliceServices, "")

api.add_resource(NetworkSliceService, "/slice-service=<string:slice_id>")

api.add_resource(

    ConnectionGroup,

    "/slice-service=<string:slice_id>/connection-groups/connection-group=<string:connection_group_id>",

)

def get_side_effect(url):

    steps = url.lstrip('https://').lstrip('http://').rstrip('/')