NBI component - TFS-API connector:

import json

import logging

from typing import Dict, List

import grpc

from flask.json import jsonify

from flask_restful import Resource, request

from werkzeug.exceptions import BadRequest

LOGGER = logging.getLogger(__name__)

_GRPC_TO_HTTP_STATUS = {

    grpc.StatusCode.INVALID_ARGUMENT: 400,

    grpc.StatusCode.FAILED_PRECONDITION: 400,

    grpc.StatusCode.NOT_FOUND: 404,

    grpc.StatusCode.ALREADY_EXISTS: 409,

    grpc.StatusCode.PERMISSION_DENIED: 403,

    grpc.StatusCode.UNAUTHENTICATED: 401,

    grpc.StatusCode.UNAVAILABLE: 503,

    grpc.StatusCode.DEADLINE_EXCEEDED: 504,

}

def _format_grpc_error(exc : grpc.RpcError):

    grpc_status = exc.code()

    http_status = _GRPC_TO_HTTP_STATUS.get(grpc_status, 500)

    details = exc.details() if hasattr(exc, 'details') else str(exc)

    LOGGER.warning('Mapping gRPC error to HTTP status: grpc_status=%s http_status=%s details=%s',

                   grpc_status, http_status, details)

    return jsonify({

        'error': grpc_status.name if grpc_status is not None else 'UNKNOWN',

        'message': details,

    }), http_status

class _Resource(Resource):

    def __init__(self) -> None:

        for reservation in json_requests:

            if context_uuid != reservation['reservation_id']['context_id']['context_uuid']['uuid']:

                raise BadRequest('Mismatching context_uuid')

        try:

            return jsonify([

                grpc_message_to_json(self.context_client.SetOpticalSpectrumReservation(

                    grpc_optical_spectrum_reservation(reservation)

                ))

                for reservation in json_requests

            ])

        except grpc.RpcError as exc:

            return _format_grpc_error(exc)

class OpticalSpectrumReservation(_Resource):

    def get(self, context_uuid : str, reservation_uuid : str):

            raise BadRequest('Mismatching context_uuid')

        if reservation_uuid != reservation['reservation_id']['reservation_uuid']['uuid']:

            raise BadRequest('Mismatching reservation_uuid')

        try:

            return format_grpc_to_json(self.context_client.SetOpticalSpectrumReservation(

                grpc_optical_spectrum_reservation(reservation)

            ))

        except grpc.RpcError as exc:

            return _format_grpc_error(exc)

    def delete(self, context_uuid : str, reservation_uuid : str):

        try:

            return format_grpc_to_json(self.context_client.ReleaseOpticalSpectrumReservation(

                grpc_optical_spectrum_reservation_id(context_uuid, reservation_uuid)

            ))

        except grpc.RpcError as exc:

            return _format_grpc_error(exc)

class OpticalSpectrumReservationConsume(_Resource):

    def post(self, context_uuid : str, reservation_uuid : str):

            raise BadRequest('Mismatching context_uuid')

        if reservation_uuid != reservation['reservation_id']['reservation_uuid']['uuid']:

            raise BadRequest('Mismatching reservation_uuid')

        try:

            return format_grpc_to_json(self.context_client.ConsumeOpticalSpectrumReservation(

                grpc_optical_spectrum_reservation(reservation)

            ))

        except grpc.RpcError as exc:

            return _format_grpc_error(exc)

class OpticalSpectrumReservationRelease(_Resource):

    def post(self, context_uuid : str, reservation_uuid : str):

        try:

            return format_grpc_to_json(self.context_client.ReleaseOpticalSpectrumReservation(

                grpc_optical_spectrum_reservation_id(context_uuid, reservation_uuid)

            ))

        except grpc.RpcError as exc:

            return _format_grpc_error(exc)

class ConnectionIds(_Resource):

    def get(self, context_uuid : str, service_uuid : str):

        reply, reservation_uuid, status='OPTICALSPECTRUMRESERVATIONSTATUS_RESERVED'

    )

    overlap_uuid = 'reservation-nbi-overlap'

    overlap_reservation = _optical_spectrum_reservation(DEFAULT_CONTEXT_NAME, overlap_uuid)

    overlap_reservation['n_start'] = 20

    overlap_reservation['n_end'] = 30

    overlap_reservation['required_slots'] = 11

    overlap_reservation['correlation_id'] = overlap_uuid

    reply = do_rest_post_request(

        '/tfs-api/context/{:s}/optical_spectrum_reservations'.format(context_uuid),

        overlap_reservation,

        expected_status_codes={409}

    )

    assert reply['error'] == 'ALREADY_EXISTS'

    assert 'overlapping spectrum reservation' in reply['message']

    occupied_uuid = 'reservation-nbi-occupied'

    occupied_reservation = _optical_spectrum_reservation(DEFAULT_CONTEXT_NAME, occupied_uuid)

    occupied_reservation['n_start'] = 0

    occupied_reservation['n_end'] = 0

    occupied_reservation['required_slots'] = 1

    occupied_reservation['correlation_id'] = occupied_uuid

    reply = do_rest_post_request(

        '/tfs-api/context/{:s}/optical_spectrum_reservations'.format(context_uuid),

        occupied_reservation,

        expected_status_codes={409}

    )

    assert reply['error'] == 'ALREADY_EXISTS'

    assert 'is not available on optical link' in reply['message']

    reservation['owner_id'] = 'nbi-unit-test-updated'

    reply = do_rest_put_request(

        '/tfs-api/context/{:s}/optical_spectrum_reservation/{:s}'.format(

# Copyright 2022-2026 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 grpc

from flask import Flask

from nbi.service.tfs_api.Resources import _format_grpc_error

class _FakeRpcError(grpc.RpcError):

    def __init__(self, code, details):

        super().__init__()

        self._code = code

        self._details = details

    def code(self):

        return self._code

    def details(self):

        return self._details

def _mapped_error(code, details='mapped error'):

    app = Flask(__name__)

    with app.app_context():

        response, status = _format_grpc_error(_FakeRpcError(code, details))

        return response.get_json(), status

def test_grpc_already_exists_maps_to_http_conflict():

    reply, status = _mapped_error(grpc.StatusCode.ALREADY_EXISTS, 'overlapping spectrum reservation')

    assert status == 409

    assert reply['error'] == 'ALREADY_EXISTS'

    assert reply['message'] == 'overlapping spectrum reservation'

def test_grpc_invalid_argument_maps_to_http_bad_request():

    reply, status = _mapped_error(grpc.StatusCode.INVALID_ARGUMENT, 'invalid optical link')

    assert status == 400

    assert reply['error'] == 'INVALID_ARGUMENT'

    assert reply['message'] == 'invalid optical link'

def test_grpc_not_found_maps_to_http_not_found():

    reply, status = _mapped_error(grpc.StatusCode.NOT_FOUND, 'reservation not found')

    assert status == 404

    assert reply['error'] == 'NOT_FOUND'

    assert reply['message'] == 'reservation not found'

# Spectrum Negotiation Integration Tests

This folder contains executable integration-test seeds for the optical

spectrum reservation workflow used by the prototype.

These tests are intentionally not wired into GitLab CI yet. They require a

running TeraFlowSDN deployment with:

- TFS-API reachable over HTTP.

- At least one context and topology loaded.

- Optical links exposed through `/tfs-api/optical_links`.

- Optical spectrum reservation endpoints exposed through `/tfs-api/context/<context_uuid>/...`.

## Live Reservation Validation

Run against a local TFS deployment:

```bash

python3 src/tests/spectrum_negotiation/live_reservation_validation.py

```

Run against a specific testbed node:

```bash

python3 src/tests/spectrum_negotiation/live_reservation_validation.py \

  --base-url http://172.16.0.101/tfs-api

```

The script validates:

- TFS-API context and topology discovery.

- Optical link inventory discovery.

- Creation of a temporary optical spectrum reservation.

- Retrieval of the reservation in `RESERVED` state.

- Rejection of an overlapping reservation.

- Rejection of a reservation over an unavailable slot.

- Release of the temporary reservation.

- Final cleanup verification that no active `codex-live-test` reservation remains.

Expected rejection status for overlapping or occupied-slot requests is HTTP

`409 Conflict`, mapped from gRPC `ALREADY_EXISTS`.

#!/usr/bin/env python3

# Copyright 2022-2026 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 argparse

import json

import time

import urllib.error

import urllib.parse

import urllib.request

DEFAULT_BASE_URL = 'http://127.0.0.1/tfs-api'

OWNER_ID = 'codex-live-test'

def http_request(base_url, method, path, payload=None, timeout=15):

    data = None

    headers = {}

    if payload is not None:

        data = json.dumps(payload).encode('utf-8')

        headers['Content-Type'] = 'application/json'

    request = urllib.request.Request(base_url + path, data=data, headers=headers, method=method)

    try:

        with urllib.request.urlopen(request, timeout=timeout) as response:

            body = response.read().decode('utf-8')

            return response.status, json.loads(body) if body else None

    except urllib.error.HTTPError as exc:

        body = exc.read().decode('utf-8')

        try:

            payload = json.loads(body) if body else None

        except json.JSONDecodeError:

            payload = body

        return exc.code, payload

def first_context_and_topology(base_url):

    status, contexts = http_request(base_url, 'GET', '/contexts')

    assert status == 200, (status, contexts)

    assert contexts.get('contexts'), contexts

    context = contexts['contexts'][0]

    assert context.get('topology_ids'), context

    return (

        context['context_id']['context_uuid']['uuid'],

        context['topology_ids'][0]['topology_uuid']['uuid'],

    )

def optical_link_slots(link):

    return link.get('optical_details', {}).get('c_slots', {})

def optical_link_uuid(link):

    return link['link_id']['link_uuid']['uuid']

def select_optical_link(base_url, n_start, n_end):

    status, optical_links = http_request(base_url, 'GET', '/optical_links')

    assert status == 200, (status, optical_links)

    links = optical_links.get('optical_links', [])

    assert links, 'No optical links exposed by TFS-API'

    candidate = None

    unavailable = None

    for link in links:

        link_uuid = optical_link_uuid(link)

        c_slots = optical_link_slots(link)

        if candidate is None and all(c_slots.get(str(slot)) == 1 for slot in range(n_start, n_end + 1)):

            candidate = (link_uuid, c_slots)

        if unavailable is None:

            for slot, state in c_slots.items():

                if state != 1:

                    unavailable = (link_uuid, int(slot), state)

                    break

        if candidate and unavailable:

            break

    assert candidate is not None, 'No optical link has the requested validation slots available'

    return len(links), candidate[0], unavailable

def reservation_payload(context_uuid, topology_uuid, reservation_uuid, link_uuid, n_start, n_end):

    return {

        'reservation_id': {

            'context_id': {'context_uuid': {'uuid': context_uuid}},

            'reservation_uuid': {'uuid': reservation_uuid},

        },

        'topology_id': {

            'context_id': {'context_uuid': {'uuid': context_uuid}},

            'topology_uuid': {'uuid': topology_uuid},

        },

        'optical_link_ids': [{'link_uuid': {'uuid': link_uuid}}],

        'band': 'c_slots',

        'n_start': n_start,

        'n_end': n_end,

        'required_slots': n_end - n_start + 1,

        'owner_id': OWNER_ID,

        'correlation_id': reservation_uuid,

    }

def active_test_reservations(reply):

    reservations = reply.get('reservations', [])

    return [

        reservation for reservation in reservations

        if reservation.get('owner_id') == OWNER_ID

        and reservation.get('status') not in ('OPTICALSPECTRUMRESERVATIONSTATUS_RELEASED', 'RELEASED')

    ]

def main():

    parser = argparse.ArgumentParser(description='Validate live optical spectrum reservation behavior through TFS-API.')

    parser.add_argument('--base-url', default=DEFAULT_BASE_URL, help='TFS-API base URL, default: %(default)s')

    parser.add_argument('--n-start', type=int, default=10, help='First validation slot, default: %(default)s')

    parser.add_argument('--n-end', type=int, default=25, help='Last validation slot, default: %(default)s')

    args = parser.parse_args()

    context_uuid, topology_uuid = first_context_and_topology(args.base_url)

    link_count, link_uuid, unavailable = select_optical_link(args.base_url, args.n_start, args.n_end)

    list_path = '/context/{:s}/optical_spectrum_reservations'.format(urllib.parse.quote(context_uuid))

    status, before = http_request(args.base_url, 'GET', list_path)

    assert status == 200, (status, before)

    reservation_uuid = 'live-reservation-{:d}'.format(int(time.time()))

    reservation = reservation_payload(

        context_uuid, topology_uuid, reservation_uuid, link_uuid, args.n_start, args.n_end

    )

    created = False

    try:

        status, create_reply = http_request(args.base_url, 'POST', list_path, reservation)

        assert status == 200, (status, create_reply)

        created = True

        get_path = '/context/{:s}/optical_spectrum_reservation/{:s}'.format(

            urllib.parse.quote(context_uuid), urllib.parse.quote(reservation_uuid)

        )

        status, got = http_request(args.base_url, 'GET', get_path)

        assert status == 200, (status, got)

        assert got.get('status') in ('OPTICALSPECTRUMRESERVATIONSTATUS_RESERVED', 'RESERVED'), got

        overlap_uuid = reservation_uuid + '-overlap'

        overlap = reservation_payload(

            context_uuid, topology_uuid, overlap_uuid, link_uuid, args.n_start + 10, args.n_end + 5

        )

        status, overlap_reply = http_request(args.base_url, 'POST', list_path, overlap)

        assert status == 409, (status, overlap_reply)

        if unavailable is not None:

            _, occupied_slot, _ = unavailable

            occupied_uuid = reservation_uuid + '-occupied'

            occupied = reservation_payload(

                context_uuid, topology_uuid, occupied_uuid, link_uuid, occupied_slot, occupied_slot

            )

            status, occupied_reply = http_request(args.base_url, 'POST', list_path, occupied)

            assert status == 409, (status, occupied_reply)

    finally:

        if created:

            release_path = '/context/{:s}/optical_spectrum_reservation/{:s}/release'.format(

                urllib.parse.quote(context_uuid), urllib.parse.quote(reservation_uuid)

            )

            status, release_reply = http_request(args.base_url, 'POST', release_path)

            assert status == 200, (status, release_reply)

            get_path = '/context/{:s}/optical_spectrum_reservation/{:s}'.format(

                urllib.parse.quote(context_uuid), urllib.parse.quote(reservation_uuid)

            )

            status, got = http_request(args.base_url, 'GET', get_path)

            assert status == 200, (status, got)

            assert got.get('status') in ('OPTICALSPECTRUMRESERVATIONSTATUS_RELEASED', 'RELEASED'), got

    status, after = http_request(args.base_url, 'GET', list_path)

    assert status == 200, (status, after)

    assert not active_test_reservations(after), after

    print('LIVE_RESERVATION_VALIDATION_OK')

    print('context_uuid={:s}'.format(context_uuid))

    print('topology_uuid={:s}'.format(topology_uuid))

    print('optical_links={:d}'.format(link_count))

    print('selected_link={:s}'.format(link_uuid))

    print('reserved_slots={:d}-{:d}'.format(args.n_start, args.n_end))

if __name__ == '__main__':

    main()