[unconfig] Adjust device discovery

uniconfig/python/CHANGELOG.md

@@ -47,4 +47,7 @@
 - Update Inventory properties and bump Uniconfig API.
 
 # 2.3.3
-- Fix Device Discovery RPC based on new OpenAPI model.
+- Fix Device Discovery RPC based on new OpenAPI model.
+
+# 2.3.4
+- Device Discovery RPC workaround caused by choice-nodes in UniConfig Yang models.

uniconfig/python/frinx_worker/uniconfig/connection_manager.py

@@ -17,15 +17,15 @@
 
 from . import class_to_json
 from . import handle_response
+from . import snake_to_kebab_case
 
 
 class ConnectionManager(ServiceWorkersImpl):
     class InstallNode(WorkerImpl):
-        from frinx_api.uniconfig.connection.manager.installnode import Cli
-        from frinx_api.uniconfig.connection.manager.installnode import Gnmi
-        from frinx_api.uniconfig.connection.manager.installnode import Input
-        from frinx_api.uniconfig.connection.manager.installnode import Netconf
         from frinx_api.uniconfig.rest_api import InstallNode as UniconfigApi
+        # from frinx_api.uniconfig.connection.manager.installnode import Cli
+        # from frinx_api.uniconfig.connection.manager.installnode import Gnmi
+        # from frinx_api.uniconfig.connection.manager.installnode import Netconf
 
         class ExecutionProperties(TaskExecutionProperties):
             exclude_empty_inputs: bool = True
@@ -48,31 +48,60 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
             if self.UniconfigApi.request is None:
                 raise Exception(f"Failed to create request {self.UniconfigApi.request}")
 
+            import json
             response = requests.request(
                 url=worker_input.uniconfig_url_base + self.UniconfigApi.uri,
                 method=self.UniconfigApi.method,
-                data=class_to_json(
-                    self.UniconfigApi.request(
-                        input=self.Input(
-                            node_id=worker_input.node_id,
-                            cli=self.Cli(**worker_input.install_params)
-                            if worker_input.connection_type == "cli"
-                            else None,
-                            netconf=self.Netconf(**worker_input.install_params)
-                            if worker_input.connection_type == "netconf"
-                            else None,
-                            gnmi=self.Gnmi(**worker_input.install_params)
-                            if worker_input.connection_type == "gnmi"
-                            else None,
-                        ),
-                    ),
-                ),
+                # FIXME prepare input with credentials (TEMPORARY SOLUTION)
+                data=json.dumps(snake_to_kebab_case(self._prepare_input(worker_input))),
+                # data=class_to_json(
+                #     self.UniconfigApi.request(
+                #         input=self.Input(
+                #             node_id=worker_input.node_id,
+                #             cli=self.Cli(**worker_input.install_params)
+                #             if worker_input.connection_type == "cli"
+                #             else None,
+                #             netconf=self.Netconf(**worker_input.install_params)
+                #             if worker_input.connection_type == "netconf"
+                #             else None,
+                #             gnmi=self.Gnmi(**worker_input.install_params)
+                #             if worker_input.connection_type == "gnmi"
+                #             else None,
+                #         ),
+                #     ),
+                # ),
                 headers=dict(UNICONFIG_HEADERS),
                 params=UNICONFIG_REQUEST_PARAMS,
             )
 
             return handle_response(response, self.WorkerOutput)
 
+        def _prepare_input(self, worker_input: WorkerInput) -> DictAny:
+            if worker_input.connection_type == "cli":
+                return {
+                    "input": {
+                        "node_id": worker_input.node_id,
+                        "cli": worker_input.install_params
+                    }
+                }
+            elif worker_input.connection_type == "netconf":
+                return {
+                    "input": {
+                        "node_id": worker_input.node_id,
+                        "netconf": worker_input.install_params
+                    }
+                }
+            elif worker_input.connection_type == "gnmi":
+                return {
+                    "input": {
+                        "node_id": worker_input.node_id,
+                        "gnmi": worker_input.install_params
+                    }
+                }
+            else:
+                raise ValueError(f"Unknown connection type '{worker_input.connection_type}'")
+
+
     class UninstallNode(WorkerImpl):
         from frinx_api.uniconfig.connection.manager import ConnectionType
         from frinx_api.uniconfig.connection.manager.uninstallnode import Input

uniconfig/python/frinx_worker/uniconfig/device_discovery.py

@@ -1,5 +1,6 @@
 from ipaddress import IPv4Address
 from ipaddress import IPv6Address
+from typing import Any
 
 import pydantic
 import requests
@@ -36,6 +37,31 @@
 from .util import parse_ranges
 
 
+def _unwrap_data(discovery_input: Input) -> dict[str, Any]:
+    tcp_port: list[dict[str, Any]] = []
+    udp_port: list[dict[str, Any]] = []
+    address: list[dict[str, Any]] = []
+    if discovery_input.tcp_port is not None:
+        for tcp_port_item in discovery_input.tcp_port:
+            if tcp_port_item.type_of_port is not None:
+                tcp_port.append(tcp_port_item.type_of_port.model_dump())
+    if discovery_input.udp_port is not None:
+        for udp_port_item in discovery_input.udp_port:
+            if udp_port_item.type_of_port is not None:
+                udp_port.append(udp_port_item.type_of_port.model_dump())
+    if discovery_input.address is not None:
+        for address_item in discovery_input.address:
+            if address_item.type_of_address is not None:
+                address.append(address_item.type_of_address.model_dump())
+    return {
+        "input": {
+            "address": None if not address else address,
+            "tcp_port": None if not tcp_port else tcp_port,
+            "udp_port": None if not udp_port else udp_port
+        }
+    }
+
+
 class DeviceDiscoveryWorkers(ServiceWorkersImpl):
     class DeviceDiscoveryWorker(WorkerImpl):
         class ExecutionProperties(TaskExecutionProperties):
@@ -129,8 +155,9 @@ def execute(self, worker_input: WorkerInput) -> TaskResult[WorkerOutput]:
                 headers=dict(UNICONFIG_HEADERS),
                 params=UNICONFIG_REQUEST_PARAMS,
                 method=Discover.method,
+                # temporary workaround until UC adds possibility to accept choice nodes
                 data=class_to_json(
-                    Discover.request(input=template),
+                    _unwrap_data(template),
                 ),
             )
 

uniconfig/python/pyproject.toml

@@ -21,7 +21,7 @@ packages = [{ include = "frinx_worker" }]
 name = "frinx-uniconfig-worker"
 description = "Conductor worker for Frinx Uniconfig"
 authors = ["Jozef Volak <[email protected]>"]
-version = "2.3.3"
+version = "2.3.4"
 readme = ["README.md", "CHANGELOG.md", "RELEASE.md"]
 keywords = ["frinx-machine", "uniconfig", "worker"]
 license = "Apache 2.0"