Merge pull request #8802 from nak3/sanity_check

Add format check of openshift_pkg_version and openshift_release

roles/lib_utils/action_plugins/sanity_checks.py

@@ -38,6 +38,18 @@ ENTERPRISE_TAG_REGEX = {'re': '(^v\\d+\\.\\d+(\\.\\d+)*(-\\d+(\\.\\d+)*)?$)',
 IMAGE_TAG_REGEX = {'origin': ORIGIN_TAG_REGEX,
                    'openshift-enterprise': ENTERPRISE_TAG_REGEX}
 
+PKG_VERSION_REGEX_ERROR = """openshift_pkg_version must be in the format
+-[optional.release]. Examples: -3.6.0, -3.7.0-0.126.0.git.0.9351aae.el7 -3.11*
+You specified openshift_pkg_version={}"""
+PKG_VERSION_REGEX = {'re': '(^-.*)',
+                     'error_msg': PKG_VERSION_REGEX_ERROR}
+
+RELEASE_REGEX_ERROR = """openshift_release must be in the format
+v#[.#[.#]]. Examples: v3.9, v3.10.0
+You specified openshift_release={}"""
+RELEASE_REGEX = {'re': '(^v?\\d+(\\.\\d+(\\.\\d+)?)?$)',
+                 'error_msg': RELEASE_REGEX_ERROR}
+
 STORAGE_KIND_TUPLE = (
     'openshift_hosted_registry_storage_kind',
     'openshift_loggingops_storage_kind',
@@ -262,6 +274,30 @@ class ActionModule(ActionBase):
             msg = msg.format(str(openshift_image_tag))
             raise errors.AnsibleModuleError(msg)
 
+    def check_pkg_version_format(self, hostvars, host):
+        """Ensure openshift_pkg_version is formatted correctly"""
+        openshift_pkg_version = self.template_var(hostvars, host, 'openshift_pkg_version')
+        if not openshift_pkg_version:
+            return None
+        regex_to_match = PKG_VERSION_REGEX['re']
+        res = re.match(regex_to_match, str(openshift_pkg_version))
+        if res is None:
+            msg = PKG_VERSION_REGEX['error_msg']
+            msg = msg.format(str(openshift_pkg_version))
+            raise errors.AnsibleModuleError(msg)
+
+    def check_release_format(self, hostvars, host):
+        """Ensure openshift_release is formatted correctly"""
+        openshift_release = self.template_var(hostvars, host, 'openshift_release')
+        if not openshift_release:
+            return None
+        regex_to_match = RELEASE_REGEX['re']
+        res = re.match(regex_to_match, str(openshift_release))
+        if res is None:
+            msg = RELEASE_REGEX['error_msg']
+            msg = msg.format(str(openshift_release))
+            raise errors.AnsibleModuleError(msg)
+
     def network_plugin_check(self, hostvars, host):
         """Ensure only one type of network plugin is enabled"""
         res = []
@@ -394,6 +430,8 @@ class ActionModule(ActionBase):
         self.check_whitelisted_registries(hostvars, host)
         self.check_python_version(hostvars, host, distro)
         self.check_image_tag_format(hostvars, host, odt)
+        self.check_pkg_version_format(hostvars, host)
+        self.check_release_format(hostvars, host)
         self.network_plugin_check(hostvars, host)
         self.check_hostname_vars(hostvars, host)
         self.check_session_auth_secrets(hostvars, host)

roles/lib_utils/test/sanity_check_test.py

@@ -0,0 +1,82 @@
+import os
+import pytest
+import sys
+
+from ansible.playbook.play_context import PlayContext
+from ansible.template import Templar
+from ansible import errors
+
+sys.path.insert(1, os.path.join(os.path.dirname(__file__), os.pardir, "action_plugins"))
+from sanity_checks import ActionModule  # noqa: E402
+
+
+@pytest.mark.parametrize('hostvars, host, varname, result', [
+    ({"example.com": {"param": 3.11}}, "example.com", "param", 3.11),
+    ({"example.com": {"param": 3.11}}, "example.com", "another_param", None)
+])
+def test_template_var(hostvars, host, varname, result):
+    task = FakeTask('sanity_checks', {'checks': []})
+    plugin = ActionModule(task, None, PlayContext(), None, Templar(None, None, None), None)
+    check = plugin.template_var(hostvars, host, varname)
+    assert check == result
+
+
+@pytest.mark.parametrize('hostvars, host, result', [
+    ({"example.com": {"openshift_pkg_version": "-3.6.0"}}, "example.com", None),
+    ({"example.com": {"openshift_pkg_version": "-3.7.0-0.126.0.git.0.9351aae.el7"}}, "example.com", None),
+    ({"example.com": {"openshift_pkg_version": "-3.9.0-2.fc28"}}, "example.com", None),
+    ({"example.com": {"openshift_pkg_version": "-3.11*"}}, "example.com", None),
+    ({"example.com": {"openshift_pkg_version": "-3"}}, "example.com", None),
+])
+def test_valid_check_pkg_version_format(hostvars, host, result):
+    task = FakeTask('sanity_checks', {'checks': []})
+    plugin = ActionModule(task, None, PlayContext(), None, Templar(None, None, None), None)
+    check = plugin.check_pkg_version_format(hostvars, host)
+    assert check == result
+
+
+@pytest.mark.parametrize('hostvars, host, result', [
+    ({"example.com": {"openshift_pkg_version": "3.11.0"}}, "example.com", None),
+    ({"example.com": {"openshift_pkg_version": "v3.11.0"}}, "example.com", None),
+])
+def test_invalid_check_pkg_version_format(hostvars, host, result):
+    with pytest.raises(errors.AnsibleModuleError):
+        task = FakeTask('sanity_checks', {'checks': []})
+        plugin = ActionModule(task, None, PlayContext(), None, Templar(None, None, None), None)
+        plugin.check_pkg_version_format(hostvars, host)
+
+
+@pytest.mark.parametrize('hostvars, host, result', [
+    ({"example.com": {"openshift_release": "v3"}}, "example.com", None),
+    ({"example.com": {"openshift_release": "v3.11"}}, "example.com", None),
+    ({"example.com": {"openshift_release": "v3.11.0"}}, "example.com", None),
+    ({"example.com": {"openshift_release": "3.11"}}, "example.com", None),
+])
+def test_valid_check_release_format(hostvars, host, result):
+    task = FakeTask('sanity_checks', {'checks': []})
+    plugin = ActionModule(task, None, PlayContext(), None, Templar(None, None, None), None)
+    check = plugin.check_release_format(hostvars, host)
+    assert check == result
+
+
+@pytest.mark.parametrize('hostvars, host, result', [
+    ({"example.com": {"openshift_release": "-3.11.0"}}, "example.com", None),
+    ({"example.com": {"openshift_release": "-3.7.0-0.126.0.git.0.9351aae.el7"}}, "example.com", None),
+    ({"example.com": {"openshift_release": "3.1.2.3"}}, "example.com", None),
+])
+def test_invalid_check_release_format(hostvars, host, result):
+    with pytest.raises(errors.AnsibleModuleError):
+        task = FakeTask('sanity_checks', {'checks': []})
+        plugin = ActionModule(task, None, PlayContext(), None, Templar(None, None, None), None)
+        plugin.check_release_format(hostvars, host)
+
+
+def fake_execute_module(*args):
+    raise AssertionError('this function should not be called')
+
+
+class FakeTask(object):
+    def __init__(self, action, args):
+        self.action = action
+        self.args = args
+        self.async = 0