Unit testing for ip_range

- modifies custom module ip_range.py in such a way that its individual
  stages can be easily tested
- tests for this module (test_ip_range.py) are created

Change-Id: Id1ee5db19c673da318ec21d98bdd02c766a219f5
Partial-Bug: #1594785

test-requirements.txt

@@ -13,3 +13,4 @@ testrepository>=0.0.18 # Apache-2.0/BSD
 testscenarios>=0.4 # Apache-2.0/BSD
 testtools>=1.4.0 # MIT
 reno>=1.8.0 # Apache-2.0
+netaddr>=0.7.13,!=0.7.16 # BSD

tox.ini

@@ -5,7 +5,7 @@ skipsdist = True
 
 [testenv]
 usedevelop = True
-install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
+install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages} "ansible>=2"
 setenv =
    VIRTUAL_ENV={envdir}
 deps = -r{toxinidir}/test-requirements.txt

tripleo_validations/tests/test_ip_range.py

@@ -0,0 +1,65 @@
+# -*- coding: utf-8 -*-
+
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+"""
+test_ip_range
+----------------------------------
+
+Tests for `ip_range` module.
+"""
+
+from tripleo_validations.tests import base
+import validations.library.ip_range as validation
+
+
+class TestIPRange(base.TestCase):
+
+    def test_check_arguments_non_IP(self):
+        '''Test ip_range when start is not an IP'''
+        errors = validation.check_arguments('something', '192.168.0.1', 1)
+        self.assertEqual(len(errors), 1)
+        self.assertEqual('Argument start (something) must be an IP', errors[0])
+
+    def test_check_arguments_IP_versions(self):
+        '''Test ip_range when start is IPv4 and end is IPv6'''
+        errors = validation.check_arguments('191.168.0.1', '::1', 2)
+        self.assertEqual(len(errors), 1)
+        self.assertEqual('Arguments start, end must share the same IP version',
+                         errors[0])
+
+    def test_check_arguments_neg_range(self):
+        '''Test ip_range when min_size is a negative number'''
+        errors = validation.check_arguments('192.168.0.1', '192.168.0.2', -3)
+        self.assertEqual(len(errors), 1)
+        self.assertEqual('Argument min_size(-3) must be greater than 0',
+                         errors[0])
+
+    def test_check_arguments_IPv4_ok(self):
+        '''Test ip_range on valid IPv4 arguments'''
+        errors = validation.check_arguments('192.168.0.1', '192.169.0.254', 5)
+        self.assertEqual(errors, [])
+
+    def test_check_arguments_IPv6_ok(self):
+        '''Test ip_range on valid IPv6 arguments'''
+        errors = validation.check_arguments('2001:d8::1', '2001:d8::1:1', 120)
+        self.assertEqual(errors, [])
+
+    def test_check_IP_range_too_small(self):
+        '''Test ip_range when range is less than minimal'''
+        warnings = validation.check_IP_range('192.168.0.1', '192.168.0.5', 6)
+        self.assertEqual(len(warnings), 2)
+        self.assertEqual('The IP range 192.168.0.1 - 192.168.0.5 ' +
+                         'contains 5 addresses.', warnings[0])
+        self.assertEqual('This might not be enough for the deployment ' +
+                         'or later scaling.', warnings[1])

validations/library/ip_range.py

@@ -19,6 +19,48 @@ import netaddr
 from ansible.module_utils.basic import *  # NOQA
 
 
+def check_arguments(start, end, min_size):
+    '''Validate format of arguments'''
+
+    errors = []
+
+    # Check format of arguments
+    try:
+        startIP = netaddr.IPAddress(start)
+    except netaddr.core.AddrFormatError:
+        errors.append('Argument start ({}) must be an IP'.format(start))
+
+    try:
+        endIP = netaddr.IPAddress(end)
+    except netaddr.core.AddrFormatError:
+        errors.append('Argument end ({}) must be an IP'.format(end))
+
+    if (not errors) and (startIP.version != endIP.version):
+        errors.append('Arguments start, end must share the same IP version')
+
+    if min_size < 0:
+        errors.append('Argument min_size({}) must be greater than 0'
+                      .format(min_size))
+
+    return errors
+
+
+def check_IP_range(start, end, min_size):
+    '''Compare IP range with minimum size'''
+
+    warnings = []
+    iprange = netaddr.IPRange(start, end)
+
+    if len(iprange) < min_size:
+        warnings = [
+            'The IP range {} - {} contains {} addresses.'.format(
+                start, end, len(iprange)),
+            'This might not be enough for the deployment or later scaling.'
+        ]
+
+    return warnings
+
+
 def main():
     module = AnsibleModule(argument_spec=dict(
         start=dict(required=True, type='str'),
@@ -28,19 +70,20 @@ def main():
 
     start = module.params.get('start')
     end = module.params.get('end')
+    min_size = module.params.get('min_size')
 
-    iprange = netaddr.IPRange(start, end)
-
-    if len(iprange) < module.params.get('min_size'):
-        module.exit_json(
-            changed=True,
-            warnings=[
-                'The IP range {} - {} contains {} addresses.'.format(
-                    start, end, len(iprange)),
-                'This might not be enough for the deployment or later scaling.'
-            ])
+    # Check arguments
+    errors = check_arguments(start, end, min_size)
+    if errors:
+        module.fail_json(msg='\n'.join(errors))
     else:
-        module.exit_json(msg='success')
+        # Check IP range
+        warnings = check_IP_range(start, end, min_size)
+
+        if warnings:
+            module.exit_json(changed=True, warnings=warnings)
+        else:
+            module.exit_json(msg='success')
 
 
 if __name__ == '__main__':