<commit_before>#!/usr/bin/python
import blockfinder
import unittest
import os
import shutil
from tempfile import mkdtemp
import test_data

try:
    import IPy
except ImportError:
    IPy = None


class BlockFinderTestExtras:
    def __init__(self):
        self.base_test_dir = mkdtemp()
        self.test_dir = self.base_test_dir + "/test/"
        self.block_f = blockfinder.Blockfinder(self.test_dir, "Mozilla")

    def create_new_test_cache_dir(self):
        self.block_f.create_blockfinder_cache_dir()
        self.block_f.connect_to_database()
        self.block_f.create_sql_database()

    def load_del_test_data(self):
        delegations = [test_data.return_sub_apnic_del()]
        self.block_f.insert_into_sql_database(delegations)

    def load_lir_test_data(self):
        self.block_f.update_lir_delegation_cache("https://github.com/downloads/d1b/blockfinder/tiny_lir_data_for_test.gz")
        self.block_f.create_or_replace_lir_table_in_db()
        self.block_f.extract_info_from_lir_file_and_insert_into_sqlite("tiny_lir_data_for_test")

    def copy_country_code_txt(self):
        shutil.copy(str(os.path.expanduser('~')) + "/.blockfinder/countrycodes.txt", self.block_f.cache_dir + "countrycodes.txt")

    def clean_up(self):
        shutil.rmtree(self.base_test_dir, True)

class CheckReverseLookup(unittest.TestCase):
    ipValues = ( (3229318011, '192.168.127.12'),
            (3463778365, '172.16.58.3'),
            (4278190202, '255.0.0.122'),
            (3654084623, '172.16.17.32'),
            (134217728, '8.0.0.0'))

    rirValues = ( ('172.16.17.32', 'GB'),
                  ('192.168.3.11', 'DE'),
                  ('8.8.8.1', 'US'),
                  ('192.168.127.12', 'HU'),
                  ('172.16.17.32', 'PL'),
                  )
    def setUp(self):
        self.block_f = blockfinder.Blockfinder(str(os.path.expanduser('~')) + "/.blockfinder/", "Mozilla")
        self.block_f.connect_to_database()

    def test_rir_lookup(self):
        for ip, cc in self.rirValues:
            result = self.block_f.rir_lookup(ip)
            self.assertEqual(result[0], cc)

    def test_ip_address_to_dec(self):
        for dec, ip in self.ipValues:
            result = blockfinder.ip_address_to_dec(ip)
            self.assertEqual(result, dec)

class CheckBlockFinder(unittest.TestCase):
    def setUp(self):
        self.extra_block_test_f = BlockFinderTestExtras()
        self.block_f = blockfinder.Blockfinder(self.extra_block_test_f.test_dir, "Mozilla")

        self.extra_block_test_f.create_new_test_cache_dir()
        self.extra_block_test_f.load_del_test_data()

    def tearDown(self):
        self.extra_block_test_f.clean_up()

    # You can add known blocks to the tuple as a list
    # they will be looked up and checked
    known_ipv4_Results = ( ('mm', ['192.168.3.11/20', '192.168.3.11/19']),
                             ('kp', ['172.16.17.32/22']))

    def test_ipv4_bf(self):
        self.block_f.connect_to_database()
        for cc, values in self.known_ipv4_Results:
            result = self.block_f.use_sql_database("ipv4", cc.upper())
            self.assertEqual(result, values)
        self.block_f.conn.close()
    def test_ipv6_bf(self):
        self.block_f.connect_to_database()
        expected = ['2001:200:2000::/35', '2001:200:4000::/34', '2001:200:8000::/33', '2001:200::/35']
        result = self.block_f.use_sql_database("ipv6", "JP")
        self.assertEqual(result, expected)
        self.block_f.conn.close()

    def test_lir_fetching_and_use(self):
        """ test LIR fetching and use. """<commit_msg>Update the test_rir_lookup test case to use values from local test data. Also add a BaseBlockfinderTest class which CheckReverseLookup and CheckBlockFinder now inherent from. Signed-off-by: David <<EMAIL>><commit_after>#!/usr/bin/python
import blockfinder
import unittest
import os
import shutil
from tempfile import mkdtemp
import test_data

try:
    import IPy
except ImportError:
    IPy = None


class BlockFinderTestExtras:
    def __init__(self):
        self.base_test_dir = mkdtemp()
        self.test_dir = self.base_test_dir + "/test/"
        self.block_f = blockfinder.Blockfinder(self.test_dir, "Mozilla")

    def create_new_test_cache_dir(self):
        self.block_f.create_blockfinder_cache_dir()
        self.block_f.connect_to_database()
        self.block_f.create_sql_database()

    def load_del_test_data(self):
        delegations = [test_data.return_sub_apnic_del()]
        self.block_f.insert_into_sql_database(delegations)

    def load_lir_test_data(self):
        self.block_f.update_lir_delegation_cache("https://github.com/downloads/d1b/blockfinder/tiny_lir_data_for_test.gz")
        self.block_f.create_or_replace_lir_table_in_db()
        self.block_f.extract_info_from_lir_file_and_insert_into_sqlite("tiny_lir_data_for_test")

    def copy_country_code_txt(self):
        shutil.copy(str(os.path.expanduser('~')) + "/.blockfinder/countrycodes.txt", self.block_f.cache_dir + "countrycodes.txt")

    def clean_up(self):
        shutil.rmtree(self.base_test_dir, True)

class BaseBlockfinderTest(unittest.TestCase):
    """ This is the base blockfinder test class and provides
        a setUp and a tearDown which create and destroy a temporary
	cache directory and database respectively.
    """
    def setUp(self):
        self.extra_block_test_f = BlockFinderTestExtras()
        self.block_f = blockfinder.Blockfinder(self.extra_block_test_f.test_dir, "Mozilla")
        self.extra_block_test_f.create_new_test_cache_dir()
        self.extra_block_test_f.load_del_test_data()

    def tearDown(self):
        self.extra_block_test_f.clean_up()

class CheckReverseLookup(BaseBlockfinderTest):
    ipValues = ( (3229318011, '192.168.127.12'),
            (3463778365, '172.16.58.3'),
            (4278190202, '255.0.0.122'),
            (3654084623, '172.16.17.32'),
            (134217728, '8.0.0.0'))

    rirValues = ( ('192.168.3.11', 'KP'),
                  ('192.168.127.12', 'HU'),
                  ('172.16.17.32', 'PL'),
                  ('172.16.17.32', 'PL'),
                  )
    def tearDown(self):
        self.extra_block_test_f.clean_up()

    def test_rir_lookup(self):
        self.block_f.connect_to_database()
        self.block_f.download_country_code_file()

        for ip, cc in self.rirValues:
            result = self.block_f.rir_lookup(ip)
            self.assertEqual(result[0], cc)

    def test_ip_address_to_dec(self):
        for dec, ip in self.ipValues:
            result = blockfinder.ip_address_to_dec(ip)
            self.assertEqual(result, dec)

class CheckBlockFinder(BaseBlockfinderTest):
    # You can add known blocks to the tuple as a list
    # they will be looked up and checked
    known_ipv4_Results = ( ('mm', ['192.168.3.11/20', '192.168.3.11/19']),
                             ('kp', ['172.16.17.32/22']))

    def test_ipv4_bf(self):
        self.block_f.connect_to_database()
        for cc, values in self.known_ipv4_Results:
            result = self.block_f.use_sql_database("ipv4", cc.upper())
            self.assertEqual(result, values)
        self.block_f.conn.close()
    def test_ipv6_bf(self):
        self.block_f.connect_to_database()
        expected = ['2001:200:2000::/35', '2001:200:4000::/34', '2001:200:8000::/33', '2001:200::/35']
        result = self.block_f.use_sql_database("ipv6", "JP")
        self.assertEqual(result, expected)
        self.block_f.conn.close()

    def test_lir_fetching_and_use(self):
        """ test LIR fetching and use. """
{'fraction_non_alphanumeric': 0.1239265424758885, 'fraction_numerical': 0.05760338221693751, 'mean_word_length': 2.9081053175012905, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 25, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4508686', 'n_tokens_mistral': 2862, 'n_tokens_neox': 2595, 'n_words': 419}