<s> import dedupe <EOL> from collections import defaultdict <EOL> import unittest <EOL> from future . utils import viewitems , viewvalues <EOL> class BlockingTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . frozendict = dedupe . core . frozendict <EOL> field_definition = [ { '<STR_LIT>' : '<STR_LIT:name>' , '<STR_LIT:type>' : '<STR_LIT>' } ] <EOL> self . data_model = dedupe . Dedupe ( field_definition ) . data_model <EOL> self . training_pairs = { <EOL> '<STR_LIT>' : [ ( self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) ) , <EOL> ( self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) ) ] , <EOL> '<STR_LIT>' : [ ( self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) ) , <EOL> ( self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) ) , <EOL> ( self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> self . frozendict ( { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } ) ) ] <EOL> } <EOL> self . training = self . training_pairs [ '<STR_LIT>' ] + self . training_pairs [ '<STR_LIT>' ] <EOL> self . training_records = set ( ) . union ( * self . training ) <EOL> self . simple = lambda x : set ( [ str ( k ) for k in x <EOL> if "<STR_LIT>" not in str ( k ) ] ) <EOL> def test_dedupe_coverage ( self ) : <EOL> predicates = self . data_model . predicates ( ) <EOL> blocker = dedupe . blocking . Blocker ( predicates ) <EOL> blocker . indexAll ( { i : x for i , x in enumerate ( self . training_records ) } ) <EOL> coverage = dedupe . training . coveredPairs ( blocker . predicates , <EOL> self . training ) <EOL> assert self . simple ( coverage . keys ( ) ) . issuperset ( <EOL> set ( [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" ] ) ) <EOL> class TfidfTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . data_d = { <EOL> <NUM_LIT:100> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> } <EOL> def test_unconstrained_inverted_index ( self ) : <EOL> blocker = dedupe . blocking . Blocker ( [ dedupe . predicates . TfidfTextSearchPredicate ( <NUM_LIT:0.0> , "<STR_LIT:name>" ) ] ) <EOL> blocker . index ( set ( record [ "<STR_LIT:name>" ] <EOL> for record <EOL> in viewvalues ( self . data_d ) ) , <EOL> "<STR_LIT:name>" ) <EOL> blocks = defaultdict ( set ) <EOL> for block_key , record_id in blocker ( self . data_d . items ( ) ) : <EOL> blocks [ block_key ] . add ( record_id ) <EOL> blocks = set ( [ frozenset ( block ) for block in blocks . values ( ) <EOL> if len ( block ) > <NUM_LIT:1> ] ) <EOL> assert blocks == set ( [ frozenset ( [ <NUM_LIT> , <NUM_LIT> ] ) , frozenset ( [ <NUM_LIT> , <NUM_LIT> ] ) ] ) <EOL> class TfIndexUnindex ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> data_d = { <EOL> <NUM_LIT:100> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:1> } , <EOL> <NUM_LIT> : { "<STR_LIT:name>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : <NUM_LIT:0> } , <EOL> } <EOL> self . blocker = dedupe . blocking . Blocker ( [ dedupe . predicates . TfidfTextSearchPredicate ( <NUM_LIT:0.0> , "<STR_LIT:name>" ) ] ) <EOL> self . records_1 = dict ( ( record_id , record ) <EOL> for record_id , record <EOL> in viewitems ( data_d ) <EOL> if record [ "<STR_LIT>" ] == <NUM_LIT:0> ) <EOL> self . fields_2 = dict ( ( record_id , record [ "<STR_LIT:name>" ] ) <EOL> for record_id , record <EOL> in viewitems ( data_d ) <EOL> if record [ "<STR_LIT>" ] == <NUM_LIT:1> ) <EOL> def test_index ( self ) : <EOL> self . blocker . index ( set ( self . fields_2 . values ( ) ) , "<STR_LIT:name>" ) <EOL> blocks = defaultdict ( set ) <EOL> for block_key , record_id in self . blocker ( self . records_1 . items ( ) ) : <EOL> blocks [ block_key ] . add ( record_id ) <EOL> assert list ( blocks . items ( ) ) [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] == set ( [ <NUM_LIT> ] ) <EOL> def test_doubled_index ( self ) : <EOL> self . blocker . index ( self . fields_2 . values ( ) , "<STR_LIT:name>" ) <EOL> self . blocker . index ( self . fields_2 . values ( ) , "<STR_LIT:name>" ) <EOL> blocks = defaultdict ( set ) <EOL> for block_key , record_id in self . blocker ( self . records_1 . items ( ) ) : <EOL> blocks [ block_key ] . add ( record_id ) <EOL> assert list ( blocks . items ( ) ) == [ ( u'<STR_LIT>' , set ( [ <NUM_LIT> ] ) ) ] <EOL> def test_unindex ( self ) : <EOL> self . blocker . index ( self . fields_2 . values ( ) , "<STR_LIT:name>" ) <EOL> self . blocker . unindex ( self . fields_2 . values ( ) , "<STR_LIT:name>" ) <EOL> blocks = defaultdict ( set ) <EOL> for block_key , record_id in self . blocker ( self . records_1 . items ( ) ) : <EOL> blocks [ block_key ] . add ( record_id ) <EOL> assert len ( blocks . values ( ) ) == <NUM_LIT:0> <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
