<s> """<STR_LIT>""" <EOL> import collections <EOL> import logging <EOL> import re <EOL> import tempfile <EOL> import unittest <EOL> import google . cloud . dataflow as df <EOL> from google . cloud . dataflow . examples . complete import autocomplete <EOL> from google . cloud . dataflow . pvalue import AsIter <EOL> from google . cloud . dataflow . utils import options <EOL> def assert_that ( pcoll , matcher ) : <EOL> """<STR_LIT>""" <EOL> singleton = pcoll . pipeline | df . Create ( '<STR_LIT>' , [ None ] ) <EOL> def check_matcher ( _ , side_value ) : <EOL> assert matcher ( side_value ) <EOL> return [ ] <EOL> singleton | df . FlatMap ( check_matcher , AsIter ( pcoll ) ) <EOL> def contains_in_any_order ( expected ) : <EOL> def matcher ( value ) : <EOL> vs = collections . Counter ( value ) <EOL> es = collections . Counter ( expected ) <EOL> if vs != es : <EOL> raise ValueError ( <EOL> '<STR_LIT>' % ( vs - es , es - vs ) ) <EOL> return True <EOL> return matcher <EOL> class WordCountTest ( unittest . TestCase ) : <EOL> WORDS = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:to>' , '<STR_LIT:to>' , '<STR_LIT:to>' ] <EOL> def test_top_prefixes ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' ) <EOL> words = p | df . Create ( '<STR_LIT>' , self . WORDS ) <EOL> result = words | autocomplete . TopPerPrefix ( '<STR_LIT:test>' , <NUM_LIT:5> ) <EOL> result = result | df . Map ( lambda ( k , vs ) : ( k , tuple ( vs ) ) ) <EOL> assert_that ( result , contains_in_any_order ( <EOL> [ <EOL> ( '<STR_LIT:t>' , ( ( <NUM_LIT:3> , '<STR_LIT:to>' ) , ( <NUM_LIT:2> , '<STR_LIT>' ) , ( <NUM_LIT:1> , '<STR_LIT>' ) ) ) , <EOL> ( '<STR_LIT:to>' , ( ( <NUM_LIT:3> , '<STR_LIT:to>' ) , ) ) , <EOL> ( '<STR_LIT>' , ( ( <NUM_LIT:2> , '<STR_LIT>' ) , ( <NUM_LIT:1> , '<STR_LIT>' ) ) ) , <EOL> ( '<STR_LIT>' , ( ( <NUM_LIT:2> , '<STR_LIT>' ) , ) ) , <EOL> ( '<STR_LIT>' , ( ( <NUM_LIT:2> , '<STR_LIT>' ) , ) ) , <EOL> ( '<STR_LIT>' , ( ( <NUM_LIT:1> , '<STR_LIT>' ) , ) ) , <EOL> ( '<STR_LIT>' , ( ( <NUM_LIT:1> , '<STR_LIT>' ) , ) ) , <EOL> ] ) ) <EOL> p . run ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
