diff --git a/nornir/core/inventory.py b/nornir/core/inventory.py index 0d09b15b..9d483592 100644 --- a/nornir/core/inventory.py +++ b/nornir/core/inventory.py @@ -655,3 +655,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", {}) + ), + ) diff --git a/tests/core/test_inventory.py b/tests/core/test_inventory.py index 01d0bae8..a45dc43c 100644 --- a/tests/core/test_inventory.py +++ b/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")