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 (
  16     assert_equal,
  17     assert_false,
  18     assert_is_instance,
  19     assert_raises,
  20     assert_regexp_matches,
  21     assert_true)
  22
  23 from apex.network.ip_utils import (
  24     IPUtilsException,
  25     _validate_ip_range,
  26     find_gateway,
  27     get_interface,
  28     get_ip,
  29     get_ip_range)
  30
  31 ip4_pattern = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
  32 ip4_range_pattern = re.compile('\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3},\d{1,'
  33                                '3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
  34
  35
  36 def get_default_gateway_linux():
  37     """Read the default gateway directly from /proc."""
  38     with open("/proc/net/route") as fh:
  39         for line in fh:
  40             fields = line.strip().split()
  41             if fields[2] not in ('00000000', 'Gateway'):
  42                 return fields[0]
  43
  44
  45 class TestIpUtils:
  46     @classmethod
  47     def setup_class(cls):
  48         """This method is run once for each class before any tests are run"""
  49         cls.iface_name = get_default_gateway_linux()
  50         iface = get_interface(cls.iface_name)
  51         cls.iface = iface
  52
  53     @classmethod
  54     def teardown_class(cls):
  55         """This method is run once for each class _after_ all tests are run"""
  56
  57     def setup(self):
  58         """This method is run once before _each_ test method is executed"""
  59
  60     def teardown(self):
  61         """This method is run once after _each_ test method is executed"""
  62
  63     def test_get_interface(self):
  64         assert_equal(get_interface(''), None)
  65         assert_equal(get_interface('notreal'), None)
  66         assert_is_instance(get_interface(self.iface_name,
  67                                          address_family=4),
  68                            IPv4Address)
  69         # can't enable this until there's a v6 address on the ci hosts
  70         # assert_is_instance(get_interface(
  71         #                       self.iface_name,
  72         #                       address_family=6), IPv6Address)
  73         assert_raises(IPUtilsException,
  74                       get_interface, self.iface_name, 0)
  75
  76     def test_find_gateway(self):
  77         assert_is_instance(find_gateway(self.iface), str)
  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)