<reponame>anhaidgroup/py_entitymatching<filename>py_entitymatching/tests/test_overlap_blocker.py
import os
from nose.tools import *
import pandas as pd
import unittest

import py_entitymatching as em

p = em.get_install_path()
path_a = os.sep.join([p, 'tests', 'test_datasets', 'A.csv'])
path_b = os.sep.join([p, 'tests', 'test_datasets', 'B.csv'])
l_overlap_attr_1 = 'name'
l_overlap_attr_2 = 'address'
r_overlap_attr_1 = 'name'
r_overlap_attr_2 = 'address'
l_output_attrs = ['name', 'address']
r_output_attrs = ['name', 'address']
l_output_prefix = 'l_'
r_output_prefix = 'r_'

# overlap on [r,l]_overlap_attr_1 with overlap_size=1
expected_ids_1 = [('a2', 'b3'), ('a2', 'b6'), ('a3', 'b2'), ('a5', 'b5')]

# overlap on [r,l]_overlap_attr_2 with overlap_size=4
expected_ids_2 = [('a2', 'b3'), ('a3', 'b2')]

# overlap on birth_year with q_val=3, overlap_size=2 (no padding) =6 (padding)
expected_ids_3 = [('a2', 'b3'), ('a3', 'b2'), ('a4', 'b1'), ('a4', 'b6'),
                  ('a5', 'b5')]

# block tables on [l|r]_overlap_attr_2, block candset on [l|r]overlap_attr_3
expected_ids_2_and_3 = [('a2', 'b3'), ('a3', 'b2')]

# overlap on [r,l]_overlap_attr_1, overlap_size=1 on tables with missing values
expected_ids_4 = [('a1', 'b4'), ('a2', 'b3'), ('a2', 'b4'), ('a2', 'b6'),
                  ('a3', 'b2'), ('a3', 'b4'), ('a4', 'b1'), ('a4', 'b2'),
                  ('a4', 'b3'), ('a4', 'b4'), ('a4', 'b5'), ('a4', 'b6'),
                  ('a5', 'b4'), ('a5', 'b5')]

# overlap on [l,r]_overlap_attr_2, overlap_size=4 on tables with missing values
expected_ids_5 = [('a2', 'b3'), ('a2', 'b6'), ('a3', 'b2'), ('a4', 'b6'),
                  ('a5', 'b4'), ('a5', 'b5')]

class OverlapBlockerTestCases(unittest.TestCase):

    def setUp(self):
        self.A = em.read_csv_metadata(path_a)
        em.set_key(self.A, 'ID')
        self.B = em.read_csv_metadata(path_b)
        em.set_key(self.B, 'ID')
        self.ob = em.OverlapBlocker()

    def tearDown(self):
        del self.A
        del self.B
        del self.ob

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_1(self):
        self.ob.block_tables(None, self.B, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_2(self):
        self.ob.block_tables([10, 10], self.B, l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_3(self):
        self.ob.block_tables(pd.DataFrame(), self.B, l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_1(self):
        self.ob.block_tables(self.A, None, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_2(self):
        self.ob.block_tables(self.A, [10, 10], l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_3(self):
        self.ob.block_tables(self.A, pd.DataFrame(), l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_1(self):
        self.ob.block_tables(self.A, self.B, None, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_2(self):
        self.ob.block_tables(self.A, self.B, 10, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_3(self):
        self.ob.block_tables(self.A, self.B, True, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_bogus_l_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, 'bogus_attr', r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_multi_l_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, ['birth_year', 'zipcode'],
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, 10)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, True)

    @raises(AssertionError)
    def test_ob_block_tables_bogus_r_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, 'bogus_attr')

    @raises(AssertionError)
    def test_ob_block_tables_multi_r_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             ['birth_year', 'zipcode'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs='name')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=[1, 2])

    @raises(AssertionError)
    def test_ob_block_tables_bogus_l_output_attrs(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=['bogus_attr'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs='name')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=[1, 2])

    @raises(AssertionError)
    def test_ob_block_tables_bogus_r_output_attrs(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=['bogus_attr'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=True)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=True)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_qval_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False, q_val=1.5)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_qval_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False, q_val='1')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level='yes')

    @raises(SyntaxError)
    def test_ob_block_tables_invalid_qval_word_level_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, q_val=1)

    @raises(SyntaxError)
    def test_ob_block_tables_invalid_qval_word_level_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=1.5)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size='1')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_4(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=-1)

    def test_ob_block_tables(self):
        C = self.ob.block_tables(self.A, self.B,
                                 l_overlap_attr_1, r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_empty_ltable(self):
        empty_A = pd.DataFrame(columns=self.A.columns)
        print(empty_A.dtypes)
        em.set_key(empty_A, 'ID')
        C = self.ob.block_tables(empty_A, self.B,
                                 l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_empty_rtable(self):
        empty_B = pd.DataFrame(columns=self.B.columns)
        em.set_key(empty_B, 'ID')
        C = self.ob.block_tables(self.A, empty_B,
                                 l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_wi_no_output_tuples(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, overlap_size=2)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_wi_null_l_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=None,
                                 r_output_attrs=r_output_attrs)
        validate_metadata(C, r_output_attrs=r_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_null_r_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=None)
        validate_metadata(C, l_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_empty_l_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=[],
                                 r_output_attrs=r_output_attrs)
        validate_metadata(C, [], r_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_empty_r_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=l_output_attrs,
                                 r_output_attrs=[])
        validate_metadata(C, l_output_attrs, [])
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_qval_non_str_attr(self):
         C = self.ob.block_tables(self.A, self.B, 'birth_year', 'birth_year',
                                  q_val=3, word_level=False, overlap_size=6)
         validate_metadata(C)
         validate_data(C, expected_ids_3)

    def test_ob_block_tables_wi_missing_values_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1, r_overlap_attr_1,
                                 allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)

    def test_ob_block_tables_wi_missing_values_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C, expected_ids_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_candset_1(self):
        self.ob.block_candset(None, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_candset_2(self):
        self.ob.block_candset([10, 10], l_overlap_attr_1, r_overlap_attr_1)

    @raises(KeyError)
    def test_ob_block_candset_invalid_candset_3(self):
        self.ob.block_candset(pd.DataFrame(), l_overlap_attr_1,
                              r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, None, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, 10, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, True, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_bogus_l_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, 'bogus_attr', r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_multi_l_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, ['birth_year', 'zipcode'], r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, 10)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, True)

    @raises(AssertionError)
    def test_ob_block_candset_bogus_r_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, 'bogus_attr')

    @raises(AssertionError)
    def test_ob_block_candset_multi_r_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, ['birth_year', 'zipcode'])

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              verbose=None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2, verbose=1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              verbose='yes')

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress=None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress=1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress='yes')

    def test_ob_block_candset(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True,overlap_size=4)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_candset_empty_input(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, overlap_size=2)
        validate_metadata(C)
        validate_data(C)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2)
        validate_metadata_two_candsets(C, D)
        validate_data(D)

    def test_ob_block_candset_empty_output(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_2,
                                 r_overlap_attr_2, overlap_size=4)
        validate_metadata(C)
        validate_data(C, expected_ids_2)
        D = self.ob.block_candset(C, l_overlap_attr_1, r_overlap_attr_1,
                                  overlap_size=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D)

    def test_ob_block_candset_wi_qval_non_str_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_2,
                                 r_overlap_attr_2, overlap_size=4)
        validate_metadata(C)
        validate_data(C, expected_ids_2)
        D = self.ob.block_candset(C, 'birth_year', 'birth_year',
                                  q_val=3, word_level=False, overlap_size=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2_and_3)

    def test_ob_block_candset_wi_missing_vals_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1,
                                 r_overlap_attr_1, allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  allow_missing=True)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_5)

    def test_ob_block_candset_wi_missing_vals_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1,
                                 r_overlap_attr_1, allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_tuples_whitespace(self):
        assert_equal(self.ob.block_tuples(self.A.loc[1], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1),
                     False)
        assert_equal(self.ob.block_tuples(self.A.loc[2], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1),
                     True)

    def test_ob_block_tuples_qgram(self):
        assert_equal(self.ob.block_tuples(self.A.loc[1], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1,
                                          q_val=3, word_level=False),
                     False)
        assert_equal(self.ob.block_tuples(self.A.loc[0], self.B.loc[0],
                                          l_overlap_attr_1, r_overlap_attr_1,
                                          q_val=3, word_level=False),
                     True)

    def test_ob_block_tuples_wi_missing_vals_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        assert_equal(self.ob.block_tuples(A.loc[1], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[2], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     True)

    def test_ob_block_tuples_wi_missing_vals_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        assert_equal(self.ob.block_tuples(A.loc[1], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1), True)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1), True)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1), True)



class OverlapBlockerMulticoreTestCases(unittest.TestCase):

    def setUp(self):
        self.A = em.read_csv_metadata(path_a)
        em.set_key(self.A, 'ID')
        self.B = em.read_csv_metadata(path_b)
        em.set_key(self.B, 'ID')
        self.ob = em.OverlapBlocker()

    def tearDown(self):
        del self.A
        del self.B
        del self.ob

    def test_ob_block_tables_njobs_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix, n_jobs=2)
        validate_metadata(C, l_output_attrs, r_output_attrs, l_output_prefix,
                          r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_njobs_all(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix, n_jobs=-1)
        validate_metadata(C, l_output_attrs, r_output_attrs, l_output_prefix,
                          r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_candset_njobs_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  n_jobs=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_candset_njobs_all(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  n_jobs=-1)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

# helper functions for validating the output

def validate_metadata(C, l_output_attrs=None, r_output_attrs=None,
                      l_output_prefix='ltable_', r_output_prefix='rtable_',
                      l_key='ID', r_key='ID'):
    s1 = ['_id', l_output_prefix + l_key, r_output_prefix + r_key]
    if l_output_attrs:
        s1 += [l_output_prefix + x for x in l_output_attrs if x != l_key]
    if r_output_attrs:
        s1 += [r_output_prefix + x for x in r_output_attrs if x != r_key]
    s1 = sorted(s1)
    assert_equal(s1, sorted(C.columns))
    assert_equal(em.get_key(C), '_id')
    assert_equal(em.get_property(C, 'fk_ltable'), l_output_prefix + l_key)
    assert_equal(em.get_property(C, 'fk_rtable'), r_output_prefix + r_key)

def validate_data(C, expected_ids=None):
    if expected_ids:
        lid = em.get_property(C, 'fk_ltable')
        rid = em.get_property(C, 'fk_rtable')
        C_ids = C[[lid, rid]].set_index([lid, rid])
        actual_ids = sorted(C_ids.index.values.tolist())
        assert_equal(expected_ids, actual_ids)
    else:
        assert_equal(len(C), 0)

def validate_metadata_two_candsets(C, D):
    assert_equal(sorted(C.columns), sorted(D.columns))
    assert_equal(em.get_key(D), em.get_key(C))
    assert_equal(em.get_property(D, 'fk_ltable'), em.get_property(C, 'fk_ltable'))
    assert_equal(em.get_property(D, 'fk_rtable'), em.get_property(C, 'fk_rtable'))

import os
from nose.tools import *
import pandas as pd
import unittest

import py_entitymatching as em

p = em.get_install_path()
path_a = os.sep.join([p, 'tests', 'test_datasets', 'A.csv'])
path_b = os.sep.join([p, 'tests', 'test_datasets', 'B.csv'])
l_overlap_attr_1 = 'name'
l_overlap_attr_2 = 'address'
r_overlap_attr_1 = 'name'
r_overlap_attr_2 = 'address'
l_output_attrs = ['name', 'address']
r_output_attrs = ['name', 'address']
l_output_prefix = 'l_'
r_output_prefix = 'r_'

# overlap on [r,l]_overlap_attr_1 with overlap_size=1
expected_ids_1 = [('a2', 'b3'), ('a2', 'b6'), ('a3', 'b2'), ('a5', 'b5')]

# overlap on [r,l]_overlap_attr_2 with overlap_size=4
expected_ids_2 = [('a2', 'b3'), ('a3', 'b2')]

# overlap on birth_year with q_val=3, overlap_size=2 (no padding) =6 (padding)
expected_ids_3 = [('a2', 'b3'), ('a3', 'b2'), ('a4', 'b1'), ('a4', 'b6'),
                  ('a5', 'b5')]

# block tables on [l|r]_overlap_attr_2, block candset on [l|r]overlap_attr_3
expected_ids_2_and_3 = [('a2', 'b3'), ('a3', 'b2')]

# overlap on [r,l]_overlap_attr_1, overlap_size=1 on tables with missing values
expected_ids_4 = [('a1', 'b4'), ('a2', 'b3'), ('a2', 'b4'), ('a2', 'b6'),
                  ('a3', 'b2'), ('a3', 'b4'), ('a4', 'b1'), ('a4', 'b2'),
                  ('a4', 'b3'), ('a4', 'b4'), ('a4', 'b5'), ('a4', 'b6'),
                  ('a5', 'b4'), ('a5', 'b5')]

# overlap on [l,r]_overlap_attr_2, overlap_size=4 on tables with missing values
expected_ids_5 = [('a2', 'b3'), ('a2', 'b6'), ('a3', 'b2'), ('a4', 'b6'),
                  ('a5', 'b4'), ('a5', 'b5')]

class OverlapBlockerTestCases(unittest.TestCase):

    def setUp(self):
        self.A = em.read_csv_metadata(path_a)
        em.set_key(self.A, 'ID')
        self.B = em.read_csv_metadata(path_b)
        em.set_key(self.B, 'ID')
        self.ob = em.OverlapBlocker()

    def tearDown(self):
        del self.A
        del self.B
        del self.ob

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_1(self):
        self.ob.block_tables(None, self.B, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_2(self):
        self.ob.block_tables([10, 10], self.B, l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_ltable_3(self):
        self.ob.block_tables(pd.DataFrame(), self.B, l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_1(self):
        self.ob.block_tables(self.A, None, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_2(self):
        self.ob.block_tables(self.A, [10, 10], l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rtable_3(self):
        self.ob.block_tables(self.A, pd.DataFrame(), l_overlap_attr_1,
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_1(self):
        self.ob.block_tables(self.A, self.B, None, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_2(self):
        self.ob.block_tables(self.A, self.B, 10, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_overlap_attr_3(self):
        self.ob.block_tables(self.A, self.B, True, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_bogus_l_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, 'bogus_attr', r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_multi_l_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, ['birth_year', 'zipcode'],
                             r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, 10)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_overlap_attr_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, True)

    @raises(AssertionError)
    def test_ob_block_tables_bogus_r_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1, 'bogus_attr')

    @raises(AssertionError)
    def test_ob_block_tables_multi_r_overlap_attr(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             ['birth_year', 'zipcode'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs='name')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_attrs_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=[1, 2])

    @raises(AssertionError)
    def test_ob_block_tables_bogus_l_output_attrs(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_attrs=['bogus_attr'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs='name')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_attrs_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=[1, 2])

    @raises(AssertionError)
    def test_ob_block_tables_bogus_r_output_attrs(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_attrs=['bogus_attr'])

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_l_output_prefix_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, l_output_prefix=True)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_r_output_prefix_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, r_output_prefix=True)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_allow_missing_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, allow_missing='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_verbose_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, verbose='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_rem_stop_words_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, rem_stop_words='yes')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_qval_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False, q_val=1.5)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_qval_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False, q_val='1')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=1)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_word_level_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level='yes')

    @raises(SyntaxError)
    def test_ob_block_tables_invalid_qval_word_level_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, q_val=1)

    @raises(SyntaxError)
    def test_ob_block_tables_invalid_qval_word_level_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, word_level=False)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_1(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=None)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_2(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=1.5)

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_3(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size='1')

    @raises(AssertionError)
    def test_ob_block_tables_invalid_overlap_size_4(self):
        self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                             r_overlap_attr_1, overlap_size=-1)

    def test_ob_block_tables(self):
        C = self.ob.block_tables(self.A, self.B,
                                 l_overlap_attr_1, r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_empty_ltable(self):
        empty_A = pd.DataFrame(columns=self.A.columns)
        print(empty_A.dtypes)
        em.set_key(empty_A, 'ID')
        C = self.ob.block_tables(empty_A, self.B,
                                 l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_empty_rtable(self):
        empty_B = pd.DataFrame(columns=self.B.columns)
        em.set_key(empty_B, 'ID')
        C = self.ob.block_tables(self.A, empty_B,
                                 l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_wi_no_output_tuples(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, overlap_size=2)
        validate_metadata(C)
        validate_data(C)

    def test_ob_block_tables_wi_null_l_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=None,
                                 r_output_attrs=r_output_attrs)
        validate_metadata(C, r_output_attrs=r_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_null_r_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=None)
        validate_metadata(C, l_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_empty_l_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=[],
                                 r_output_attrs=r_output_attrs)
        validate_metadata(C, [], r_output_attrs)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_empty_r_output_attrs(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, l_output_attrs=l_output_attrs,
                                 r_output_attrs=[])
        validate_metadata(C, l_output_attrs, [])
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_wi_qval_non_str_attr(self):
         C = self.ob.block_tables(self.A, self.B, 'birth_year', 'birth_year',
                                  q_val=3, word_level=False, overlap_size=6)
         validate_metadata(C)
         validate_data(C, expected_ids_3)

    def test_ob_block_tables_wi_missing_values_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1, r_overlap_attr_1,
                                 allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)

    def test_ob_block_tables_wi_missing_values_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1, r_overlap_attr_1)
        validate_metadata(C)
        validate_data(C, expected_ids_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_candset_1(self):
        self.ob.block_candset(None, l_overlap_attr_1, r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_candset_2(self):
        self.ob.block_candset([10, 10], l_overlap_attr_1, r_overlap_attr_1)

    @raises(KeyError)
    def test_ob_block_candset_invalid_candset_3(self):
        self.ob.block_candset(pd.DataFrame(), l_overlap_attr_1,
                              r_overlap_attr_1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, None, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, 10, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_l_overlap_attr_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, True, r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_bogus_l_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, 'bogus_attr', r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_multi_l_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, ['birth_year', 'zipcode'], r_overlap_attr_2)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, 10)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_r_overlap_attr_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, True)

    @raises(AssertionError)
    def test_ob_block_candset_bogus_r_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, 'bogus_attr')

    @raises(AssertionError)
    def test_ob_block_candset_multi_r_overlap_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, ['birth_year', 'zipcode'])

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              verbose=None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2, verbose=1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_verbose_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              verbose='yes')

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_1(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress=None)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress=1)

    @raises(AssertionError)
    def test_ob_block_candset_invalid_show_progress_3(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1)
        self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                              show_progress='yes')

    def test_ob_block_candset(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True,overlap_size=4)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_candset_empty_input(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1, overlap_size=2)
        validate_metadata(C)
        validate_data(C)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2)
        validate_metadata_two_candsets(C, D)
        validate_data(D)

    def test_ob_block_candset_empty_output(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_2,
                                 r_overlap_attr_2, overlap_size=4)
        validate_metadata(C)
        validate_data(C, expected_ids_2)
        D = self.ob.block_candset(C, l_overlap_attr_1, r_overlap_attr_1,
                                  overlap_size=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D)

    def test_ob_block_candset_wi_qval_non_str_attr(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_2,
                                 r_overlap_attr_2, overlap_size=4)
        validate_metadata(C)
        validate_data(C, expected_ids_2)
        D = self.ob.block_candset(C, 'birth_year', 'birth_year',
                                  q_val=3, word_level=False, overlap_size=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2_and_3)

    def test_ob_block_candset_wi_missing_vals_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1,
                                 r_overlap_attr_1, allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  allow_missing=True)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_5)

    def test_ob_block_candset_wi_missing_vals_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        C = self.ob.block_tables(A, B, l_overlap_attr_1,
                                 r_overlap_attr_1, allow_missing=True)
        validate_metadata(C)
        validate_data(C, expected_ids_4)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_tuples_whitespace(self):
        assert_equal(self.ob.block_tuples(self.A.loc[1], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1),
                     False)
        assert_equal(self.ob.block_tuples(self.A.loc[2], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1),
                     True)

    def test_ob_block_tuples_qgram(self):
        assert_equal(self.ob.block_tuples(self.A.loc[1], self.B.loc[2],
                                          l_overlap_attr_1, r_overlap_attr_1,
                                          q_val=3, word_level=False),
                     False)
        assert_equal(self.ob.block_tuples(self.A.loc[0], self.B.loc[0],
                                          l_overlap_attr_1, r_overlap_attr_1,
                                          q_val=3, word_level=False),
                     True)

    def test_ob_block_tuples_wi_missing_vals_allow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        assert_equal(self.ob.block_tuples(A.loc[1], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     False)
        assert_equal(self.ob.block_tuples(A.loc[2], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1, allow_missing=True),
                     True)

    def test_ob_block_tuples_wi_missing_vals_disallow_missing(self):
        path_a = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_A_wi_missing_vals.csv'])
        path_b = os.sep.join([p, 'tests', 'test_datasets', 'blocker',
                              'table_B_wi_missing_vals.csv'])
        A = em.read_csv_metadata(path_a)
        em.set_key(A, 'ID')
        B = em.read_csv_metadata(path_b)
        em.set_key(B, 'ID')
        assert_equal(self.ob.block_tuples(A.loc[1], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1), True)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[2], l_overlap_attr_1,
                                          r_overlap_attr_1), True)
        assert_equal(self.ob.block_tuples(A.loc[3], B.loc[3], l_overlap_attr_1,
                                          r_overlap_attr_1), True)



class OverlapBlockerMulticoreTestCases(unittest.TestCase):

    def setUp(self):
        self.A = em.read_csv_metadata(path_a)
        em.set_key(self.A, 'ID')
        self.B = em.read_csv_metadata(path_b)
        em.set_key(self.B, 'ID')
        self.ob = em.OverlapBlocker()

    def tearDown(self):
        del self.A
        del self.B
        del self.ob

    def test_ob_block_tables_njobs_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix, n_jobs=2)
        validate_metadata(C, l_output_attrs, r_output_attrs, l_output_prefix,
                          r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_tables_njobs_all(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix, n_jobs=-1)
        validate_metadata(C, l_output_attrs, r_output_attrs, l_output_prefix,
                          r_output_prefix)
        validate_data(C, expected_ids_1)

    def test_ob_block_candset_njobs_2(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  n_jobs=2)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

    def test_ob_block_candset_njobs_all(self):
        C = self.ob.block_tables(self.A, self.B, l_overlap_attr_1,
                                 r_overlap_attr_1,
                                 l_output_attrs=l_output_attrs,
                                 r_output_attrs=r_output_attrs,
                                 l_output_prefix=l_output_prefix,
                                 r_output_prefix=r_output_prefix)
        validate_metadata(C, l_output_attrs, r_output_attrs,
                          l_output_prefix, r_output_prefix)
        validate_data(C, expected_ids_1)
        D = self.ob.block_candset(C, l_overlap_attr_2, r_overlap_attr_2,
                                  rem_stop_words=True, overlap_size=4,
                                  n_jobs=-1)
        validate_metadata_two_candsets(C, D)
        validate_data(D, expected_ids_2)

# helper functions for validating the output

def validate_metadata(C, l_output_attrs=None, r_output_attrs=None,
                      l_output_prefix='ltable_', r_output_prefix='rtable_',
                      l_key='ID', r_key='ID'):
    s1 = ['_id', l_output_prefix + l_key, r_output_prefix + r_key]
    if l_output_attrs:
        s1 += [l_output_prefix + x for x in l_output_attrs if x != l_key]
    if r_output_attrs:
        s1 += [r_output_prefix + x for x in r_output_attrs if x != r_key]
    s1 = sorted(s1)
    assert_equal(s1, sorted(C.columns))
    assert_equal(em.get_key(C), '_id')
    assert_equal(em.get_property(C, 'fk_ltable'), l_output_prefix + l_key)
    assert_equal(em.get_property(C, 'fk_rtable'), r_output_prefix + r_key)

def validate_data(C, expected_ids=None):
    if expected_ids:
        lid = em.get_property(C, 'fk_ltable')
        rid = em.get_property(C, 'fk_rtable')
        C_ids = C[[lid, rid]].set_index([lid, rid])
        actual_ids = sorted(C_ids.index.values.tolist())
        assert_equal(expected_ids, actual_ids)
    else:
        assert_equal(len(C), 0)

def validate_metadata_two_candsets(C, D):
    assert_equal(sorted(C.columns), sorted(D.columns))
    assert_equal(em.get_key(D), em.get_key(C))
    assert_equal(em.get_property(D, 'fk_ltable'), em.get_property(C, 'fk_ltable'))
    assert_equal(em.get_property(D, 'fk_rtable'), em.get_property(C, 'fk_rtable'))
# overlap on [r,l]_overlap_attr_1 with overlap_size=1 # overlap on [r,l]_overlap_attr_2 with overlap_size=4 # overlap on birth_year with q_val=3, overlap_size=2 (no padding) =6 (padding) # block tables on [l|r]_overlap_attr_2, block candset on [l|r]overlap_attr_3 # overlap on [r,l]_overlap_attr_1, overlap_size=1 on tables with missing values # overlap on [l,r]_overlap_attr_2, overlap_size=4 on tables with missing values # helper functions for validating the output