<s> """<STR_LIT>""" <EOL> import logging <EOL> import tempfile <EOL> import unittest <EOL> from google . cloud . dataflow . examples . cookbook import mergecontacts <EOL> class MergeContactsTest ( unittest . TestCase ) : <EOL> CONTACTS_EMAIL = '<STR_LIT:\n>' . join ( [ '<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> CONTACTS_PHONE = '<STR_LIT:\n>' . join ( [ '<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> CONTACTS_SNAILMAIL = '<STR_LIT:\n>' . join ( <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> EXPECTED_TSV = '<STR_LIT:\n>' . join ( <EOL> [ '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT:\t>' . join ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) , <EOL> '<STR_LIT>' ] ) <EOL> EXPECTED_STATS = '<STR_LIT:\n>' . join ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> def create_temp_file ( self , contents ) : <EOL> with tempfile . NamedTemporaryFile ( delete = False ) as f : <EOL> f . write ( contents ) <EOL> return f . name <EOL> def normalize_tsv_results ( self , tsv_data ) : <EOL> """<STR_LIT>""" <EOL> lines_in = tsv_data . strip ( ) . split ( '<STR_LIT:\n>' ) <EOL> lines_out = [ ] <EOL> for line in lines_in : <EOL> name , email , phone , snailmail = line . split ( '<STR_LIT:\t>' ) <EOL> lines_out . append ( '<STR_LIT:\t>' . join ( <EOL> [ name , <EOL> '<STR_LIT>' % '<STR_LIT:U+002C>' . join ( sorted ( email . strip ( '<STR_LIT:">' ) . split ( '<STR_LIT:U+002C>' ) ) ) , <EOL> '<STR_LIT>' % '<STR_LIT:U+002C>' . join ( sorted ( phone . strip ( '<STR_LIT:">' ) . split ( '<STR_LIT:U+002C>' ) ) ) , <EOL> snailmail ] ) ) <EOL> return '<STR_LIT:\n>' . join ( sorted ( lines_out ) ) + '<STR_LIT:\n>' <EOL> def test_mergecontacts ( self ) : <EOL> path_email = self . create_temp_file ( self . CONTACTS_EMAIL ) <EOL> path_phone = self . create_temp_file ( self . CONTACTS_PHONE ) <EOL> path_snailmail = self . create_temp_file ( self . CONTACTS_SNAILMAIL ) <EOL> result_prefix = self . create_temp_file ( '<STR_LIT>' ) <EOL> mergecontacts . run ( [ <EOL> '<STR_LIT>' % path_email , <EOL> '<STR_LIT>' % path_phone , <EOL> '<STR_LIT>' % path_snailmail , <EOL> '<STR_LIT>' % result_prefix , <EOL> '<STR_LIT>' % result_prefix ] , assert_results = ( <NUM_LIT:2> , <NUM_LIT:1> , <NUM_LIT:3> ) ) <EOL> with open ( '<STR_LIT>' % result_prefix ) as f : <EOL> contents = f . read ( ) <EOL> self . assertEqual ( self . EXPECTED_TSV , self . normalize_tsv_results ( contents ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . getLogger ( ) . setLevel ( logging . INFO ) <EOL> unittest . main ( ) </s>
