blockfindertest.py

   1 #!/usr/bin/python
   2 import blockfinder
   3 import unittest
   4 import os
   5 import shutil
   6 from tempfile import mkdtemp
   7 import test_data
   8
   9 try:
  10     import IPy
  11 except ImportError:
  12     IPy = None
  13
  14
  15 class BlockFinderTestExtras:
  16     def __init__(self):
  17         self.base_test_dir = mkdtemp()
  18         self.test_dir = self.base_test_dir + "/test/"
  19         self.block_f = blockfinder.Blockfinder(self.test_dir, "Mozilla")
  20
  21     def create_new_test_cache_dir(self):
  22         self.block_f.create_blockfinder_cache_dir()
  23         self.block_f.connect_to_database()
  24         self.block_f.create_sql_database()
  25
  26     def load_del_test_data(self):
  27         delegations = [test_data.return_sub_apnic_del()]
  28         self.block_f.insert_into_sql_database(delegations)
  29
  30     def load_lir_test_data(self):
  31         self.block_f.update_lir_delegation_cache("https://github.com/downloads/d1b/blockfinder/tiny_lir_data_for_test.gz")
  32         self.block_f.create_or_replace_lir_table_in_db()
  33         self.block_f.extract_info_from_lir_file_and_insert_into_sqlite("tiny_lir_data_for_test")
  34
  35     def copy_country_code_txt(self):
  36         shutil.copy(str(os.path.expanduser('~')) + "/.blockfinder/countrycodes.txt", self.block_f.cache_dir + "countrycodes.txt")
  37
  38     def clean_up(self):
  39         shutil.rmtree(self.base_test_dir, True)
  40
  41 class CheckReverseLookup(unittest.TestCase):
  42     ipValues = ( (3229318011, '192.123.123.123'),
  43             (3463778365, '206.117.16.61'),
  44             (4278190202, '255.0.0.122'),
  45             (3654084623, '217.204.232.15'),
  46             (134217728, '8.0.0.0'))
  47
  48     rirValues = ( ('217.204.232.15', 'GB'),
  49                   ('188.72.225.100', 'DE'),
  50                   ('8.8.8.1', 'US'),
  51                   ('193.9.26.0', 'HU'),
  52                   ('193.9.25.255', 'PL'),
  53                   )
  54     def setUp(self):
  55         self.block_f = blockfinder.Blockfinder(str(os.path.expanduser('~')) + "/.blockfinder/", "Mozilla")
  56         self.block_f.connect_to_database()
  57
  58     def test_rir_lookup(self):
  59         for ip, cc in self.rirValues:
  60             result = self.block_f.rir_lookup(ip)
  61             self.assertEqual(result[0], cc)
  62
  63     def test_ip_address_to_dec(self):
  64         for dec, ip in self.ipValues:
  65             result = blockfinder.ip_address_to_dec(ip)
  66             self.assertEqual(result, dec)
  67
  68 class CheckBlockFinder(unittest.TestCase):
  69     def setUp(self):
  70         self.extra_block_test_f = BlockFinderTestExtras()
  71         self.block_f = blockfinder.Blockfinder(self.extra_block_test_f.test_dir, "Mozilla")
  72
  73         self.extra_block_test_f.create_new_test_cache_dir()
  74         self.extra_block_test_f.load_del_test_data()
  75
  76     def tearDown(self):
  77         self.extra_block_test_f.clean_up()
  78
  79     # You can add known blocks to the tuple as a list
  80     # they will be looked up and checked
  81     known_ipv4_Results = ( ('mm', ['203.81.160.0/20', '203.81.64.0/19']),
  82                              ('kp', ['175.45.176.0/22']))
  83
  84     def test_ipv4_bf(self):
  85         self.block_f.connect_to_database()
  86         for cc, values in self.known_ipv4_Results:
  87             result = self.block_f.use_sql_database("ipv4", cc.upper())
  88             self.assertEqual(result, values)
  89         self.block_f.conn.close()
  90     def test_ipv6_bf(self):
  91         self.block_f.connect_to_database()
  92         expected = ['2001:200:2000::/35', '2001:200:4000::/34', '2001:200:8000::/33', '2001:200::/35']
  93         result = self.block_f.use_sql_database("ipv6", "JP")
  94         self.assertEqual(result, expected)
  95         self.block_f.conn.close()
  96
  97     def test_lir_fetching_and_use_ipv4(self):
  98         self.block_f.connect_to_database()
  99         self.extra_block_test_f.load_lir_test_data()
 100         self.block_f.download_country_code_file()
 101         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv4("80.16.151.184", "LIR"), ["IT", "Italy"])
 102         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv4("80.16.151.180", "LIR"), ["IT", "Italy"])
 103         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv4("213.95.6.32", "LIR"), ["DE", "Germany"])
 104         self.block_f.conn.close()
 105
 106     def test_lir_fetching_and_use_ipv6(self):
 107         """ ipv6 """
 108         self.block_f.connect_to_database()
 109         self.extra_block_test_f.load_lir_test_data()
 110         self.extra_block_test_f.copy_country_code_txt()
 111         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv6("2001:0658:021A::", "2001%", "LIR"), u"DE")
 112         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv6("2001:67c:320::", "2001%", "LIR"), u"DE")
 113         self.assertEqual(self.block_f.rir_or_lir_lookup_ipv6("2001:670:0085::", "2001%", "LIR"), u"FI")
 114
 115         self.block_f.conn.close()
 116
 117 class CheckBasicFunctionOperation(unittest.TestCase):
 118     def test_calc_ipv4_subnet_boundary(self):
 119         for i in range(0, 29):
 120             host_count = 2 ** i
 121             subnet = 32 - i
 122             self.assertEqual(blockfinder.calculate_ipv4_subnet(host_count), subnet)
 123
 124     def test_calc_ipv4_subnet_not_on_boundary(self):
 125         self.assertEqual(blockfinder.calculate_ipv4_subnet(254), 24)
 126         self.assertEqual(blockfinder.calculate_ipv4_subnet(255), 24)
 127         self.assertEqual(blockfinder.calculate_ipv4_subnet(257), 23)
 128         self.assertEqual(blockfinder.calculate_ipv4_subnet(259), 23)
 129
 130     def test_ipv4_address_to_dec(self):
 131         self.assertEqual(blockfinder.ip_address_to_dec("0.0.0.0"), 0)
 132         self.assertEqual(blockfinder.ip_address_to_dec("4.2.2.2"), 67240450)
 133         self.assertEqual(blockfinder.ip_address_to_dec("217.204.232.15"), 3654084623)
 134         self.assertEqual(blockfinder.ip_address_to_dec("255.255.255.255"), 4294967295)
 135
 136     def test_ipv4_address_to_dec_against_IPy(self):
 137         if IPy is not None:
 138             for i in range(0, 255):
 139                 ipaddr = "%s.%s.%s.%s" % (i, i, i, i)
 140                 self.assertEqual(blockfinder.ip_address_to_dec(ipaddr), IPy.IP(ipaddr).int())
 141
 142     def test_return_first_ip_and_number_in_inetnum(self):
 143         line = "1.1.1.1 - 1.1.1.2"
 144         self.assertEqual(blockfinder.return_first_ip_and_number_in_inetnum(line), ("1.1.1.1", 2) )
 145
 146 if __name__ == '__main__':
 147     for test_class in [CheckReverseLookup, CheckBlockFinder, CheckBasicFunctionOperation]:
 148         unittest.TextTestRunner(verbosity=2).run(unittest.makeSuite(test_class))
 149