diff --git a/plugins/lookup/lookup.py b/plugins/lookup/lookup.py index 43082bff..17486f46 100644 --- a/plugins/lookup/lookup.py +++ b/plugins/lookup/lookup.py @@ -344,6 +344,8 @@ def run(self, terms, variables=None, **kwargs): api_token = kwargs.get("token") or os.getenv("NAUTOBOT_TOKEN") api_endpoint = kwargs.get("api_endpoint") or os.getenv("NAUTOBOT_URL") + if not api_endpoint or not api_token: + raise AnsibleError("Both api_endpoint and token are required") if kwargs.get("validate_certs") is not None: ssl_verify = kwargs.get("validate_certs") elif os.getenv("NAUTOBOT_VALIDATE_CERTS") is not None: @@ -352,6 +354,8 @@ def run(self, terms, variables=None, **kwargs): ssl_verify = True num_retries = kwargs.get("num_retries", "0") api_filter = kwargs.get("api_filter") + if api_filter: + api_filter = self._templar.do_template(api_filter) raw_return = kwargs.get("raw_data") plugin = kwargs.get("plugin") api_version = kwargs.get("api_version") @@ -360,7 +364,6 @@ def run(self, terms, variables=None, **kwargs): terms = [terms] nautobot = pynautobot.api(api_endpoint, token=api_token if api_token else None, api_version=api_version, verify=ssl_verify, retries=num_retries) - results = [] for term in terms: if plugin: diff --git a/tests/unit/lookup/test_lookup.py b/tests/unit/lookup/test_lookup.py new file mode 100644 index 00000000..4140059c --- /dev/null +++ b/tests/unit/lookup/test_lookup.py @@ -0,0 +1,115 @@ +"""Tests for Nautobot Query Lookup Plugin.""" + +from ansible.errors import AnsibleError +from ansible.template import Templar +from ansible.parsing.dataloader import DataLoader +from ansible.vars.manager import VariableManager +from ansible.inventory.manager import InventoryManager +import pytest +from unittest.mock import patch, MagicMock + + +try: + from plugins.lookup.lookup import LookupModule +except ImportError: + import sys + + sys.path.append("tests") + sys.path.append("plugins/lookup") + + from lookup import LookupModule + + +@pytest.fixture +def lookup(): + """Fixture to create an instance of your lookup plugin.""" + return LookupModule() + + +@patch("plugins.lookup.lookup.pynautobot.api") +def test_basic_run(mock_pynautobot, lookup): + """Test basic functionality of the run method.""" + mock_api = MagicMock() + mock_pynautobot.return_value = mock_api + mock_api.dcim.devices.all.return_value = [{"id": 1, "name": "device1"}] + + terms = ["devices"] + kwargs = { + "token": "fake-token", + "api_endpoint": "https://nautobot.local", + } + result = lookup.run(terms, **kwargs) + + mock_pynautobot.assert_called_once_with("https://nautobot.local", token="fake-token", api_version=None, verify=True, retries="0") + assert result == [{"key": 1, "value": {"id": 1, "name": "device1"}}], "Expected a successful result" + + +def test_invalid_terms(lookup): + """Test when terms is not a list or valid input.""" + with pytest.raises(AnsibleError, match="Unrecognised term"): + with patch("plugins.lookup.lookup.get_endpoint", side_effect=KeyError): + kwargs = { + "token": "fake-token", + "api_endpoint": "https://nautobot.local", + } + lookup.run("invalid.term", **kwargs) + + +@patch("plugins.lookup.lookup.pynautobot.api") +def test_no_token_or_endpoint(mock_pynautobot, lookup): + """Test when neither token nor endpoint is provided.""" + with pytest.raises(AnsibleError): + lookup.run(["devices"], token=None, api_endpoint=None) + + +@patch("plugins.lookup.lookup.pynautobot.api") +def test_run_with_static_filter(mock_pynautobot, lookup): + """Test filters functionality of the run method.""" + mock_api = MagicMock() + mock_pynautobot.return_value = mock_api + mock_api.dcim.devices.filter.return_value = [{"id": 1, "name": "device1"}] + + # Initialize Ansible's Templar with necessary components + loader = DataLoader() + inventory = InventoryManager(loader=loader, sources=[]) + variable_manager = VariableManager(loader=loader, inventory=inventory) + templar = Templar(loader=loader, variables=variable_manager.get_vars()) + lookup._templar = templar + + terms = ["devices"] + kwargs = { + "token": "fake-token", + "api_endpoint": "https://nautobot.local", + "api_filter": "{'name': 'device1'}", + } + result = lookup.run(terms, **kwargs) + + mock_pynautobot.assert_called_once_with("https://nautobot.local", token="fake-token", api_version=None, verify=True, retries="0") + mock_api.dcim.devices.filter.assert_called_once_with(_raw_params=["{'name':", "'device1'}"]) + assert result == [{"key": 1, "value": {"id": 1, "name": "device1"}}], "Expected a successful result" + + +@patch("plugins.lookup.lookup.pynautobot.api") +def test_run_with_dynamic_filter(mock_pynautobot, lookup): + """Test dynamic filters functionality of the run method.""" + mock_api = MagicMock() + mock_pynautobot.return_value = mock_api + mock_api.dcim.devices.filter.return_value = [{"id": 1, "name": "device1"}] + + # Initialize Ansible's Templar with necessary components + loader = DataLoader() + variables = {"device_name": "device1"} + templar = Templar(loader=loader, variables=variables) + lookup._templar = templar + + terms = ["devices"] + kwargs = { + "token": "fake-token", + "api_endpoint": "https://nautobot.local", + "api_filter": "{'name': '{{ device_name }}'}", + } + result = lookup.run(terms, **kwargs) + + mock_pynautobot.assert_called_once_with("https://nautobot.local", token="fake-token", api_version=None, verify=True, retries="0") + mock_api.dcim.devices.filter.assert_called_once_with(_raw_params=["{'name':", "'device1'}"]) + assert result == [{"key": 1, "value": {"id": 1, "name": "device1"}}], "Expected a successful result"