<s> import unittest <EOL> from functools import wraps <EOL> import jsonstreamer <EOL> json_file_name = lambda test_fn : '<STR_LIT>' + test_fn . __name__ [ <NUM_LIT:5> : ] + '<STR_LIT>' <EOL> def load_test_data ( func ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( func ) <EOL> def wrapper ( self , * args , ** kwargs ) : <EOL> with open ( json_file_name ( func ) , encoding = '<STR_LIT:utf-8>' ) as json_file : <EOL> json_input = json_file . read ( ) <EOL> return func ( self , json_input ) <EOL> return wrapper <EOL> class JSONStreamerTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _assertions = [ ] <EOL> self . _streamer = jsonstreamer . JSONStreamer ( ) <EOL> self . _streamer . add_catch_all_listener ( self . _catch_all ) <EOL> def tearDown ( self ) : <EOL> self . _streamer . close ( ) <EOL> self . assertEqual ( len ( self . _assertions ) , <NUM_LIT:0> ) <EOL> def _catch_all ( self , event_name , * args ) : <EOL> value = args [ <NUM_LIT:0> ] if args else None <EOL> try : <EOL> e , v = self . _assertions . pop ( <NUM_LIT:0> ) <EOL> except IndexError : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> self . assertEqual ( event_name , e ) <EOL> self . assertEqual ( value , v ) <EOL> @ load_test_data <EOL> def test_simple_object ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT:key>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:value>' , <NUM_LIT:8> ) , <EOL> ( '<STR_LIT:key>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT:value>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , None ) ] <EOL> self . _streamer . consume ( json_input ) <EOL> class ObjectStreamerTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _assertions = [ ] <EOL> self . _streamer = jsonstreamer . ObjectStreamer ( ) <EOL> self . _streamer . add_catch_all_listener ( self . _catch_all ) <EOL> def tearDown ( self ) : <EOL> self . _streamer . close ( ) <EOL> self . assertEqual ( len ( self . _assertions ) , <NUM_LIT:0> ) <EOL> def _catch_all ( self , event_name , * args ) : <EOL> value = args [ <NUM_LIT:0> ] if args else None <EOL> try : <EOL> expected_event , expected_value = self . _assertions . pop ( <NUM_LIT:0> ) <EOL> except IndexError : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> self . assertEqual ( event_name , expected_event ) <EOL> self . _assert_value ( expected_value , value ) <EOL> def _assert_value ( self , expected_value , value ) : <EOL> if value and isinstance ( value , tuple ) and len ( value ) == <NUM_LIT:2> : <EOL> if isinstance ( value [ <NUM_LIT:1> ] , dict ) : <EOL> self . assertDictEqual ( value [ <NUM_LIT:1> ] , expected_value [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( expected_value [ <NUM_LIT:0> ] , value [ <NUM_LIT:0> ] ) <EOL> elif isinstance ( value [ <NUM_LIT:1> ] , list ) : <EOL> self . assertListEqual ( value [ <NUM_LIT:1> ] , expected_value [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( expected_value [ <NUM_LIT:0> ] , value [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . assertEqual ( value , expected_value ) <EOL> else : <EOL> self . assertEqual ( value , expected_value ) <EOL> @ load_test_data <EOL> def test_nested_dict ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] } ) ) , <EOL> ( '<STR_LIT>' , None ) ] <EOL> self . _streamer . consume ( json_input ) <EOL> @ load_test_data <EOL> def test_array ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , "<STR_LIT:a>" ) , <EOL> ( '<STR_LIT>' , <NUM_LIT:2> ) , <EOL> ( '<STR_LIT>' , True ) , <EOL> ( '<STR_LIT>' , { "<STR_LIT>" : "<STR_LIT>" } ) , <EOL> ( '<STR_LIT>' , None ) ] <EOL> self . _streamer . consume ( json_input ) <EOL> @ load_test_data <EOL> def test_spl_chars_in_value ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , <EOL> ( '<STR_LIT>' , <EOL> [ <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> True , <EOL> <NUM_LIT> <EOL> ] <EOL> ) <EOL> ) , <EOL> ( '<STR_LIT>' , None ) ] <EOL> self . _streamer . consume ( json_input ) <EOL> @ load_test_data <EOL> def test_space_preservation ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , None ) <EOL> ] <EOL> self . _streamer . consume ( json_input ) <EOL> @ load_test_data <EOL> def test_arbit_1 ( self , json_input ) : <EOL> self . _assertions = [ ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT:to>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT:type>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : <EOL> { '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT:url>' : '<STR_LIT>' , '<STR_LIT:id>' : <NUM_LIT:5> , '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:value>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:value>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:value>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:value>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT:name>' , '<STR_LIT>' : '<STR_LIT:name>' } , <EOL> { '<STR_LIT:value>' : <NUM_LIT:30> , '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:30> <EOL> } <EOL> } <EOL> ) <EOL> ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , None ) ) , <EOL> ( '<STR_LIT>' , ( '<STR_LIT>' , '<STR_LIT>' ) ) , <EOL> ( '<STR_LIT>' , None ) ] <EOL> self . _streamer . consume ( json_input ) <EOL> class ObjectStreamerListenerTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . _streamer = jsonstreamer . ObjectStreamer ( ) <EOL> def tearDown ( self ) : <EOL> self . _streamer . close ( ) <EOL> self . assertEqual ( len ( self . _assertions ) , <NUM_LIT:0> ) <EOL> @ load_test_data <EOL> def test_on_element ( self , json_input ) : <EOL> self . _assertions = [ "<STR_LIT:a>" , <NUM_LIT:2> , True , { "<STR_LIT>" : "<STR_LIT>" } ] <EOL> def _on_element ( value ) : <EOL> try : <EOL> expected_value = self . _assertions . pop ( <NUM_LIT:0> ) <EOL> except IndexError : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> self . assertEqual ( expected_value , value ) <EOL> self . _streamer . add_listener ( '<STR_LIT>' , _on_element ) <EOL> self . _streamer . consume ( json_input ) <EOL> @ load_test_data <EOL> def test_on_element_multiple_parses ( self , json_input ) : <EOL> self . _assertions = [ "<STR_LIT:a>" , <NUM_LIT:2> , True , { "<STR_LIT>" : "<STR_LIT>" } , ] <EOL> def _on_element ( value ) : <EOL> try : <EOL> expected_value = self . _assertions . pop ( <NUM_LIT:0> ) <EOL> except IndexError : <EOL> raise AssertionError ( '<STR_LIT>' ) <EOL> self . assertEqual ( expected_value , value ) <EOL> self . _streamer . add_listener ( '<STR_LIT>' , _on_element ) <EOL> self . _streamer . consume ( json_input [ <NUM_LIT:0> : <NUM_LIT:8> ] ) <EOL> self . _streamer . consume ( json_input [ <NUM_LIT:8> : ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( verbosity = <NUM_LIT:2> ) </s>
