SpineLeaf:

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

import logging

import os

from common.proto.context_pb2 import ConfigRule

from common.proto.spineleaf_pb2 import DeployDeviceConfigRequest

from common.tools.context_queries.Device import get_device

from common.tools.object_factory.ConfigRule import json_config_rule_set

from context.client.ContextClient import ContextClient

from device.client.DeviceClient import DeviceClient

from spine_leaf.client.SpineLeafClient import SpineLeafClient

logging.basicConfig(level=logging.INFO)

LOGGER = logging.getLogger("apply_spineleaf_config_test")

CONTEXTSERVICE_HOST = os.getenv("CONTEXTSERVICE_SERVICE_HOST", "10.152.183.46")

CONTEXTSERVICE_PORT = int(os.getenv("CONTEXTSERVICE_SERVICE_PORT_GRPC", "1010"))

DEVICESERVICE_HOST = os.getenv("DEVICESERVICE_SERVICE_HOST", "10.152.183.102")

DEVICESERVICE_PORT = int(os.getenv("DEVICESERVICE_SERVICE_PORT_GRPC", "2020"))

SPINE_LEAFSERVICE_HOST = os.getenv("SPINE_LEAFSERVICE_SERVICE_HOST", "10.152.183.147")

SPINE_LEAFSERVICE_PORT = int(os.getenv("SPINE_LEAFSERVICE_SERVICE_PORT_GRPC", "10095"))

def parse_spineleaf_rule(entry):

    if isinstance(entry, (list, tuple)) and len(entry) == 2:

        resource_key, resource_value = entry

    elif isinstance(entry, dict):

        resource_key = entry.get("resource_key") or entry.get("key")

        resource_value = entry.get("resource_value") or entry.get("value")

    else:

        return None

    if isinstance(resource_value, str):

        try:

            resource_value = json.loads(resource_value)

        except Exception:

            pass

    return resource_key, resource_value

class ApplySpineLeafConfig:

    def __init__(self) -> None:

        self.spine_leaf_client = SpineLeafClient(

            host=SPINE_LEAFSERVICE_HOST,

            port=SPINE_LEAFSERVICE_PORT,

        )

    def run(self):

        context_client = ContextClient(

            host=CONTEXTSERVICE_HOST,

            port=CONTEXTSERVICE_PORT,

        )

        device_client = DeviceClient(

            host=DEVICESERVICE_HOST,

            port=DEVICESERVICE_PORT,

        )

        try:

            request = DeployDeviceConfigRequest(

                fabric_id="354a08b7-3730-5dd6-9472-47554115818b",

                device_uuid="e162b163-6969-512a-8ee9-cce9f0687c64",

            )

            response = self.spine_leaf_client.DeployDeviceConfig(request)

            LOGGER.info("Raw response: %s", response)

            payload = json.loads(

                getattr(response, "message", "")

                or getattr(response, "payload", "{}")

            )

            LOGGER.info("Parsed payload: %s", payload)

            config_rules = []

            for entry in payload.get("driver_config_rules", []):

                parsed = parse_spineleaf_rule(entry)

                if parsed is None:

                    continue

                resource_key, resource_value = parsed

                config_rules.append(

                    ConfigRule(**json_config_rule_set(resource_key, resource_value))

                )

            LOGGER.info("Normalized config rules: %s", config_rules)

            device = get_device(

                context_client,

                request.device_uuid,

                rw_copy=True,

                include_endpoints=False,

                include_config_rules=False,

                include_components=False,

            )

            if device is None:

                LOGGER.error("Device not found in Context: %s", request.device_uuid)

                return

            device.device_config.config_rules.extend(config_rules)

            apply_config = os.getenv('APPLY_CONFIG', 'false').lower() in ('1', 'true', 'yes')

            if apply_config:

                result = device_client.ConfigureDevice(device)

                LOGGER.info("ConfigureDevice result: %s", result)

            else:

                LOGGER.info("Dry-run: not applying configuration to device %s. Set APPLY_CONFIG=1 to apply.", request.device_uuid)

        finally:

            device_client.close()

            context_client.close()

if __name__ == "__main__":

    ApplySpineLeafConfig().run()

 No newline at end of file