BugFix: Fix connection error handling to Netbox (#80)

Author: FragmentedPacket

.travis.yml

@@ -26,7 +26,7 @@ matrix:
         - cd ..
         - pip install -U pip
         - pip install pytest==4.6.5 pytest-mock pytest-xdist jinja2 PyYAML black==19.10b0
-        - pip install pynetbox==4.0.6 cryptography codecov jmespath ansible
+        - pip install pynetbox cryptography codecov jmespath ansible
 
     - name: "Python 3.6 - Netbox Latest - Ansible Devel"
       python: 3.6
@@ -41,7 +41,7 @@ matrix:
         - cd ..
         - pip install -U pip
         - pip install pytest==4.6.5 pytest-mock pytest-xdist jinja2 PyYAML black==19.10b0
-        - pip install pynetbox==4.0.6 cryptography jmespath
+        - pip install pynetbox cryptography jmespath
         # This is due to ansible-test only being available within devel branch
         - git clone https://github.com/ansible/ansible.git
         - cd ansible
@@ -60,4 +60,5 @@ script:
   - black . --check
   - timeout 300 bash -c 'while [[ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:32768)" != "200" ]]; do echo "waiting for Netbox"; sleep 5; done' || false
   - python tests/integration/netbox-deploy.py
+  - ansible-playbook tests/integration/regression-tests.yml -vvvv
   - ansible-playbook tests/integration/integration-tests.yml -vvvv

README.md

@@ -6,7 +6,7 @@
 ## Requirements
 
 - Netbox 2.5+
-- pynetbox 4.0.6+
+- **pynetbox 4.1.0+**
 - Python 3.6+
 - Ansible 2.9+
 

plugins/module_utils/netbox_circuits.py

@@ -68,10 +68,7 @@ def run(self):
                 data["slug"] = self._to_slug(name)
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

plugins/module_utils/netbox_dcim.py

@@ -88,10 +88,7 @@ def run(self):
             data["color"] = data["color"].lower()
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

plugins/module_utils/netbox_extras.py

@@ -45,10 +45,7 @@ def run(self):
         data["slug"] = self._to_slug(name)
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

plugins/module_utils/netbox_ipam.py

@@ -81,7 +81,7 @@ def _ensure_ip_in_prefix_present_on_netif(
 
     def _get_new_available_ip_address(self, nb_app, data, endpoint_name):
         prefix_query = self._build_query_params("prefix", data)
-        prefix = nb_app.prefixes.get(**prefix_query)
+        prefix = self._nb_endpoint_get(nb_app.prefixes, prefix_query, data["prefix"])
         if not prefix:
             self.result["changed"] = False
             self.result["msg"] = "%s does not exist - please create first" % (
@@ -170,15 +170,13 @@ def run(self):
         object_query_params = self._build_query_params(endpoint_name, data)
         if data.get("prefix") and self.endpoint == "ip_addresses":
             object_query_params = self._build_query_params("prefix", data)
-            try:
-                self.nb_object = nb_app.prefixes.get(**object_query_params)
-            except ValueError:
-                self._handle_errors(msg="More than one result returned for %s" % (name))
+            self.nb_object = self._nb_endpoint_get(
+                nb_app.prefixes, object_query_params, name
+            )
         else:
-            try:
-                self.nb_object = nb_endpoint.get(**object_query_params)
-            except ValueError:
-                self._handle_errors(msg="More than one result returned for %s" % (name))
+            self.nb_object = self._nb_endpoint_get(
+                nb_endpoint, object_query_params, name
+            )
 
         if self.state in ("new", "present") and endpoint_name == "ip_address":
             self._handle_state_new_present(

plugins/module_utils/netbox_secrets.py

@@ -45,10 +45,7 @@ def run(self):
         data["slug"] = self._to_slug(name)
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

plugins/module_utils/netbox_tenancy.py

@@ -51,10 +51,7 @@ def run(self):
         data["slug"] = self._to_slug(name)
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

plugins/module_utils/netbox_utils.py

@@ -442,10 +442,31 @@ def __init__(self, module, endpoint, nb_client=None):
 
     def _connect_netbox_api(self, url, token, ssl_verify):
         try:
-            return pynetbox.api(url, token=token, ssl_verify=ssl_verify)
+            nb = pynetbox.api(url, token=token, ssl_verify=ssl_verify)
+            try:
+                self.version = nb.version
+            except AttributeError:
+                self.module.fail_json(msg="Must have pynetbox >=4.1.0")
+            except Exception:
+                self.module.fail_json(
+                    msg="Failed to establish connection to Netbox API"
+                )
+            return nb
         except Exception:
             self.module.fail_json(msg="Failed to establish connection to Netbox API")
 
+    def _nb_endpoint_get(self, nb_endpoint, query_params, search_item):
+        try:
+            response = nb_endpoint.get(**query_params)
+        except pynetbox.RequestError as e:
+            self._handle_errors(msg=e.error)
+        except ValueError:
+            self._handle_errors(
+                msg="More than one result returned for %s" % (search_item)
+            )
+
+        return response
+
     def _handle_errors(self, msg):
         """
         Returns message and changed = False
@@ -498,7 +519,10 @@ def _get_query_param_id(self, match, data):
             app = self._find_app(endpoint)
             nb_app = getattr(self.nb, app)
             nb_endpoint = getattr(nb_app, endpoint)
-            result = nb_endpoint.get(**{QUERY_TYPES.get(match): data[match]})
+
+            query_params = {QUERY_TYPES.get(match): data[match]}
+            result = self._nb_endpoint_get(nb_endpoint, query_params, match)
+
             if result:
                 return result.id
             else:
@@ -607,27 +631,22 @@ def _find_ids(self, data):
                         nb_app = getattr(self.nb, "virtualization")
                         nb_endpoint = getattr(nb_app, endpoint)
                     query_params = self._build_query_params(k, data, v)
-                    query_id = nb_endpoint.get(**query_params)
+                    query_id = self._nb_endpoint_get(nb_endpoint, query_params, k)
 
                 elif isinstance(v, list):
                     id_list = list()
                     for list_item in v:
                         norm_data = self._normalize_data(list_item)
                         temp_dict = self._build_query_params(k, data, norm_data)
-                        query_id = nb_endpoint.get(**temp_dict)
+                        query_id = self._nb_endpoint_get(nb_endpoint, temp_dict, k)
                         if query_id:
                             id_list.append(query_id.id)
                         else:
                             self._handle_errors(msg="%s not found" % (list_item))
 
                 else:
-                    try:
-                        query_id = nb_endpoint.get(**{QUERY_TYPES.get(k, "q"): search})
-                    except ValueError:
-                        self._handle_errors(
-                            msg="Multiple results found while searching for key: %s"
-                            % (k)
-                        )
+                    query_params = {QUERY_TYPES.get(k, "q"): search}
+                    query_id = self._nb_endpoint_get(nb_endpoint, query_params, k)
 
                 if isinstance(v, list):
                     data[k] = id_list

plugins/module_utils/netbox_virtualization.py

@@ -67,10 +67,7 @@ def run(self):
                 data["slug"] = self._to_slug(name)
 
         object_query_params = self._build_query_params(endpoint_name, data)
-        try:
-            self.nb_object = nb_endpoint.get(**object_query_params)
-        except ValueError:
-            self._handle_errors(msg="More than one result returned for %s" % (name))
+        self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":
             self._ensure_object_exists(nb_endpoint, endpoint_name, name, data)

tests/integration/regression-tests.yml

@@ -0,0 +1,35 @@
+---
+##
+##
+### TEST NETBOX CONNECTION FAILURE
+##
+##
+- name: "TEST NETBOX CONNECTION FAILURE"
+  hosts: localhost
+  connection: local
+  gather_facts: no
+  collections:
+    - netbox_community.ansible_modules
+  vars:
+    netbox_url: "http://localhost:32768"
+
+  tasks:
+    - name: "1 - Device with required information"
+      netbox_device:
+        netbox_url: netbox_url
+        netbox_token: "0123456789abcdef0123456789abcdef01234567"
+        data:
+          name: "R1"
+          device_type: "Cisco Test"
+          device_role: "Core Switch"
+          site: "Test Site"
+          status: "Staged"
+        state: present
+      register: test_one
+      ignore_errors: yes
+
+    - name: "1 - ASSERT"
+      assert:
+        that:
+          - test_one is failed
+          - test_one['msg'] == "Failed to establish connection to Netbox API"