add _version_sanitize and improve _version_check_greater in netbox_utils

changelogs/fragments/netbox_version_check_greater.yml

@@ -0,0 +1,3 @@
+---
+minor_changes:
+  - improve version_check_greater to be more universal

changelogs/fragments/netbox_version_sanitize.yml

@@ -0,0 +1,3 @@
+---
+minor_changes:
+  - sanitize netbox versions received from api

plugins/module_utils/netbox_utils.py

@@ -771,9 +771,11 @@ def __init__(self, module, endpoint, nb_client=None):
         else:
             self.nb = nb_client
             try:
-                self.version = self.nb.version
+                self.version = self._version_sanitize(self.nb.version)
                 try:
-                    self.full_version = self.nb.status().get("netbox-version")
+                    self.full_version = self._version_sanitize(
+                        self.nb.status().get("netbox-version")
+                    )
                 except Exception:
                     # For NetBox versions without /api/status endpoint
                     self.full_version = f"{self.version}.0"
@@ -790,32 +792,40 @@ def __init__(self, module, endpoint, nb_client=None):
         data = self._find_ids(choices_data, query_params)
         self.data = self._convert_identical_keys(data)
 
-    def _version_check_greater(self, greater, lesser, greater_or_equal=False):
+    @staticmethod
+    def _version_sanitize(raw_value: str) -> str:
+        """Return sanitized Netbox version.
+        Sanitize 4.2.9-Docker-3.2.1 and return only 4.2.9.
+        """
+        if not isinstance(raw_value, str):
+            raise ValueError(f"Invalid value {raw_value!r}: expected a string")
+
+        version_match = re.match(r"^(\d[\d.]*)", raw_value)
+        if version_match:
+            return version_match.group(1).rstrip(".")
+
+        raise ValueError(
+            f"Invalid version {raw_value!r}: must start with a digit (e.g. '1', '2.5', '4.2.9')"
+        )
+
+    def _version_check_greater(
+        self, greater: str, lesser: str, greater_or_equal=False
+    ) -> bool:
         """Determine if first argument is greater than second argument.
 
         Args:
             greater (str): decimal string
-            lesser (str): decimal string
+            lesser (str):  decimal string
         """
-        g_major, g_minor = greater.split(".")
-        l_major, l_minor = lesser.split(".")
-
-        # convert to ints
-        g_major = int(g_major)
-        g_minor = int(g_minor)
-        l_major = int(l_major)
-        l_minor = int(l_minor)
-
-        # If major version is higher then return true right off the bat
-        if g_major > l_major:
-            return True
-        elif greater_or_equal and g_major == l_major and g_minor >= l_minor:
-            return True
-        # If major versions are equal, and minor version is higher, return True
-        elif g_major == l_major and g_minor > l_minor:
-            return True
-
-        return False
+        t_greater = tuple(int(x) for x in self._version_sanitize(greater).split("."))
+        t_lesser = tuple(int(x) for x in self._version_sanitize(lesser).split("."))
+
+        # Pad shorter tuple with zeros
+        max_len = max(len(t_greater), len(t_lesser))
+        t_greater += (0,) * (max_len - len(t_greater))
+        t_lesser += (0,) * (max_len - len(t_lesser))
+
+        return t_greater > t_lesser if not greater_or_equal else t_greater >= t_lesser
 
     def _connect_netbox_api(self, url, token, ssl_verify, cert, headers=None):
         try:
@@ -830,9 +840,11 @@ def _connect_netbox_api(self, url, token, ssl_verify, cert, headers=None):
             nb = pynetbox.api(url, token=token)
             nb.http_session = session
             try:
-                self.version = nb.version
+                self.version = self._version_sanitize(nb.version)
                 try:
-                    self.full_version = nb.status().get("netbox-version")
+                    self.full_version = self._version_sanitize(
+                        nb.status().get("netbox-version")
+                    )
                 except Exception:
                     # For NetBox versions without /api/status endpoint
                     self.full_version = f"{self.version}.0"

tests/unit/module_utils/test_netbox_base_class.py

@@ -7,6 +7,7 @@
 
 __metaclass__ = type
 
+import re
 import os
 from functools import partial
 from unittest.mock import MagicMock
@@ -374,24 +375,29 @@ def test_update_netbox_object_with_changes_check_mode_true(
     assert diff == on_update_diff
 
 
-@pytest.mark.parametrize("version", ["2.9", "2.8", "2.7"])
+@pytest.mark.parametrize("version", ["2.13", "2.12", "2.11", "2.10.8", "2.10"])
 def test_version_check_greater_true(mock_netbox_module, nb_obj_mock, version):
     mock_netbox_module.nb_object = nb_obj_mock
-    assert mock_netbox_module._version_check_greater("2.10", version)
+    assert mock_netbox_module._version_check_greater(version, "2.9")
+    assert mock_netbox_module._version_check_greater(version, "2.9.11")
 
 
-@pytest.mark.parametrize("version", ["2.13", "2.12", "2.11", "2.10"])
+@pytest.mark.parametrize("version", ["2.9", "2.8", "2.7.12", "2.7"])
 def test_version_check_greater_false(mock_netbox_module, nb_obj_mock, version):
     mock_netbox_module.nb_object = nb_obj_mock
-    assert not mock_netbox_module._version_check_greater("2.10", version)
+    assert not mock_netbox_module._version_check_greater(version, "2.10")
+    assert not mock_netbox_module._version_check_greater(version, "2.10.8")
 
 
-@pytest.mark.parametrize("version", ["2.9", "2.8", "2.7"])
+@pytest.mark.parametrize("version", ["2.9", "2.8", "2.7.5", "2.7"])
 def test_version_check_greater_equal_to_true(mock_netbox_module, nb_obj_mock, version):
     mock_netbox_module.nb_object = nb_obj_mock
     assert mock_netbox_module._version_check_greater(
         version, "2.7", greater_or_equal=True
     )
+    assert mock_netbox_module._version_check_greater(
+        version, "2.6.12", greater_or_equal=True
+    )
 
 
 @pytest.mark.parametrize("version", ["2.6", "2.5", "2.4"])
@@ -400,3 +406,34 @@ def test_version_check_greater_equal_to_false(mock_netbox_module, nb_obj_mock, v
     assert not mock_netbox_module._version_check_greater(
         version, "2.7", greater_or_equal=True
     )
+    assert not mock_netbox_module._version_check_greater(
+        version, "2.7.7", greater_or_equal=True
+    )
+
+
+@pytest.mark.parametrize(
+    "raw_value,expected",
+    [
+        ("2.6", "2.6"),
+        ("2.6.", "2.6"),
+        ("4.2-dev", "4.2"),
+        ("4", "4"),
+        ("4-dev", "4"),
+        ("4.-dev", "4"),
+        ("4.2.9-Docker-3.2.1", "4.2.9"),
+        ("3.1.0-extra-info", "3.1.0"),
+        ("10.20.30foobar", "10.20.30"),
+    ],
+)
+def test_version_sanitize_to_true(mock_netbox_module, nb_obj_mock, raw_value, expected):
+    mock_netbox_module.nb_object = nb_obj_mock
+    sanitized = mock_netbox_module._version_sanitize(raw_value)
+    assert sanitized == expected
+    assert re.match(r"^\d+(\.\d+)*$", sanitized)
+
+
+@pytest.mark.parametrize("version", [None, [], {}, "", "aa-dev", "-4", ".4", "dev-4"])
+def test_version_sanitize_value_error(mock_netbox_module, nb_obj_mock, version):
+    mock_netbox_module.nb_object = nb_obj_mock
+    with pytest.raises(ValueError):
+        mock_netbox_module._version_sanitize(version)