Code refactor 2

[tool.ruff]

line-length = 120

target-version = "py312"

[tool.ruff.lint]

select = ["E", "F", "W", "I", "UP", "B", "C4", "SIM"]

ignore = [

    "E501",    # Line too long (managed by formatter and docstrings)

    "BLE001",  # Catching Exception is intentional for HTTP 500 API boundary handlers

    "SIM108",  # Use ternary operator

    "SIM102",  # Nested if statements

    "SIM105",  # Use contextlib.suppress

    "SIM117",  # Multiple with statements

]

[tool.pytest.ini_options]

testpaths = ["src/tests"]

pythonpath = ["."]

addopts = "-v --tb=short"

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

# This file is an original contribution from Telefonica Innovación Digital S.L.

"""API package for Network Slice Controller."""

from __future__ import annotations

from src.api.base_handler import BaseSliceHandler

from src.api.e2e_handler import E2EHandler

from src.api.ixia_handler import IxiaHandler

from src.api.main import Api

from src.api.restconf_handler import RestconfHandler

from src.api.tfs_handler import TfsHandler

__all__ = [

    "Api",

    "BaseSliceHandler",

    "E2EHandler",

    "IxiaHandler",

    "RestconfHandler",

    "TfsHandler",

]

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

# This file is an original contribution from Telefonica Innovación Digital S.L.

"""Base handler for transport network slice flow operations."""

from __future__ import annotations

import logging

import sys

from typing import Any

from flask import current_app

from src.database.db import (

    delete_all_data as _db_delete_all_data,

)

from src.database.db import (

    delete_data as _db_delete_data,

)

from src.database.db import (

    get_all_data as _db_get_all_data,

)

from src.database.db import (

    get_data as _db_get_data,

)

from src.realizer.tfs.helpers.tfs_connector import tfs_connector as _real_tfs_connector

from src.realizer.tfs.service_types.tfs_l2vpn import tfs_l2vpn_delete as _real_tfs_l2vpn_delete

from src.utils.safe_get import safe_get

from src.utils.send_response import send_response

logger = logging.getLogger(__name__)

def _dep(name: str, fallback: Any) -> Any:

    """Resolve dependency from src.api.main if patched, otherwise fallback."""

    main_mod = sys.modules.get("src.api.main")

    if main_mod is not None and hasattr(main_mod, name):

        return getattr(main_mod, name)

    return fallback

def _extract_slice_type(slice_dict: dict[str, Any]) -> str:

    """Extract slice type tag from slice intent with fallback to L2."""

    slice_type = safe_get(

        slice_dict,

        [

            "intent",

            "ietf-network-slice-service:network-slice-services",

            "slice-service",

            0,

            "service-tags",

            "tag-type",

            0,

            "tag-type-value",

            0,

        ],

    )

    if not slice_type:

        logger.warning("Slice type not found in slice intent. Defaulting to L2")

        return "L2"

    return str(slice_type)

def _delete_slice_from_tfs(slice_dict: dict[str, Any], slice_id: str) -> None:

    """Delete a slice in TeraFlowSDN via NBI connector."""

    slice_type = _extract_slice_type(slice_dict)

    connector = _dep("tfs_connector", _real_tfs_connector)()

    connector.nbi_delete(current_app.config["TFS_IP"], slice_type, slice_id)

class BaseSliceHandler:

    """Base handler providing CRUD flow operations for transport network slices."""

    def __init__(self, slice_service: Any) -> None:

        """Initialize handler with underlying slice service (e.g. NSController).

        Args:

            slice_service: Service instance managing controller-specific business logic.

        """

        self.slice_service = slice_service

    def add_flow(self, intent: dict[str, Any]) -> tuple[dict[str, Any], int]:

        """Create a new transport network slice."""

        try:

            result = self.slice_service.nsc(intent)

            if not result:

                return send_response(False, code=404, message="No intents found")

            if isinstance(result, tuple):

                return result

            logger.info("Slice created successfully")

            return send_response(True, code=201, data=result)

        except RuntimeError as exc:

            return send_response(False, code=200, message=str(exc))

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def get_flows(self, slice_id: str | None = None) -> tuple[dict[str, Any] | list[dict[str, Any]], int]:

        """Retrieve transport network slice information."""

        try:

            get_all_data_fn = _dep("get_all_data", _db_get_all_data)

            content = get_all_data_fn()

            if slice_id:

                for slice_item in content:

                    if slice_item.get("slice_id") == slice_id:

                        return slice_item, 200

                raise ValueError("Transport network slices not found")

            if not content:

                raise ValueError("Transport network slices not found")

            filtered = [s for s in content if s.get("controller") == self.slice_service.controller_type]

            return filtered, 200

        except ValueError as exc:

            return send_response(False, code=404, message=str(exc))

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def modify_flow(self, slice_id: str, intent: dict[str, Any]) -> tuple[dict[str, Any], int]:

        """Modify an existing transport network slice."""

        try:

            result = self.slice_service.nsc(intent, slice_id)

            if isinstance(result, tuple):

                return result

            if not result:

                return send_response(False, code=404, message="Slice not found")

            logger.info("Slice %s modified successfully", slice_id)

            return send_response(

                True,

                code=200,

                message="Slice modified successfully",

                data=result,

            )

        except ValueError as exc:

            return send_response(False, code=404, message=str(exc))

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def delete_flows(self, slice_id: str | None = None) -> tuple[dict[str, Any], int]:

        """Delete transport network slice(s)."""

        try:

            if slice_id:

                return self._delete_single_flow(slice_id)

            return self._delete_all_flows()

        except ValueError as exc:

            return send_response(False, code=404, message=str(exc))

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def _delete_single_flow(self, slice_id: str) -> tuple[dict[str, Any], int]:

        """Delete a single slice by slice_id."""

        get_data_fn = _dep("get_data", _db_get_data)

        slice_item = get_data_fn(slice_id)

        if not slice_item or slice_item.get("controller") != self.slice_service.controller_type:

            raise ValueError("Transport network slice not found")

        if not current_app.config["DUMMY_MODE"] and self.slice_service.controller_type == "TFS":

            _delete_slice_from_tfs(slice_item, slice_id)

        delete_data_fn = _dep("delete_data", _db_delete_data)

        delete_data_fn(slice_id)

        logger.info("Slice %s removed successfully", slice_id)

        return {}, 204

    def _delete_all_flows(self) -> tuple[dict[str, Any], int]:

        """Delete all slices belonging to the current controller."""

        get_all_data_fn = _dep("get_all_data", _db_get_all_data)

        if not current_app.config["DUMMY_MODE"] and self.slice_service.controller_type == "TFS":

            content = get_all_data_fn()

            for slice_item in content:

                if slice_item.get("controller") == self.slice_service.controller_type:

                    _delete_slice_from_tfs(slice_item, slice_item.get("slice_id", ""))

            if current_app.config["TFS_L2VPN_SUPPORT"]:

                if hasattr(self.slice_service, "tfs_l2vpn_delete"):

                    self.slice_service.tfs_l2vpn_delete()

                else:

                    l2vpn_del = _dep("tfs_l2vpn_delete", _real_tfs_l2vpn_delete)

                    l2vpn_del()

        delete_all_data_fn = _dep("delete_all_data", _db_delete_all_data)

        delete_all_data_fn()

        logger.info("All slices removed successfully")

        return {}, 204

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

# This file is an original contribution from Telefonica Innovación Digital S.L.

"""E2E Orchestrator slice and TAPI alert API service handler."""

from __future__ import annotations

import json

import logging

import sys

from pathlib import Path

from typing import Any

from src.api.base_handler import BaseSliceHandler

from src.database import alert_db as _real_alert_db

from src.database.db import (

    get_all_data as _db_get_all_data,

)

from src.database.db import (

    get_data as _db_get_data,

)

from src.database.db import (

    get_slice_id_by_subscription as _db_get_slice_id_by_sub,

)

from src.database.service_db import get_data as _db_get_service_db_data

from src.utils.send_response import send_response

logger = logging.getLogger(__name__)

FALLBACK_INTENT_PATH = Path("/home/llmserver/tfs-nsc/intent.json")

def _dep(name: str, fallback: Any) -> Any:

    """Resolve dependency from src.api.main if patched, otherwise fallback."""

    main_mod = sys.modules.get("src.api.main")

    if main_mod is not None and hasattr(main_mod, name):

        return getattr(main_mod, name)

    return fallback

def _parse_alert_notification(alert_data: dict[str, Any]) -> tuple[str | None, str | None, str | None]:

    """Extract alert_id, subscription_id, and service_id from TAPI alert payload."""

    context = alert_data.get("tapi-notification:notification-context", [])

    if not context or not isinstance(context, list):

        return None, None, None

    notification = context[0].get("tapi-notification:notification", {})

    alert_id = notification.get("uuid")

    subscription_id = alert_id

    additional_info = notification.get("additional-info", {})

    service_id = additional_info.get("service-id")

    return alert_id, subscription_id, service_id

def _find_slice_for_alert(subscription_id: str | None, service_id: str | None) -> dict[str, Any] | None:

    """Resolve slice information for an alert by subscription, service_id, database, or fallback file."""

    slice_id = _resolve_slice_id(subscription_id, service_id)

    slice_info = _lookup_slice_info(slice_id)

    if slice_info is not None:

        return slice_info

    return _load_fallback_intent(slice_id)

def _resolve_slice_id(subscription_id: str | None, service_id: str | None) -> str | None:

    """Resolve slice ID from subscription mapping or service database."""

    if subscription_id:

        try:

            get_slice_fn = _dep("get_slice_id_by_subscription", _db_get_slice_id_by_sub)

            mapped_slice_id = get_slice_fn(subscription_id)

            if mapped_slice_id:

                logger.info("Found slice_id %s mapped to subscription_id %s", mapped_slice_id, subscription_id)

                return mapped_slice_id

        except Exception as exc:

            logger.info("Subscription mapping lookup failed: %s", exc)

    if service_id:

        try:

            get_svc_fn = _dep("get_service_db_data", _db_get_service_db_data)

            service_info = get_svc_fn(service_id)

            resolved_slice_id = service_info.get("slice_id")

            logger.info("Found slice_id %s in service_db for service_id %s", resolved_slice_id, service_id)

            return resolved_slice_id

        except Exception as exc:

            logger.info("service_db lookup failed: %s", exc)

            return service_id

    return None

def _lookup_slice_info(slice_id: str | None) -> dict[str, Any] | None:

    """Lookup slice info in the database by ID or fall back to the first available slice."""

    if slice_id:

        try:

            get_data_fn = _dep("get_data", _db_get_data)

            slice_info = get_data_fn(slice_id)

            if slice_info:

                logger.info("Found slice_info in db by slice_id %s", slice_id)

                return slice_info

        except Exception as exc:

            logger.info("db lookup by slice_id %s failed: %s", slice_id, exc)

    try:

        get_all_data_fn = _dep("get_all_data", _db_get_all_data)

        slices = get_all_data_fn()

        logger.info("Slices in db: %s", [s.get("slice_id") for s in slices])

        for s in slices:

            if s.get("slice_id") == slice_id:

                return s

        if slices:

            first_slice = slices[0]

            logger.info("Defaulted to first slice from db: %s", first_slice.get("slice_id"))

            return first_slice

    except Exception as exc:

        logger.info("db get_all_data lookup failed: %s", exc)

    return None

def _load_fallback_intent(slice_id: str | None) -> dict[str, Any] | None:

    """Load fallback intent configuration from disk if available."""

    if not FALLBACK_INTENT_PATH.exists():

        return None

    try:

        with FALLBACK_INTENT_PATH.open("r", encoding="utf-8") as file:

            intent_data = json.load(file)

        logger.info("Loaded fallback intent from intent.json")

        return {"slice_id": slice_id or "slice", "intent": intent_data}

    except Exception as exc:

        logger.error("Failed to read fallback intent.json: %s", exc)

        return None

def _swap_p2mp_endpoints(intent: dict[str, Any]) -> tuple[bool, str | None]:

    """Modify P2MP receiver endpoints to alternate paths following an alert."""

    nss = intent.get("ietf-network-slice-service:network-slice-services", {})

    slice_services = nss.get("slice-service", [])

    modified = False

    old_service_id = None

    for service in slice_services:

        sdp_list = service.get("sdps", {}).get("sdp", [])

        sdp_ids = [sdp.get("id") for sdp in sdp_list if sdp.get("id")]

        connection_groups = service.get("connection-groups", {}).get("connection-group", [])

        for cg in connection_groups:

            connectivity_constructs = cg.get("connectivity-construct", [])

            for cc in connectivity_constructs:

                p2mp_sender = cc.get("p2mp-sender-sdp")

                p2mp_receivers = cc.get("p2mp-receiver-sdp", [])

                logger.info(

                    "sdp_ids: %s, p2mp_sender: %s, p2mp_receivers: %s",

                    sdp_ids,

                    p2mp_sender,

                    p2mp_receivers,

                )

                other_endpoints = [

                    sdp_id for sdp_id in sdp_ids if sdp_id != p2mp_sender and sdp_id not in p2mp_receivers

                ]

                if other_endpoints and p2mp_receivers:

                    new_receiver = other_endpoints[0]

                    cc["p2mp-receiver-sdp"] = (

                        [p2mp_receivers[0], new_receiver] if len(p2mp_receivers) >= 2 else [new_receiver]

                    )

                    modified = True

                    logger.info("ORIGEN: %s DESTINO: %s", p2mp_sender, new_receiver)

                else:

                    logger.warning("No alternative receiver endpoints found to swap.")

                if p2mp_receivers:

                    old_service_id = f"{p2mp_sender}_to_{','.join(p2mp_receivers)}"

    return modified, old_service_id

class E2EHandler(BaseSliceHandler):

    """API handler dedicated to E2E Orchestrator slice operations and TAPI alert processing."""

    def receive_alert(self, alert_data: dict[str, Any]) -> tuple[dict[str, Any], int]:

        """Receive and process an incoming TAPI network alert."""

        try:

            logger.info("Alert received: %s", alert_data)

            alert_id, subscription_id, service_id = _parse_alert_notification(alert_data)

            if not alert_id:

                return send_response(False, code=400, message="UUID not found in alert data")

            adb = _dep("alert_db", _real_alert_db)

            adb.save_alert(alert_id, alert_data)

            logger.info(

                "Looking up intent for subscription_id: %s, service_id: %s",

                subscription_id,

                service_id,

            )

            slice_info = _find_slice_for_alert(subscription_id, service_id)

            if slice_info:

                intent = slice_info.get("intent")

                curr_slice_id = slice_info.get("slice_id")

                logger.info("Processing intent for slice %s", curr_slice_id)

                if intent:

                    modified, old_service_id = _swap_p2mp_endpoints(intent)

                    if modified:

                        try:

                            self.slice_service.nsc(intent, curr_slice_id, old_service_id=old_service_id)

                            logger.info("Slice %s updated successfully following alert.", curr_slice_id)

                        except Exception as exc:

                            logger.error("Failed to update slice configuration: %s", exc)

            else:

                logger.warning("No slice intent found to process alert.")

            return send_response(

                True,

                code=201,

                message="Alert processed and saved successfully",

                data=alert_data,

            )

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def get_alerts(self, alert_id: str | None = None) -> tuple[dict[str, Any] | list[dict[str, Any]], int]:

        """Retrieve alert(s)."""

        try:

            adb = _dep("alert_db", _real_alert_db)

            if alert_id:

                try:

                    data = adb.get_alert(alert_id)

                    return data, 200

                except ValueError as exc:

                    return send_response(False, code=404, message=str(exc))

            data = adb.get_all_alerts()

            return data, 200

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def modify_alert(self, alert_id: str, alert_data: dict[str, Any]) -> tuple[dict[str, Any], int]:

        """Modify/update an alert."""

        try:

            adb = _dep("alert_db", _real_alert_db)

            try:

                adb.update_alert(alert_id, alert_data)

                return send_response(

                    True,

                    code=200,

                    message="Alert updated successfully",

                    data=alert_data,

                )

            except ValueError as exc:

                return send_response(False, code=404, message=str(exc))

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

    def delete_alerts(self, alert_id: str | None = None) -> tuple[dict[str, Any], int]:

        """Delete alert(s)."""

        try:

            adb = _dep("alert_db", _real_alert_db)

            if alert_id:

                try:

                    adb.delete_alert(alert_id)

                    return {}, 204

                except ValueError as exc:

                    return send_response(False, code=404, message=str(exc))

            adb.delete_all_alerts()

            return {}, 204

        except Exception as exc:

            return send_response(False, code=500, message=str(exc))

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

#

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