Adds duplicated code as function for inventory helper.

nornir/core/inventory.py

@@ -662,3 +662,52 @@ def dict(self) -> Dict[str, Any]:
             "groups": {n: g.dict() for n, g in self.groups.items()},
             "defaults": self.defaults.dict(),
         }
+
+
+def _get_connection_options_from_dict(
+    data: Dict[str, Any]
+) -> Dict[str, ConnectionOptions]:
+    """Gather connection options from a dictionary.
+
+    Gathered from the work of Nornir NetBox Inventory (2021-10-13).
+    https://github.com/wvandeun/nornir_netbox/blob/develop/nornir_netbox/plugins/inventory/netbox.py#L25-L36
+
+    Args:
+        data (Dict[str, Any]): Dictionary
+
+    Returns:
+        Dict[str, ConnectionOptions]: Dictionary of connection options
+    """
+    cp = {}
+    for cn, c in data.items():
+        cp[cn] = ConnectionOptions(
+            hostname=c.get("hostname"),
+            port=c.get("port"),
+            username=c.get("username"),
+            password=c.get("password"),
+            platform=c.get("platform"),
+            extras=c.get("extras"),
+        )
+    return cp
+
+
+def get_defaults_from_dict(data: Dict[str, Any]) -> Defaults:
+    """Get defaults from a data dictionary.
+
+    Gathered from work done in Nornir Ansible Inventory and Nornir NetBox Inventory files.
+    https://github.com/wvandeun/nornir_netbox/blob/develop/nornir_netbox/plugins/inventory/netbox.py#L39-L48
+
+    Args:
+        data (dict): Defaults
+    """
+    return Defaults(
+        hostname=data.get("hostname"),
+        port=data.get("port"),
+        username=data.get("username"),
+        password=data.get("password"),
+        platform=data.get("platform"),
+        data=data.get("data"),
+        connection_options=_get_connection_options_from_dict(
+            data.get("connection_options", {})
+        ),
+    )

tests/core/test_inventory.py

@@ -666,3 +666,37 @@ def test_remove_group_from_host(self):
 
         with pytest.raises(ValueError):
             h1.groups.remove(g3)
+
+    def test_inventory_load_defaults_no_connections(self):
+        # Set a test data to send in for a dictionary, with the results coming back
+        test_data = {
+            "port": "22",
+            "password": "Nornir123",
+            "platform": "fake",
+        }
+        test_result = inventory.get_defaults_from_dict(test_data)
+        assert isinstance(test_result, inventory.Defaults)
+        assert test_result.hostname is None
+        assert test_result.password == "Nornir123"
+        assert test_result.platform == "fake"
+
+    def test_inventory_load_defaults_with_connections(self):
+        # Set a test data to send in for a dictionary, with the results coming back
+        test_data = {
+            "port": "22",
+            "password": "Nornir123",
+            "platform": "fake",
+            "connection_options": {"netmiko": {"extras": {"use_keys": True}}},
+        }
+        test_result = inventory.get_defaults_from_dict(test_data)
+        assert isinstance(test_result, inventory.Defaults)
+        assert test_result.hostname is None
+        assert test_result.password == "Nornir123"
+        assert test_result.platform == "fake"
+        assert isinstance(
+            test_result.connection_options.get("netmiko"), inventory.ConnectionOptions
+        )
+        assert isinstance(
+            test_result.connection_options.get("netmiko").extras.get("use_keys"), bool
+        )
+        assert test_result.connection_options.get("netmiko").extras.get("use_keys")