<s> """<STR_LIT>""" <EOL> import logging <EOL> import sys <EOL> import tempfile <EOL> import unittest <EOL> import google . cloud . dataflow as df <EOL> from google . cloud . dataflow import pvalue <EOL> from google . cloud . dataflow import typehints <EOL> from google . cloud . dataflow . examples . snippets import snippets <EOL> from google . cloud . dataflow . utils . options import TypeOptions <EOL> class ParDoTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_pardo ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:c>' ] <EOL> class ComputeWordLengthFn ( df . DoFn ) : <EOL> def process ( self , context ) : <EOL> return [ len ( context . element ) ] <EOL> word_lengths = words | df . ParDo ( ComputeWordLengthFn ( ) ) <EOL> self . assertEqual ( { <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:1> } , set ( word_lengths ) ) <EOL> def test_pardo_yield ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:c>' ] <EOL> class ComputeWordLengthFn ( df . DoFn ) : <EOL> def process ( self , context ) : <EOL> yield len ( context . element ) <EOL> word_lengths = words | df . ParDo ( ComputeWordLengthFn ( ) ) <EOL> self . assertEqual ( { <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:1> } , set ( word_lengths ) ) <EOL> def test_pardo_using_map ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:c>' ] <EOL> word_lengths = words | df . Map ( len ) <EOL> self . assertEqual ( { <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:1> } , set ( word_lengths ) ) <EOL> def test_pardo_using_flatmap ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:c>' ] <EOL> word_lengths = words | df . FlatMap ( lambda word : [ len ( word ) ] ) <EOL> self . assertEqual ( { <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:1> } , set ( word_lengths ) ) <EOL> def test_pardo_using_flatmap_yield ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:C>' ] <EOL> def capitals ( word ) : <EOL> for letter in word : <EOL> if '<STR_LIT:A>' <= letter <= '<STR_LIT>' : <EOL> yield letter <EOL> all_capitals = words | df . FlatMap ( capitals ) <EOL> self . assertEqual ( { '<STR_LIT:A>' , '<STR_LIT:C>' } , set ( all_capitals ) ) <EOL> def test_pardo_with_label ( self ) : <EOL> words = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> result = words | df . Map ( '<STR_LIT>' , lambda word : len ( set ( word ) ) ) <EOL> self . assertEqual ( { <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:4> } , set ( result ) ) <EOL> def test_pardo_side_input ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' ) <EOL> words = p | df . Create ( '<STR_LIT:start>' , [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def filter_using_length ( word , lower_bound , upper_bound = float ( '<STR_LIT>' ) ) : <EOL> if lower_bound <= len ( word ) <= upper_bound : <EOL> yield word <EOL> avg_word_len = words | df . Map ( len ) | df . CombineGlobally ( df . combiners . MeanCombineFn ( ) ) <EOL> small_words = words | df . FlatMap ( '<STR_LIT>' , filter_using_length , <NUM_LIT:0> , <NUM_LIT:3> ) <EOL> larger_than_average = words | df . FlatMap ( '<STR_LIT>' , <EOL> filter_using_length , <EOL> lower_bound = pvalue . AsSingleton ( avg_word_len ) ) <EOL> small_but_nontrivial = words | df . FlatMap ( filter_using_length , <EOL> lower_bound = <NUM_LIT:2> , <EOL> upper_bound = pvalue . AsSingleton ( avg_word_len ) ) <EOL> df . assert_that ( small_words , df . equal_to ( [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> df . assert_that ( larger_than_average , df . equal_to ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> label = '<STR_LIT>' ) <EOL> df . assert_that ( small_but_nontrivial , df . equal_to ( [ '<STR_LIT>' ] ) , <EOL> label = '<STR_LIT>' ) <EOL> p . run ( ) <EOL> def test_pardo_side_input_dofn ( self ) : <EOL> words = [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class FilterUsingLength ( df . DoFn ) : <EOL> def process ( self , context , lower_bound , upper_bound = float ( '<STR_LIT>' ) ) : <EOL> if lower_bound <= len ( context . element ) <= upper_bound : <EOL> yield context . element <EOL> small_words = words | df . ParDo ( FilterUsingLength ( ) , <NUM_LIT:0> , <NUM_LIT:3> ) <EOL> self . assertEqual ( { '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' } , set ( small_words ) ) <EOL> def test_pardo_with_side_outputs ( self ) : <EOL> class ProcessWords ( df . DoFn ) : <EOL> def process ( self , context , cutoff_length , marker ) : <EOL> if len ( context . element ) <= cutoff_length : <EOL> yield context . element <EOL> else : <EOL> yield pvalue . SideOutputValue ( <EOL> '<STR_LIT>' , len ( context . element ) ) <EOL> if context . element . startswith ( marker ) : <EOL> yield pvalue . SideOutputValue ( '<STR_LIT>' , context . element ) <EOL> words = [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> results = ( words | df . ParDo ( ProcessWords ( ) , cutoff_length = <NUM_LIT:2> , marker = '<STR_LIT:x>' ) <EOL> . with_outputs ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> main = '<STR_LIT>' ) ) <EOL> below = results . below_cutoff_strings <EOL> above = results . above_cutoff_lengths <EOL> marked = results [ '<STR_LIT>' ] <EOL> self . assertEqual ( { '<STR_LIT:a>' , '<STR_LIT>' } , set ( below ) ) <EOL> self . assertEqual ( { <NUM_LIT:3> , <NUM_LIT:5> } , set ( above ) ) <EOL> self . assertEqual ( { '<STR_LIT>' } , set ( marked ) ) <EOL> below , above , marked = ( words | df . ParDo ( ProcessWords ( ) , cutoff_length = <NUM_LIT:2> , marker = '<STR_LIT:x>' ) <EOL> . with_outputs ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> main = '<STR_LIT>' ) ) <EOL> self . assertEqual ( { '<STR_LIT:a>' , '<STR_LIT>' } , set ( below ) ) <EOL> self . assertEqual ( { <NUM_LIT:3> , <NUM_LIT:5> } , set ( above ) ) <EOL> self . assertEqual ( { '<STR_LIT>' } , set ( marked ) ) <EOL> def test_pardo_with_undeclared_side_outputs ( self ) : <EOL> numbers = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> , <NUM_LIT:10> , <NUM_LIT:20> ] <EOL> def even_odd ( x ) : <EOL> yield pvalue . SideOutputValue ( '<STR_LIT>' if x % <NUM_LIT:2> else '<STR_LIT>' , x ) <EOL> if x % <NUM_LIT:10> == <NUM_LIT:0> : <EOL> yield x <EOL> results = numbers | df . FlatMap ( even_odd ) . with_outputs ( ) <EOL> evens = results . even <EOL> odds = results . odd <EOL> tens = results [ None ] <EOL> self . assertEqual ( { <NUM_LIT:2> , <NUM_LIT:4> , <NUM_LIT:10> , <NUM_LIT:20> } , set ( evens ) ) <EOL> self . assertEqual ( { <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:5> } , set ( odds ) ) <EOL> self . assertEqual ( { <NUM_LIT:10> , <NUM_LIT:20> } , set ( tens ) ) <EOL> class TypeHintsTest ( unittest . TestCase ) : <EOL> def test_bad_types ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' , argv = sys . argv ) <EOL> numbers = p | df . Create ( [ '<STR_LIT:1>' , '<STR_LIT:2>' , '<STR_LIT:3>' ] ) <EOL> evens = numbers | df . Filter ( lambda x : x % <NUM_LIT:2> == <NUM_LIT:0> ) <EOL> with self . assertRaises ( TypeError ) : <EOL> p . run ( ) <EOL> with self . assertRaises ( typehints . TypeCheckError ) : <EOL> p . options . view_as ( TypeOptions ) . pipeline_type_check = True <EOL> evens = numbers | df . Filter ( lambda x : x % <NUM_LIT:2> == <NUM_LIT:0> ) . with_input_types ( int ) <EOL> with self . assertRaises ( typehints . TypeCheckError ) : <EOL> @ df . typehints . with_input_types ( int ) <EOL> class FilterEvensDoFn ( df . DoFn ) : <EOL> def process ( self , context ) : <EOL> if context . element % <NUM_LIT:2> == <NUM_LIT:0> : <EOL> yield context . element <EOL> evens = numbers | df . ParDo ( FilterEvensDoFn ( ) ) <EOL> words = p | df . Create ( '<STR_LIT>' , [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT:c>' ] ) <EOL> T = df . typehints . TypeVariable ( '<STR_LIT:T>' ) <EOL> @ df . typehints . with_input_types ( T ) <EOL> @ df . typehints . with_output_types ( df . typehints . Tuple [ int , T ] ) <EOL> class MyTransform ( df . PTransform ) : <EOL> def apply ( self , pcoll ) : <EOL> return pcoll | df . Map ( lambda x : ( len ( x ) , x ) ) <EOL> words_with_lens = words | MyTransform ( ) <EOL> with self . assertRaises ( typehints . TypeCheckError ) : <EOL> words_with_lens | df . Map ( lambda x : x ) . with_input_types ( <EOL> df . typehints . Tuple [ int , int ] ) <EOL> def test_runtime_checks_off ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' , argv = sys . argv ) <EOL> p | df . Create ( [ '<STR_LIT:a>' ] ) | df . Map ( lambda x : <NUM_LIT:3> ) . with_output_types ( str ) <EOL> p . run ( ) <EOL> def test_runtime_checks_on ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' , argv = sys . argv ) <EOL> with self . assertRaises ( typehints . TypeCheckError ) : <EOL> p . options . view_as ( TypeOptions ) . runtime_type_check = True <EOL> p | df . Create ( [ '<STR_LIT:a>' ] ) | df . Map ( lambda x : <NUM_LIT:3> ) . with_output_types ( str ) <EOL> p . run ( ) <EOL> def test_deterministic_key ( self ) : <EOL> p = df . Pipeline ( '<STR_LIT>' , argv = sys . argv ) <EOL> lines = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> class Player ( object ) : <EOL> def __init__ ( self , team , name ) : <EOL> self . team = team <EOL> self . name = name <EOL> class PlayerCoder ( df . coders . Coder ) : <EOL> def encode ( self , player ) : <EOL> return '<STR_LIT>' % ( player . team , player . name ) <EOL> def decode ( self , s ) : <EOL> return Player ( * s . split ( '<STR_LIT::>' ) ) <EOL> def is_deterministic ( self ) : <EOL> return True <EOL> df . coders . registry . register_coder ( Player , PlayerCoder ) <EOL> def parse_player_and_score ( csv ) : <EOL> name , team , score = csv . split ( '<STR_LIT:U+002C>' ) <EOL> return Player ( team , name ) , int ( score ) <EOL> totals = ( <EOL> lines <EOL> | df . Map ( parse_player_and_score ) <EOL> | df . CombinePerKey ( sum ) . with_input_types ( df . typehints . Tuple [ Player , int ] ) ) <EOL> self . assertEquals ( <EOL> { ( '<STR_LIT>' , <NUM_LIT:3> ) , ( '<STR_LIT>' , <NUM_LIT:4> ) , ( '<STR_LIT>' , <NUM_LIT:3> ) } , <EOL> set ( totals | df . Map ( lambda ( k , v ) : ( k . name , v ) ) ) ) <EOL> class SnippetsTest ( unittest . TestCase ) : <EOL> def create_temp_file ( self , contents = '<STR_LIT>' ) : <EOL> with tempfile . NamedTemporaryFile ( delete = False ) as f : <EOL> f . write ( contents ) <EOL> return f . name <EOL> def get_output ( self , path , sorted_output = True ) : <EOL> lines = [ ] <EOL> with open ( path ) as f : <EOL> lines = f . readlines ( ) <EOL> if sorted_output : <EOL> return sorted ( s . rstrip ( '<STR_LIT:\n>' ) for s in lines ) <EOL> else : <EOL> return [ s . rstrip ( '<STR_LIT:\n>' ) for s in lines ] <EOL> def test_model_pipelines ( self ) : <EOL> temp_path = self . create_temp_file ( '<STR_LIT>' ) <EOL> result_path = temp_path + '<STR_LIT>' <EOL> snippets . model_pipelines ( [ <EOL> '<STR_LIT>' % temp_path , <EOL> '<STR_LIT>' % result_path ] ) <EOL> self . assertEqual ( <EOL> self . get_output ( result_path ) , <EOL> [ str ( s ) for s in [ ( u'<STR_LIT>' , <NUM_LIT:1> ) , ( u'<STR_LIT>' , <NUM_LIT:2> ) , ( u'<STR_LIT>' , <NUM_LIT:3> ) ] ] ) <EOL> def test_model_pcollection ( self ) : <EOL> temp_path = self . create_temp_file ( ) <EOL> snippets . model_pcollection ( [ '<STR_LIT>' % temp_path ] ) <EOL> self . assertEqual ( self . get_output ( temp_path , sorted_output = False ) , [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> def test_construct_pipeline ( self ) : <EOL> temp_path = self . create_temp_file ( <EOL> '<STR_LIT>' ) <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . construct_pipeline ( { '<STR_LIT>' : temp_path , '<STR_LIT>' : result_path } ) <EOL> self . assertEqual ( <EOL> self . get_output ( result_path ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_model_textio ( self ) : <EOL> temp_path = self . create_temp_file ( '<STR_LIT>' ) <EOL> result_path = temp_path + '<STR_LIT>' <EOL> snippets . model_textio ( { '<STR_LIT>' : temp_path , '<STR_LIT>' : result_path } ) <EOL> self . assertEqual ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . get_output ( result_path ) ) <EOL> def test_model_bigqueryio ( self ) : <EOL> self . assertEqual ( None , snippets . model_bigqueryio ( ) ) <EOL> def _run_test_pipeline_for_options ( self , fn ) : <EOL> temp_path = self . create_temp_file ( '<STR_LIT>' ) <EOL> result_path = temp_path + '<STR_LIT>' <EOL> fn ( [ <EOL> '<STR_LIT>' % temp_path , <EOL> '<STR_LIT>' % result_path ] ) <EOL> self . assertEqual ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . get_output ( result_path ) ) <EOL> def test_pipeline_options_local ( self ) : <EOL> self . _run_test_pipeline_for_options ( snippets . pipeline_options_local ) <EOL> def test_pipeline_options_remote ( self ) : <EOL> self . _run_test_pipeline_for_options ( snippets . pipeline_options_remote ) <EOL> def test_pipeline_options_command_line ( self ) : <EOL> self . _run_test_pipeline_for_options ( snippets . pipeline_options_command_line ) <EOL> def test_pipeline_logging ( self ) : <EOL> result_path = self . create_temp_file ( ) <EOL> lines = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> snippets . pipeline_logging ( lines , result_path ) <EOL> self . assertEqual ( <EOL> sorted ( '<STR_LIT:U+0020>' . join ( lines ) . split ( '<STR_LIT:U+0020>' ) ) , <EOL> self . get_output ( result_path ) ) <EOL> def test_examples_wordcount ( self ) : <EOL> pipelines = [ snippets . examples_wordcount_minimal , <EOL> snippets . examples_wordcount_wordcount , <EOL> snippets . pipeline_monitoring ] <EOL> for pipeline in pipelines : <EOL> temp_path = self . create_temp_file ( <EOL> '<STR_LIT>' ) <EOL> result_path = self . create_temp_file ( ) <EOL> pipeline ( { '<STR_LIT>' : temp_path , '<STR_LIT>' : result_path } ) <EOL> self . assertEqual ( <EOL> self . get_output ( result_path ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_examples_wordcount_debugging ( self ) : <EOL> temp_path = self . create_temp_file ( <EOL> '<STR_LIT>' ) <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . examples_wordcount_debugging ( <EOL> { '<STR_LIT>' : temp_path , '<STR_LIT>' : result_path } ) <EOL> self . assertEqual ( <EOL> self . get_output ( result_path ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_model_composite_transform_example ( self ) : <EOL> contents = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . model_composite_transform_example ( contents , result_path ) <EOL> self . assertEqual ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , self . get_output ( result_path ) ) <EOL> def test_model_multiple_pcollections_flatten ( self ) : <EOL> contents = [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' , '<STR_LIT:d>' , '<STR_LIT:e>' , '<STR_LIT:f>' ] <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . model_multiple_pcollections_flatten ( contents , result_path ) <EOL> self . assertEqual ( contents , self . get_output ( result_path ) ) <EOL> def test_model_multiple_pcollections_partition ( self ) : <EOL> contents = [ <NUM_LIT> , <NUM_LIT> , <NUM_LIT:64> , <NUM_LIT:32> , <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . model_multiple_pcollections_partition ( contents , result_path ) <EOL> self . assertEqual ( [ '<STR_LIT:0>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> self . get_output ( result_path ) ) <EOL> def test_model_group_by_key ( self ) : <EOL> contents = [ '<STR_LIT>' ] <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . model_group_by_key ( contents , result_path ) <EOL> expected = [ ( '<STR_LIT:a>' , <NUM_LIT:2> ) , ( '<STR_LIT>' , <NUM_LIT:3> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) ] <EOL> self . assertEqual ( [ str ( s ) for s in expected ] , self . get_output ( result_path ) ) <EOL> def test_model_co_group_by_key_tuple ( self ) : <EOL> email_list = [ [ '<STR_LIT:a>' , '<STR_LIT>' ] , [ '<STR_LIT:b>' , '<STR_LIT>' ] ] <EOL> phone_list = [ [ '<STR_LIT:a>' , '<STR_LIT>' ] , [ '<STR_LIT:b>' , '<STR_LIT>' ] ] <EOL> result_path = self . create_temp_file ( ) <EOL> snippets . model_co_group_by_key_tuple ( email_list , phone_list , result_path ) <EOL> expect = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . assertEqual ( expect , self . get_output ( result_path ) ) <EOL> class CombineTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_global_sum ( self ) : <EOL> pc = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> result = pc | df . CombineGlobally ( sum ) <EOL> self . assertEqual ( [ <NUM_LIT:6> ] , result ) <EOL> def test_combine_values ( self ) : <EOL> occurences = [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:9> ) , ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) ] <EOL> first_occurences = occurences | df . GroupByKey ( ) | df . CombineValues ( min ) <EOL> self . assertEqual ( { ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) } , set ( first_occurences ) ) <EOL> def test_combine_per_key ( self ) : <EOL> player_accuracies = [ <EOL> ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:9> ) , ( '<STR_LIT>' , <NUM_LIT:1> ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) ] <EOL> avg_accuracy_per_player = player_accuracies | df . CombinePerKey ( df . combiners . MeanCombineFn ( ) ) <EOL> self . assertEqual ( { ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) } , set ( avg_accuracy_per_player ) ) <EOL> def test_combine_concat ( self ) : <EOL> pc = [ '<STR_LIT:a>' , '<STR_LIT:b>' ] <EOL> def concat ( values , separator = '<STR_LIT:U+002CU+0020>' ) : <EOL> return separator . join ( values ) <EOL> with_commas = pc | df . CombineGlobally ( concat ) <EOL> with_dashes = pc | df . CombineGlobally ( concat , separator = '<STR_LIT:->' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( with_commas ) ) <EOL> self . assertTrue ( with_commas [ <NUM_LIT:0> ] in { '<STR_LIT>' , '<STR_LIT>' } ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( with_dashes ) ) <EOL> self . assertTrue ( with_dashes [ <NUM_LIT:0> ] in { '<STR_LIT>' , '<STR_LIT>' } ) <EOL> def test_bounded_sum ( self ) : <EOL> pc = [ <NUM_LIT:1> , <NUM_LIT:10> , <NUM_LIT:100> , <NUM_LIT:1000> ] <EOL> def bounded_sum ( values , bound = <NUM_LIT> ) : <EOL> return min ( sum ( values ) , bound ) <EOL> small_sum = pc | df . CombineGlobally ( bounded_sum ) <EOL> large_sum = pc | df . CombineGlobally ( bounded_sum , bound = <NUM_LIT> ) <EOL> self . assertEqual ( [ <NUM_LIT> ] , small_sum ) <EOL> self . assertEqual ( [ <NUM_LIT> ] , large_sum ) <EOL> def test_combine_reduce ( self ) : <EOL> factors = [ <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:5> , <NUM_LIT:7> ] <EOL> import functools <EOL> import operator <EOL> product = factors | df . CombineGlobally ( functools . partial ( reduce , operator . mul ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( [ <NUM_LIT> ] , product ) <EOL> def test_custom_average ( self ) : <EOL> pc = [ <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:5> , <NUM_LIT:7> ] <EOL> class AverageFn ( df . CombineFn ) : <EOL> def create_accumulator ( self ) : <EOL> return ( <NUM_LIT:0.0> , <NUM_LIT:0> ) <EOL> def add_input ( self , ( sum , count ) , input ) : <EOL> return sum + input , count + <NUM_LIT:1> <EOL> def merge_accumulators ( self , accumulators ) : <EOL> sums , counts = zip ( * accumulators ) <EOL> return sum ( sums ) , sum ( counts ) <EOL> def extract_output ( self , ( sum , count ) ) : <EOL> return sum / count if count else float ( '<STR_LIT>' ) <EOL> average = pc | df . CombineGlobally ( AverageFn ( ) ) <EOL> self . assertEqual ( [ <NUM_LIT> ] , average ) <EOL> def test_keys ( self ) : <EOL> occurrences = [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:5> ) , ( '<STR_LIT>' , <NUM_LIT:9> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) ] <EOL> unique_keys = occurrences | snippets . Keys ( ) <EOL> self . assertEqual ( { '<STR_LIT>' , '<STR_LIT>' } , set ( unique_keys ) ) <EOL> def test_count ( self ) : <EOL> occurrences = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> perkey_counts = occurrences | snippets . Count ( ) <EOL> self . assertEqual ( { ( '<STR_LIT>' , <NUM_LIT:3> ) , ( '<STR_LIT>' , <NUM_LIT:2> ) } , set ( perkey_counts ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . getLogger ( ) . setLevel ( logging . INFO ) <EOL> unittest . main ( ) </s>
