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