tests/test_apex_ip_utils.py

   1 ##############################################################################
   2 # Copyright (c) 2016 Dan Radez (Red Hat)
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import re
  11 import ipaddress
  12
  13 from apex.ip_utils import IPUtilsException
  14 from apex.ip_utils import get_interface
  15 from apex.ip_utils import find_gateway
  16 from apex.ip_utils import get_ip
  17 from apex.ip_utils import get_ip_range
  18 from apex.ip_utils import _validate_ip_range
  19
  20 from nose.tools import assert_true
  21 from nose.tools import assert_false
  22 from nose.tools import assert_equal
  23 from nose.tools import assert_raises
  24 from nose.tools import assert_is_instance
  25 from nose.tools import assert_regexp_matches
  26
  27 from ipaddress import IPv4Address
  28 from ipaddress import IPv6Address
  29 from ipaddress import ip_network
  30
  31
  32 ip4_pattern = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
  33 ip4_range_pattern = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3},\d{1,'
  34                                '3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
  35
  36
  37 def get_default_gateway_linux():
  38     """Read the default gateway directly from /proc."""
  39     with open("/proc/net/route") as fh:
  40         for line in fh:
  41             fields = line.strip().split()
  42             if fields[2] not in ('00000000', 'Gateway'):
  43                 return fields[0]
  44
  45
  46 class TestIpUtils(object):
  47     @classmethod
  48     def setup_class(klass):
  49         """This method is run once for each class before any tests are run"""
  50         klass.iface_name = get_default_gateway_linux()
  51         iface = get_interface(klass.iface_name)
  52         klass.iface = iface
  53
  54     @classmethod
  55     def teardown_class(klass):
  56         """This method is run once for each class _after_ all tests are run"""
  57
  58     def setUp(self):
  59         """This method is run once before _each_ test method is executed"""
  60
  61     def teardown(self):
  62         """This method is run once after _each_ test method is executed"""
  63
  64     def test_get_interface(self):
  65         assert_equal(get_interface(''), None)
  66         assert_equal(get_interface('notreal'), None)
  67         assert_is_instance(get_interface(self.iface_name,
  68                                          address_family=4),
  69                            IPv4Address)
  70         # can't enable this until there's a v6 address on the ci hosts
  71         # assert_is_instance(get_interface(
  72         #                       self.iface_name,
  73         #                       address_family=6), IPv6Address)
  74         assert_raises(IPUtilsException,
  75                       get_interface, self.iface_name, 0)
  76
  77     def test_find_gateway(self):
  78         assert_is_instance(find_gateway(self.iface), str)
  79         iface_virbr0 = get_interface('virbr0')
  80         assert_equal(find_gateway(iface_virbr0), None)
  81
  82     def test_get_ip(self):
  83         cidr = ipaddress.ip_network("10.10.10.0/24")
  84         assert_equal(get_ip(1, cidr=cidr), "10.10.10.1")
  85         assert_raises(IPUtilsException, get_ip, 1000, interface=self.iface)
  86         assert_regexp_matches(get_ip(1, interface=self.iface), ip4_pattern)
  87         assert_raises(IPUtilsException, get_ip, 1)
  88
  89     def test_get_ip_range_raises(self):
  90         assert_raises(IPUtilsException, get_ip_range)
  91         assert_raises(IPUtilsException, get_ip_range, interface=self.iface)
  92
  93     def test_get_ip_range_with_interface(self):
  94         assert_regexp_matches(get_ip_range(interface=self.iface,
  95                                            start_offset=1, end_offset=20),
  96                               ip4_range_pattern)
  97         assert_regexp_matches(get_ip_range(interface=self.iface,
  98                                            start_offset=1, count=10),
  99                               ip4_range_pattern)
 100         assert_regexp_matches(get_ip_range(interface=self.iface, end_offset=20,
 101                                            count=10), ip4_range_pattern)
 102
 103     def test_get_ip_range_with_cidr(self):
 104         cidr = ip_network('10.10.10.0/24')
 105         assert_raises(IPUtilsException, get_ip_range, cidr=cidr)
 106         assert_regexp_matches(get_ip_range(cidr=cidr, start_offset=1,
 107                                            end_offset=20), ip4_pattern)
 108         assert_regexp_matches(get_ip_range(cidr=cidr, start_offset=1,
 109                                            count=10), ip4_pattern)
 110         assert_regexp_matches(get_ip_range(cidr=cidr, end_offset=20,
 111                                            count=10), ip4_pattern)
 112
 113     def test__validate_ip_range(self):
 114         cidr = ip_network('10.10.10.0/24')
 115         assert_true(_validate_ip_range(
 116                     start_ip=ipaddress.IPv4Address('10.10.10.1'),
 117                     end_ip=ipaddress.IPv4Address('10.10.10.10'),
 118                     cidr=cidr))
 119         assert_false(_validate_ip_range(
 120                      start_ip=ipaddress.IPv4Address('10.10.10.10'),
 121                      end_ip=ipaddress.IPv4Address('10.10.10.1'),
 122                      cidr=cidr))
 123         assert_false(_validate_ip_range(
 124                      start_ip=ipaddress.IPv4Address('10.10.0.1'),
 125                      end_ip=ipaddress.IPv4Address('10.10.10.10'),
 126                      cidr=cidr))
 127         assert_false(_validate_ip_range(
 128                      start_ip=ipaddress.IPv4Address('10.10.10.1'),
 129                      end_ip=ipaddress.IPv4Address('10.10.11.10'),
 130                      cidr=cidr))
 131
 132     def test_exception(self):
 133         e = IPUtilsException("test")
 134         print(e)
 135         assert_is_instance(e, IPUtilsException)