<s> """<STR_LIT>""" <EOL> import logging <EOL> import unittest <EOL> from google . cloud . dataflow import coders <EOL> from google . cloud . dataflow . transforms . window import GlobalWindows <EOL> from google . cloud . dataflow . utils . counters import CounterFactory <EOL> from google . cloud . dataflow . worker . opcounters import OperationCounters <EOL> class OldClassThatDoesNotImplementLen : <EOL> def __init__ ( self ) : <EOL> pass <EOL> class ObjectThatDoesNotImplementLen ( object ) : <EOL> def __init__ ( self ) : <EOL> pass <EOL> class OperationCountersTest ( unittest . TestCase ) : <EOL> def verify_counters ( self , opcounts , expected_elements ) : <EOL> self . assertEqual ( expected_elements , opcounts . element_counter . total ) <EOL> self . assertEqual ( expected_elements , opcounts . element_counter . elements ) <EOL> def test_update_int ( self ) : <EOL> opcounts = OperationCounters ( CounterFactory ( ) , '<STR_LIT>' , <EOL> coders . PickleCoder ( ) , <NUM_LIT:0> ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:0> ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( <NUM_LIT:1> ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:1> ) <EOL> def test_update_str ( self ) : <EOL> opcounts = OperationCounters ( CounterFactory ( ) , '<STR_LIT>' , <EOL> coders . PickleCoder ( ) , <NUM_LIT:0> ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:0> ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( '<STR_LIT>' ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:1> ) <EOL> def test_update_old_object ( self ) : <EOL> opcounts = OperationCounters ( CounterFactory ( ) , '<STR_LIT>' , <EOL> coders . PickleCoder ( ) , <NUM_LIT:0> ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:0> ) <EOL> obj = OldClassThatDoesNotImplementLen ( ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( obj ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:1> ) <EOL> def test_update_new_object ( self ) : <EOL> opcounts = OperationCounters ( CounterFactory ( ) , '<STR_LIT>' , <EOL> coders . PickleCoder ( ) , <NUM_LIT:0> ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:0> ) <EOL> obj = ObjectThatDoesNotImplementLen ( ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( obj ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:1> ) <EOL> def test_update_multiple ( self ) : <EOL> opcounts = OperationCounters ( CounterFactory ( ) , '<STR_LIT>' , <EOL> coders . PickleCoder ( ) , <NUM_LIT:0> ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:0> ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( '<STR_LIT>' ) ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( '<STR_LIT>' ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:2> ) <EOL> opcounts . update_from ( GlobalWindows . WindowedValue ( '<STR_LIT>' ) ) <EOL> opcounts . update_collect ( ) <EOL> self . verify_counters ( opcounts , <NUM_LIT:3> ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . getLogger ( ) . setLevel ( logging . INFO ) <EOL> unittest . main ( ) </s>
