<s> from sqlalchemy . testing import eq_ , assert_raises_message , assert_raises , in_ , not_in_ , is_ , ne_ , le_ <EOL> from sqlalchemy import testing <EOL> from sqlalchemy . testing import fixtures , engines <EOL> from sqlalchemy import util <EOL> from sqlalchemy import ( <EOL> exc , sql , func , select , String , Integer , MetaData , ForeignKey , <EOL> VARCHAR , INT , CHAR , text , type_coerce , literal_column , <EOL> TypeDecorator , table , column , literal ) <EOL> from sqlalchemy . engine import result as _result <EOL> from sqlalchemy . testing . schema import Table , Column <EOL> import operator <EOL> from sqlalchemy . testing import assertions <EOL> from sqlalchemy import exc as sa_exc <EOL> from sqlalchemy . testing . mock import patch , Mock <EOL> from contextlib import contextmanager <EOL> from sqlalchemy . engine import default <EOL> class ResultProxyTest ( fixtures . TablesTest ) : <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( <EOL> '<STR_LIT>' , INT , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> test_needs_acid = True <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( <EOL> '<STR_LIT>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , Integer , ForeignKey ( '<STR_LIT>' ) ) , <EOL> Column ( '<STR_LIT:address>' , String ( <NUM_LIT:30> ) ) , <EOL> test_needs_acid = True <EOL> ) <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT>' , INT , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> test_needs_acid = True <EOL> ) <EOL> def test_row_iteration ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> { '<STR_LIT>' : <NUM_LIT:7> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:8> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:9> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> ) <EOL> r = users . select ( ) . execute ( ) <EOL> l = [ ] <EOL> for row in r : <EOL> l . append ( row ) <EOL> eq_ ( len ( l ) , <NUM_LIT:3> ) <EOL> @ testing . requires . subqueries <EOL> def test_anonymous_rows ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> { '<STR_LIT>' : <NUM_LIT:7> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:8> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:9> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> ) <EOL> sel = select ( [ users . c . user_id ] ) . where ( users . c . user_name == '<STR_LIT>' ) . as_scalar ( ) <EOL> for row in select ( [ sel + <NUM_LIT:1> , sel + <NUM_LIT:3> ] , bind = users . bind ) . execute ( ) : <EOL> eq_ ( row [ '<STR_LIT>' ] , <NUM_LIT:8> ) <EOL> eq_ ( row [ '<STR_LIT>' ] , <NUM_LIT:10> ) <EOL> def test_row_comparison ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:7> , user_name = '<STR_LIT>' ) <EOL> rp = users . select ( ) . execute ( ) . first ( ) <EOL> eq_ ( rp , rp ) <EOL> is_ ( not ( rp != rp ) , True ) <EOL> equal = ( <NUM_LIT:7> , '<STR_LIT>' ) <EOL> eq_ ( rp , equal ) <EOL> eq_ ( equal , rp ) <EOL> is_ ( ( not ( rp != equal ) ) , True ) <EOL> is_ ( not ( equal != equal ) , True ) <EOL> def endless ( ) : <EOL> while True : <EOL> yield <NUM_LIT:1> <EOL> ne_ ( rp , endless ( ) ) <EOL> ne_ ( endless ( ) , rp ) <EOL> for compare in [ False , <NUM_LIT:8> , endless ( ) , '<STR_LIT>' , ( <NUM_LIT:7> , '<STR_LIT>' ) ] : <EOL> for op in [ <EOL> operator . eq , operator . ne , operator . gt , <EOL> operator . lt , operator . ge , operator . le <EOL> ] : <EOL> try : <EOL> control = op ( equal , compare ) <EOL> except TypeError : <EOL> assert_raises ( TypeError , op , rp , compare ) <EOL> else : <EOL> eq_ ( control , op ( rp , compare ) ) <EOL> try : <EOL> control = op ( compare , equal ) <EOL> except TypeError : <EOL> assert_raises ( TypeError , op , compare , rp ) <EOL> else : <EOL> eq_ ( control , op ( compare , rp ) ) <EOL> @ testing . provide_metadata <EOL> def test_column_label_overlap_fallback ( self ) : <EOL> content = Table ( <EOL> '<STR_LIT:content>' , self . metadata , <EOL> Column ( '<STR_LIT:type>' , String ( <NUM_LIT:30> ) ) , <EOL> ) <EOL> bar = Table ( <EOL> '<STR_LIT:bar>' , self . metadata , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:30> ) ) <EOL> ) <EOL> self . metadata . create_all ( testing . db ) <EOL> testing . db . execute ( content . insert ( ) . values ( type = "<STR_LIT>" ) ) <EOL> row = testing . db . execute ( content . select ( use_labels = True ) ) . first ( ) <EOL> in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> row = testing . db . execute ( <EOL> select ( [ content . c . type . label ( "<STR_LIT>" ) ] ) ) . first ( ) <EOL> in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> row = testing . db . execute ( select ( [ func . now ( ) . label ( "<STR_LIT>" ) ] ) ) . first ( ) <EOL> not_in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> def test_pickled_rows ( self ) : <EOL> users = self . tables . users <EOL> addresses = self . tables . addresses <EOL> users . insert ( ) . execute ( <EOL> { '<STR_LIT>' : <NUM_LIT:7> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:8> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT:9> , '<STR_LIT>' : '<STR_LIT>' } , <EOL> ) <EOL> for pickle in False , True : <EOL> for use_labels in False , True : <EOL> result = users . select ( use_labels = use_labels ) . order_by ( <EOL> users . c . user_id ) . execute ( ) . fetchall ( ) <EOL> if pickle : <EOL> result = util . pickle . loads ( util . pickle . dumps ( result ) ) <EOL> eq_ ( <EOL> result , <EOL> [ ( <NUM_LIT:7> , "<STR_LIT>" ) , ( <NUM_LIT:8> , "<STR_LIT>" ) , ( <NUM_LIT:9> , "<STR_LIT>" ) ] <EOL> ) <EOL> if use_labels : <EOL> eq_ ( result [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <NUM_LIT:7> ) <EOL> eq_ ( <EOL> list ( result [ <NUM_LIT:0> ] . keys ( ) ) , <EOL> [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> else : <EOL> eq_ ( result [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <NUM_LIT:7> ) <EOL> eq_ ( list ( result [ <NUM_LIT:0> ] . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> eq_ ( result [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , <NUM_LIT:7> ) <EOL> eq_ ( result [ <NUM_LIT:0> ] [ users . c . user_id ] , <NUM_LIT:7> ) <EOL> eq_ ( result [ <NUM_LIT:0> ] [ users . c . user_name ] , '<STR_LIT>' ) <EOL> if not pickle or use_labels : <EOL> assert_raises ( <EOL> exc . NoSuchColumnError , <EOL> lambda : result [ <NUM_LIT:0> ] [ addresses . c . user_id ] ) <EOL> else : <EOL> eq_ ( result [ <NUM_LIT:0> ] [ addresses . c . user_id ] , <NUM_LIT:7> ) <EOL> assert_raises ( <EOL> exc . NoSuchColumnError , lambda : result [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> assert_raises ( <EOL> exc . NoSuchColumnError , <EOL> lambda : result [ <NUM_LIT:0> ] [ addresses . c . address_id ] ) <EOL> def test_column_error_printing ( self ) : <EOL> result = testing . db . execute ( select ( [ <NUM_LIT:1> ] ) ) <EOL> row = result . first ( ) <EOL> class unprintable ( object ) : <EOL> def __str__ ( self ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> msg = r"<STR_LIT>" <EOL> for accessor , repl in [ <EOL> ( "<STR_LIT:x>" , "<STR_LIT:x>" ) , <EOL> ( Column ( "<STR_LIT:q>" , Integer ) , "<STR_LIT:q>" ) , <EOL> ( Column ( "<STR_LIT:q>" , Integer ) + <NUM_LIT:12> , r"<STR_LIT>" ) , <EOL> ( unprintable ( ) , "<STR_LIT>" ) , <EOL> ] : <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> msg % repl , <EOL> result . _getter , accessor <EOL> ) <EOL> is_ ( result . _getter ( accessor , False ) , None ) <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> msg % repl , <EOL> lambda : row [ accessor ] <EOL> ) <EOL> def test_fetchmany ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:7> , user_name = '<STR_LIT>' ) <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:8> , user_name = '<STR_LIT>' ) <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:9> , user_name = '<STR_LIT>' ) <EOL> r = users . select ( ) . execute ( ) <EOL> l = [ ] <EOL> for row in r . fetchmany ( size = <NUM_LIT:2> ) : <EOL> l . append ( row ) <EOL> eq_ ( len ( l ) , <NUM_LIT:2> ) <EOL> def test_column_slices ( self ) : <EOL> users = self . tables . users <EOL> addresses = self . tables . addresses <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT>' ) <EOL> addresses . insert ( ) . execute ( <EOL> address_id = <NUM_LIT:1> , user_id = <NUM_LIT:2> , address = '<STR_LIT>' ) <EOL> r = text ( <EOL> "<STR_LIT>" , bind = testing . db ) . execute ( ) . first ( ) <EOL> eq_ ( r [ <NUM_LIT:0> : <NUM_LIT:1> ] , ( <NUM_LIT:1> , ) ) <EOL> eq_ ( r [ <NUM_LIT:1> : ] , ( <NUM_LIT:2> , '<STR_LIT>' ) ) <EOL> eq_ ( r [ : - <NUM_LIT:1> ] , ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> def test_column_accessor_basic_compiled ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> dict ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT>' ) <EOL> ) <EOL> r = users . select ( users . c . user_id == <NUM_LIT:2> ) . execute ( ) . first ( ) <EOL> eq_ ( r . user_id , <NUM_LIT:2> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> eq_ ( r [ users . c . user_id ] , <NUM_LIT:2> ) <EOL> eq_ ( r . user_name , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ users . c . user_name ] , '<STR_LIT>' ) <EOL> def test_column_accessor_basic_text ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> dict ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT>' ) <EOL> ) <EOL> r = testing . db . execute ( <EOL> text ( "<STR_LIT>" ) ) . first ( ) <EOL> eq_ ( r . user_id , <NUM_LIT:2> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> eq_ ( r [ users . c . user_id ] , <NUM_LIT:2> ) <EOL> eq_ ( r . user_name , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ users . c . user_name ] , '<STR_LIT>' ) <EOL> def test_column_accessor_textual_select ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> dict ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT>' ) <EOL> ) <EOL> r = testing . db . execute ( <EOL> select ( [ <EOL> column ( '<STR_LIT>' ) , column ( '<STR_LIT>' ) <EOL> ] ) . select_from ( table ( '<STR_LIT>' ) ) . <EOL> where ( text ( '<STR_LIT>' ) ) <EOL> ) . first ( ) <EOL> eq_ ( r . user_id , <NUM_LIT:2> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> eq_ ( r [ users . c . user_id ] , <NUM_LIT:2> ) <EOL> eq_ ( r . user_name , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ users . c . user_name ] , '<STR_LIT>' ) <EOL> def test_column_accessor_dotted_union ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> r = testing . db . execute ( <EOL> text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> ) . first ( ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> eq_ ( list ( r . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_column_accessor_sqlite_raw ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> r = text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bind = testing . db ) . execution_options ( sqlite_raw_colnames = True ) . execute ( ) . first ( ) <EOL> if testing . against ( "<STR_LIT>" ) : <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> eq_ ( list ( r . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> else : <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> eq_ ( list ( r . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_column_accessor_sqlite_translated ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> r = text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> bind = testing . db ) . execute ( ) . first ( ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> if testing . against ( "<STR_LIT>" ) : <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> else : <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> not_in_ ( '<STR_LIT>' , r ) <EOL> eq_ ( list ( r . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_column_accessor_labels_w_dots ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> r = text ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , bind = testing . db ) . execution_options ( sqlite_raw_colnames = True ) . execute ( ) . first ( ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , "<STR_LIT>" ) <EOL> not_in_ ( "<STR_LIT>" , r ) <EOL> eq_ ( list ( r . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_column_accessor_unary ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> r = select ( [ users . c . user_name . distinct ( ) ] ) . order_by ( <EOL> users . c . user_name ) . execute ( ) . first ( ) <EOL> eq_ ( r [ users . c . user_name ] , '<STR_LIT>' ) <EOL> eq_ ( r . user_name , '<STR_LIT>' ) <EOL> def test_column_accessor_err ( self ) : <EOL> r = testing . db . execute ( select ( [ <NUM_LIT:1> ] ) ) . first ( ) <EOL> assert_raises_message ( <EOL> AttributeError , <EOL> "<STR_LIT>" , <EOL> getattr , r , "<STR_LIT:foo>" <EOL> ) <EOL> assert_raises_message ( <EOL> KeyError , <EOL> "<STR_LIT>" , <EOL> lambda : r [ '<STR_LIT:foo>' ] <EOL> ) <EOL> def test_graceful_fetch_on_non_rows ( self ) : <EOL> """<STR_LIT>""" <EOL> users = self . tables . users <EOL> conn = testing . db . connect ( ) <EOL> for meth in [ <EOL> lambda r : r . fetchone ( ) , <EOL> lambda r : r . fetchall ( ) , <EOL> lambda r : r . first ( ) , <EOL> lambda r : r . scalar ( ) , <EOL> lambda r : r . fetchmany ( ) , <EOL> lambda r : r . _getter ( '<STR_LIT:user>' ) , <EOL> lambda r : r . _has_key ( '<STR_LIT:user>' ) , <EOL> ] : <EOL> trans = conn . begin ( ) <EOL> result = conn . execute ( users . insert ( ) , user_id = <NUM_LIT:1> ) <EOL> assert_raises_message ( <EOL> exc . ResourceClosedError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> meth , result , <EOL> ) <EOL> trans . rollback ( ) <EOL> def test_fetchone_til_end ( self ) : <EOL> result = testing . db . execute ( "<STR_LIT>" ) <EOL> eq_ ( result . fetchone ( ) , None ) <EOL> eq_ ( result . fetchone ( ) , None ) <EOL> eq_ ( result . fetchone ( ) , None ) <EOL> result . close ( ) <EOL> assert_raises_message ( <EOL> exc . ResourceClosedError , <EOL> "<STR_LIT>" , <EOL> result . fetchone <EOL> ) <EOL> def test_row_case_sensitive ( self ) : <EOL> row = testing . db . execute ( <EOL> select ( [ <EOL> literal_column ( "<STR_LIT:1>" ) . label ( "<STR_LIT>" ) , <EOL> literal_column ( "<STR_LIT:2>" ) . label ( "<STR_LIT>" ) <EOL> ] ) <EOL> ) . first ( ) <EOL> eq_ ( list ( row . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> not_in_ ( "<STR_LIT>" , row . _keymap ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> assert_raises ( <EOL> KeyError , <EOL> lambda : row [ "<STR_LIT>" ] <EOL> ) <EOL> assert_raises ( <EOL> KeyError , <EOL> lambda : row [ "<STR_LIT>" ] <EOL> ) <EOL> def test_row_case_sensitive_unoptimized ( self ) : <EOL> ins_db = engines . testing_engine ( options = { "<STR_LIT>" : True } ) <EOL> row = ins_db . execute ( <EOL> select ( [ <EOL> literal_column ( "<STR_LIT:1>" ) . label ( "<STR_LIT>" ) , <EOL> literal_column ( "<STR_LIT:2>" ) . label ( "<STR_LIT>" ) , <EOL> text ( "<STR_LIT>" ) <EOL> ] ) <EOL> ) . first ( ) <EOL> eq_ ( <EOL> list ( row . keys ( ) ) , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> not_in_ ( "<STR_LIT>" , row . _keymap ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:3> ) <EOL> assert_raises ( KeyError , lambda : row [ "<STR_LIT>" ] ) <EOL> assert_raises ( KeyError , lambda : row [ "<STR_LIT>" ] ) <EOL> assert_raises ( KeyError , lambda : row [ "<STR_LIT>" ] ) <EOL> def test_row_case_insensitive ( self ) : <EOL> ins_db = engines . testing_engine ( options = { "<STR_LIT>" : False } ) <EOL> row = ins_db . execute ( <EOL> select ( [ <EOL> literal_column ( "<STR_LIT:1>" ) . label ( "<STR_LIT>" ) , <EOL> literal_column ( "<STR_LIT:2>" ) . label ( "<STR_LIT>" ) <EOL> ] ) <EOL> ) . first ( ) <EOL> eq_ ( list ( row . keys ( ) ) , [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> def test_row_case_insensitive_unoptimized ( self ) : <EOL> ins_db = engines . testing_engine ( options = { "<STR_LIT>" : False } ) <EOL> row = ins_db . execute ( <EOL> select ( [ <EOL> literal_column ( "<STR_LIT:1>" ) . label ( "<STR_LIT>" ) , <EOL> literal_column ( "<STR_LIT:2>" ) . label ( "<STR_LIT>" ) , <EOL> text ( "<STR_LIT>" ) <EOL> ] ) <EOL> ) . first ( ) <EOL> eq_ ( <EOL> list ( row . keys ( ) ) , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> in_ ( "<STR_LIT>" , row . _keymap ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:3> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:2> ) <EOL> eq_ ( row [ "<STR_LIT>" ] , <NUM_LIT:3> ) <EOL> def test_row_as_args ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> r = users . select ( users . c . user_id == <NUM_LIT:1> ) . execute ( ) . first ( ) <EOL> users . delete ( ) . execute ( ) <EOL> users . insert ( ) . execute ( r ) <EOL> eq_ ( users . select ( ) . execute ( ) . fetchall ( ) , [ ( <NUM_LIT:1> , '<STR_LIT>' ) ] ) <EOL> def test_result_as_args ( self ) : <EOL> users = self . tables . users <EOL> users2 = self . tables . users2 <EOL> users . insert ( ) . execute ( [ <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) , <EOL> dict ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT>' ) ] ) <EOL> r = users . select ( ) . execute ( ) <EOL> users2 . insert ( ) . execute ( list ( r ) ) <EOL> eq_ ( <EOL> users2 . select ( ) . order_by ( users2 . c . user_id ) . execute ( ) . fetchall ( ) , <EOL> [ ( <NUM_LIT:1> , '<STR_LIT>' ) , ( <NUM_LIT:2> , '<STR_LIT>' ) ] <EOL> ) <EOL> users2 . delete ( ) . execute ( ) <EOL> r = users . select ( ) . execute ( ) <EOL> users2 . insert ( ) . execute ( * list ( r ) ) <EOL> eq_ ( <EOL> users2 . select ( ) . order_by ( users2 . c . user_id ) . execute ( ) . fetchall ( ) , <EOL> [ ( <NUM_LIT:1> , '<STR_LIT>' ) , ( <NUM_LIT:2> , '<STR_LIT>' ) ] <EOL> ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_ambiguous_column ( self ) : <EOL> users = self . tables . users <EOL> addresses = self . tables . addresses <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> result = users . outerjoin ( addresses ) . select ( ) . execute ( ) <EOL> r = result . first ( ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : r [ '<STR_LIT>' ] <EOL> ) <EOL> eq_ ( r [ users . c . user_id ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ addresses . c . user_id ] , None ) <EOL> fake_table = Table ( '<STR_LIT>' , MetaData ( ) , Column ( '<STR_LIT>' , Integer ) ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : r [ fake_table . c . user_id ] <EOL> ) <EOL> r = util . pickle . loads ( util . pickle . dumps ( r ) ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : r [ '<STR_LIT>' ] <EOL> ) <EOL> result = users . outerjoin ( addresses ) . select ( ) . execute ( ) <EOL> result = _result . BufferedColumnResultProxy ( result . context ) <EOL> r = result . first ( ) <EOL> assert isinstance ( r , _result . BufferedColumnRow ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : r [ '<STR_LIT>' ] <EOL> ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_ambiguous_column_by_col ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> ua = users . alias ( ) <EOL> u2 = users . alias ( ) <EOL> result = select ( [ users . c . user_id , ua . c . user_id ] ) . execute ( ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ users . c . user_id ] , <NUM_LIT:1> ) <EOL> eq_ ( row [ ua . c . user_id ] , <NUM_LIT:1> ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ u2 . c . user_id ] <EOL> ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_ambiguous_column_case_sensitive ( self ) : <EOL> eng = engines . testing_engine ( options = dict ( case_sensitive = False ) ) <EOL> row = eng . execute ( select ( [ <EOL> literal_column ( '<STR_LIT:1>' ) . label ( '<STR_LIT>' ) , <EOL> literal_column ( '<STR_LIT:1>' ) . label ( '<STR_LIT>' ) , <EOL> ] ) ) . first ( ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ '<STR_LIT>' ] <EOL> ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_ambiguous_column_contains ( self ) : <EOL> users = self . tables . users <EOL> addresses = self . tables . addresses <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> result = select ( [ users . c . user_id , addresses . c . user_id ] ) . select_from ( users . outerjoin ( addresses ) ) . execute ( ) <EOL> row = result . first ( ) <EOL> eq_ ( <EOL> set ( [ users . c . user_id in row , addresses . c . user_id in row ] ) , <EOL> set ( [ True ] ) <EOL> ) <EOL> def test_ambiguous_column_by_col_plus_label ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT>' ) <EOL> result = select ( <EOL> [ users . c . user_id , <EOL> type_coerce ( users . c . user_id , Integer ) . label ( '<STR_LIT:foo>' ) ] ) . execute ( ) <EOL> row = result . first ( ) <EOL> eq_ ( <EOL> row [ users . c . user_id ] , <NUM_LIT:1> <EOL> ) <EOL> eq_ ( <EOL> row [ <NUM_LIT:1> ] , <NUM_LIT:1> <EOL> ) <EOL> def test_fetch_partial_result_map ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:7> , user_name = '<STR_LIT>' ) <EOL> t = text ( "<STR_LIT>" ) . columns ( <EOL> user_name = String ( ) <EOL> ) <EOL> eq_ ( <EOL> testing . db . execute ( t ) . fetchall ( ) , [ ( <NUM_LIT:7> , '<STR_LIT>' ) ] <EOL> ) <EOL> def test_fetch_unordered_result_map ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:7> , user_name = '<STR_LIT>' ) <EOL> class Goofy1 ( TypeDecorator ) : <EOL> impl = String <EOL> def process_result_value ( self , value , dialect ) : <EOL> return value + "<STR_LIT:a>" <EOL> class Goofy2 ( TypeDecorator ) : <EOL> impl = String <EOL> def process_result_value ( self , value , dialect ) : <EOL> return value + "<STR_LIT:b>" <EOL> class Goofy3 ( TypeDecorator ) : <EOL> impl = String <EOL> def process_result_value ( self , value , dialect ) : <EOL> return value + "<STR_LIT:c>" <EOL> t = text ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) . columns ( <EOL> a = Goofy1 ( ) , b = Goofy2 ( ) , c = Goofy3 ( ) <EOL> ) <EOL> eq_ ( <EOL> testing . db . execute ( t ) . fetchall ( ) , [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> ] <EOL> ) <EOL> @ testing . requires . subqueries <EOL> def test_column_label_targeting ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:7> , user_name = '<STR_LIT>' ) <EOL> for s in ( <EOL> users . select ( ) . alias ( '<STR_LIT:foo>' ) , <EOL> users . select ( ) . alias ( users . name ) , <EOL> ) : <EOL> row = s . select ( use_labels = True ) . execute ( ) . first ( ) <EOL> eq_ ( row [ s . c . user_id ] , <NUM_LIT:7> ) <EOL> eq_ ( row [ s . c . user_name ] , '<STR_LIT>' ) <EOL> def test_keys ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> result = users . select ( ) . execute ( ) <EOL> eq_ ( <EOL> result . keys ( ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> row = result . first ( ) <EOL> eq_ ( <EOL> row . keys ( ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_keys_anon_labels ( self ) : <EOL> """<STR_LIT>""" <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> result = testing . db . execute ( <EOL> select ( [ <EOL> users . c . user_id , <EOL> users . c . user_name . label ( None ) , <EOL> func . count ( literal_column ( '<STR_LIT:1>' ) ) ] ) . <EOL> group_by ( users . c . user_id , users . c . user_name ) <EOL> ) <EOL> eq_ ( <EOL> result . keys ( ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> row = result . first ( ) <EOL> eq_ ( <EOL> row . keys ( ) , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_items ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> r = users . select ( ) . execute ( ) . first ( ) <EOL> eq_ ( <EOL> [ ( x [ <NUM_LIT:0> ] . lower ( ) , x [ <NUM_LIT:1> ] ) for x in list ( r . items ( ) ) ] , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:1> ) , ( '<STR_LIT>' , '<STR_LIT:foo>' ) ] ) <EOL> def test_len ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> r = users . select ( ) . execute ( ) . first ( ) <EOL> eq_ ( len ( r ) , <NUM_LIT:2> ) <EOL> r = testing . db . execute ( '<STR_LIT>' ) . first ( ) <EOL> eq_ ( len ( r ) , <NUM_LIT:2> ) <EOL> r = testing . db . execute ( '<STR_LIT>' ) . first ( ) <EOL> eq_ ( len ( r ) , <NUM_LIT:1> ) <EOL> def test_sorting_in_python ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( <EOL> dict ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) , <EOL> dict ( user_id = <NUM_LIT:2> , user_name = '<STR_LIT:bar>' ) , <EOL> dict ( user_id = <NUM_LIT:3> , user_name = '<STR_LIT>' ) , <EOL> ) <EOL> rows = users . select ( ) . order_by ( users . c . user_name ) . execute ( ) . fetchall ( ) <EOL> eq_ ( rows , [ ( <NUM_LIT:2> , '<STR_LIT:bar>' ) , ( <NUM_LIT:3> , '<STR_LIT>' ) , ( <NUM_LIT:1> , '<STR_LIT:foo>' ) ] ) <EOL> eq_ ( sorted ( rows ) , [ ( <NUM_LIT:1> , '<STR_LIT:foo>' ) , ( <NUM_LIT:2> , '<STR_LIT:bar>' ) , ( <NUM_LIT:3> , '<STR_LIT>' ) ] ) <EOL> def test_column_order_with_simple_query ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> r = users . select ( users . c . user_id == <NUM_LIT:1> ) . execute ( ) . first ( ) <EOL> eq_ ( r [ <NUM_LIT:0> ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ <NUM_LIT:1> ] , '<STR_LIT:foo>' ) <EOL> eq_ ( [ x . lower ( ) for x in list ( r . keys ( ) ) ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> eq_ ( list ( r . values ( ) ) , [ <NUM_LIT:1> , '<STR_LIT:foo>' ] ) <EOL> def test_column_order_with_text_query ( self ) : <EOL> users = self . tables . users <EOL> users . insert ( ) . execute ( user_id = <NUM_LIT:1> , user_name = '<STR_LIT:foo>' ) <EOL> r = testing . db . execute ( '<STR_LIT>' ) . first ( ) <EOL> eq_ ( r [ <NUM_LIT:0> ] , '<STR_LIT:foo>' ) <EOL> eq_ ( r [ <NUM_LIT:1> ] , <NUM_LIT:1> ) <EOL> eq_ ( [ x . lower ( ) for x in list ( r . keys ( ) ) ] , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> eq_ ( list ( r . values ( ) ) , [ '<STR_LIT:foo>' , <NUM_LIT:1> ] ) <EOL> @ testing . crashes ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ testing . crashes ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ testing . provide_metadata <EOL> def test_column_accessor_shadow ( self ) : <EOL> shadowed = Table ( <EOL> '<STR_LIT>' , self . metadata , <EOL> Column ( '<STR_LIT>' , INT , primary_key = True ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT> ) ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> Column ( '<STR_LIT>' , VARCHAR ( <NUM_LIT:20> ) ) , <EOL> ) <EOL> self . metadata . create_all ( ) <EOL> shadowed . insert ( ) . execute ( <EOL> shadow_id = <NUM_LIT:1> , shadow_name = '<STR_LIT>' , parent = '<STR_LIT>' , <EOL> row = '<STR_LIT>' , <EOL> _parent = '<STR_LIT>' , _row = '<STR_LIT>' ) <EOL> r = shadowed . select ( shadowed . c . shadow_id == <NUM_LIT:1> ) . execute ( ) . first ( ) <EOL> eq_ ( r . shadow_id , <NUM_LIT:1> ) <EOL> eq_ ( r [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> eq_ ( r [ shadowed . c . shadow_id ] , <NUM_LIT:1> ) <EOL> eq_ ( r . shadow_name , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ shadowed . c . shadow_name ] , '<STR_LIT>' ) <EOL> eq_ ( r . parent , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ shadowed . c . parent ] , '<STR_LIT>' ) <EOL> eq_ ( r . row , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ shadowed . c . row ] , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> eq_ ( r [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_nontuple_row ( self ) : <EOL> """<STR_LIT>""" <EOL> from sqlalchemy . engine import RowProxy <EOL> class MyList ( object ) : <EOL> def __init__ ( self , l ) : <EOL> self . l = l <EOL> def __len__ ( self ) : <EOL> return len ( self . l ) <EOL> def __getitem__ ( self , i ) : <EOL> return list . __getitem__ ( self . l , i ) <EOL> proxy = RowProxy ( object ( ) , MyList ( [ '<STR_LIT:value>' ] ) , [ None ] , { <EOL> '<STR_LIT:key>' : ( None , None , <NUM_LIT:0> ) , <NUM_LIT:0> : ( None , None , <NUM_LIT:0> ) } ) <EOL> eq_ ( list ( proxy ) , [ '<STR_LIT:value>' ] ) <EOL> eq_ ( proxy [ <NUM_LIT:0> ] , '<STR_LIT:value>' ) <EOL> eq_ ( proxy [ '<STR_LIT:key>' ] , '<STR_LIT:value>' ) <EOL> @ testing . provide_metadata <EOL> def test_no_rowcount_on_selects_inserts ( self ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . metadata <EOL> engine = engines . testing_engine ( ) <EOL> t = Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:data>' , String ( <NUM_LIT:10> ) ) <EOL> ) <EOL> metadata . create_all ( engine ) <EOL> with patch . object ( <EOL> engine . dialect . execution_ctx_cls , "<STR_LIT>" ) as mock_rowcount : <EOL> mock_rowcount . __get__ = Mock ( ) <EOL> engine . execute ( t . insert ( ) , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:data>' : '<STR_LIT>' } ) <EOL> eq_ ( len ( mock_rowcount . __get__ . mock_calls ) , <NUM_LIT:0> ) <EOL> eq_ ( <EOL> engine . execute ( t . select ( ) ) . fetchall ( ) , <EOL> [ ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) , ( '<STR_LIT>' , ) ] <EOL> ) <EOL> eq_ ( len ( mock_rowcount . __get__ . mock_calls ) , <NUM_LIT:0> ) <EOL> engine . execute ( t . update ( ) , { '<STR_LIT:data>' : '<STR_LIT>' } ) <EOL> eq_ ( len ( mock_rowcount . __get__ . mock_calls ) , <NUM_LIT:1> ) <EOL> engine . execute ( t . delete ( ) ) <EOL> eq_ ( len ( mock_rowcount . __get__ . mock_calls ) , <NUM_LIT:2> ) <EOL> def test_rowproxy_is_sequence ( self ) : <EOL> import collections <EOL> from sqlalchemy . engine import RowProxy <EOL> row = RowProxy ( <EOL> object ( ) , [ '<STR_LIT:value>' ] , [ None ] , <EOL> { '<STR_LIT:key>' : ( None , None , <NUM_LIT:0> ) , <NUM_LIT:0> : ( None , None , <NUM_LIT:0> ) } ) <EOL> assert isinstance ( row , collections . Sequence ) <EOL> @ testing . provide_metadata <EOL> def test_rowproxy_getitem_indexes_compiled ( self ) : <EOL> values = Table ( '<STR_LIT>' , self . metadata , <EOL> Column ( '<STR_LIT:key>' , String ( <NUM_LIT:10> ) , primary_key = True ) , <EOL> Column ( '<STR_LIT:value>' , String ( <NUM_LIT:10> ) ) ) <EOL> values . create ( ) <EOL> testing . db . execute ( values . insert ( ) , dict ( key = '<STR_LIT>' , value = '<STR_LIT>' ) ) <EOL> row = testing . db . execute ( values . select ( ) ) . first ( ) <EOL> eq_ ( row [ '<STR_LIT:key>' ] , '<STR_LIT>' ) <EOL> eq_ ( row [ '<STR_LIT:value>' ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ - <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ - <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:1> : <NUM_LIT:0> : - <NUM_LIT:1> ] , ( '<STR_LIT>' , ) ) <EOL> @ testing . only_on ( "<STR_LIT>" ) <EOL> def test_rowproxy_getitem_indexes_raw ( self ) : <EOL> row = testing . db . execute ( "<STR_LIT>" ) . first ( ) <EOL> eq_ ( row [ '<STR_LIT:key>' ] , '<STR_LIT>' ) <EOL> eq_ ( row [ '<STR_LIT:value>' ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ - <NUM_LIT:2> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ - <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> eq_ ( row [ <NUM_LIT:1> : <NUM_LIT:0> : - <NUM_LIT:1> ] , ( '<STR_LIT>' , ) ) <EOL> @ testing . requires . cextensions <EOL> def test_row_c_sequence_check ( self ) : <EOL> import csv <EOL> metadata = MetaData ( ) <EOL> metadata . bind = '<STR_LIT>' <EOL> users = Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:name>' , String ( <NUM_LIT> ) ) , <EOL> ) <EOL> users . create ( ) <EOL> users . insert ( ) . execute ( name = '<STR_LIT>' ) <EOL> row = users . select ( ) . execute ( ) . fetchone ( ) <EOL> s = util . StringIO ( ) <EOL> writer = csv . writer ( s ) <EOL> writer . writerow ( row ) <EOL> assert s . getvalue ( ) . strip ( ) == '<STR_LIT>' <EOL> @ testing . requires . selectone <EOL> def test_empty_accessors ( self ) : <EOL> statements = [ <EOL> ( <EOL> "<STR_LIT>" , <EOL> [ <EOL> lambda r : r . last_inserted_params ( ) , <EOL> lambda r : r . last_updated_params ( ) , <EOL> lambda r : r . prefetch_cols ( ) , <EOL> lambda r : r . postfetch_cols ( ) , <EOL> lambda r : r . inserted_primary_key <EOL> ] , <EOL> "<STR_LIT>" <EOL> ) , <EOL> ( <EOL> select ( [ <NUM_LIT:1> ] ) , <EOL> [ <EOL> lambda r : r . last_inserted_params ( ) , <EOL> lambda r : r . inserted_primary_key <EOL> ] , <EOL> r"<STR_LIT>" <EOL> ) , <EOL> ( <EOL> select ( [ <NUM_LIT:1> ] ) , <EOL> [ <EOL> lambda r : r . last_updated_params ( ) , <EOL> ] , <EOL> r"<STR_LIT>" <EOL> ) , <EOL> ( <EOL> select ( [ <NUM_LIT:1> ] ) , <EOL> [ <EOL> lambda r : r . prefetch_cols ( ) , <EOL> lambda r : r . postfetch_cols ( ) <EOL> ] , <EOL> r"<STR_LIT>" <EOL> r"<STR_LIT>" <EOL> ) , <EOL> ] <EOL> for stmt , meths , msg in statements : <EOL> r = testing . db . execute ( stmt ) <EOL> try : <EOL> for meth in meths : <EOL> assert_raises_message ( <EOL> sa_exc . InvalidRequestError , <EOL> msg , <EOL> meth , r <EOL> ) <EOL> finally : <EOL> r . close ( ) <EOL> class KeyTargetingTest ( fixtures . TablesTest ) : <EOL> run_inserts = '<STR_LIT>' <EOL> run_deletes = None <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT>' , metadata , Column ( "<STR_LIT:a>" , CHAR ( <NUM_LIT:2> ) , key = "<STR_LIT:b>" ) , <EOL> Column ( "<STR_LIT:c>" , CHAR ( <NUM_LIT:2> ) , key = "<STR_LIT:q>" ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , Column ( "<STR_LIT:a>" , CHAR ( <NUM_LIT:2> ) ) , Column ( "<STR_LIT:b>" , CHAR ( <NUM_LIT:2> ) ) ) <EOL> Table ( '<STR_LIT>' , metadata , Column ( "<STR_LIT:a>" , CHAR ( <NUM_LIT:2> ) ) , Column ( "<STR_LIT:d>" , CHAR ( <NUM_LIT:2> ) ) ) <EOL> Table ( '<STR_LIT>' , metadata , Column ( "<STR_LIT:b>" , CHAR ( <NUM_LIT:2> ) ) , Column ( "<STR_LIT:q>" , CHAR ( <NUM_LIT:2> ) ) ) <EOL> Table ( '<STR_LIT:content>' , metadata , Column ( '<STR_LIT:t>' , String ( <NUM_LIT:30> ) , key = "<STR_LIT:type>" ) ) <EOL> Table ( '<STR_LIT:bar>' , metadata , Column ( '<STR_LIT>' , String ( <NUM_LIT:30> ) , key = "<STR_LIT>" ) ) <EOL> if testing . requires . schemas . enabled : <EOL> Table ( <EOL> '<STR_LIT>' , metadata , <EOL> Column ( "<STR_LIT:a>" , CHAR ( <NUM_LIT:2> ) , key = "<STR_LIT:b>" ) , <EOL> Column ( "<STR_LIT:c>" , CHAR ( <NUM_LIT:2> ) , key = "<STR_LIT:q>" ) , <EOL> schema = testing . config . test_schema <EOL> ) <EOL> @ classmethod <EOL> def insert_data ( cls ) : <EOL> cls . tables . keyed1 . insert ( ) . execute ( dict ( b = "<STR_LIT>" , q = "<STR_LIT>" ) ) <EOL> cls . tables . keyed2 . insert ( ) . execute ( dict ( a = "<STR_LIT>" , b = "<STR_LIT>" ) ) <EOL> cls . tables . keyed3 . insert ( ) . execute ( dict ( a = "<STR_LIT>" , d = "<STR_LIT>" ) ) <EOL> cls . tables . keyed4 . insert ( ) . execute ( dict ( b = "<STR_LIT>" , q = "<STR_LIT>" ) ) <EOL> cls . tables . content . insert ( ) . execute ( type = "<STR_LIT>" ) <EOL> if testing . requires . schemas . enabled : <EOL> cls . tables [ <EOL> '<STR_LIT>' % testing . config . test_schema ] . insert ( ) . execute ( <EOL> dict ( b = "<STR_LIT>" , q = "<STR_LIT>" ) ) <EOL> @ testing . requires . schemas <EOL> def test_keyed_accessor_wschema ( self ) : <EOL> keyed1 = self . tables [ '<STR_LIT>' % testing . config . test_schema ] <EOL> row = testing . db . execute ( keyed1 . select ( ) ) . first ( ) <EOL> eq_ ( row . b , "<STR_LIT>" ) <EOL> eq_ ( row . q , "<STR_LIT>" ) <EOL> eq_ ( row . a , "<STR_LIT>" ) <EOL> eq_ ( row . c , "<STR_LIT>" ) <EOL> def test_keyed_accessor_single ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> row = testing . db . execute ( keyed1 . select ( ) ) . first ( ) <EOL> eq_ ( row . b , "<STR_LIT>" ) <EOL> eq_ ( row . q , "<STR_LIT>" ) <EOL> eq_ ( row . a , "<STR_LIT>" ) <EOL> eq_ ( row . c , "<STR_LIT>" ) <EOL> def test_keyed_accessor_single_labeled ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> row = testing . db . execute ( keyed1 . select ( ) . apply_labels ( ) ) . first ( ) <EOL> eq_ ( row . keyed1_b , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_q , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_a , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_c , "<STR_LIT>" ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_keyed_accessor_composite_conflict_2 ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> keyed2 = self . tables . keyed2 <EOL> row = testing . db . execute ( select ( [ keyed1 , keyed2 ] ) ) . first ( ) <EOL> eq_ ( row . b , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> getattr , row , "<STR_LIT:a>" <EOL> ) <EOL> def test_keyed_accessor_composite_names_precedent ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> keyed4 = self . tables . keyed4 <EOL> row = testing . db . execute ( select ( [ keyed1 , keyed4 ] ) ) . first ( ) <EOL> eq_ ( row . b , "<STR_LIT>" ) <EOL> eq_ ( row . q , "<STR_LIT>" ) <EOL> eq_ ( row . a , "<STR_LIT>" ) <EOL> eq_ ( row . c , "<STR_LIT>" ) <EOL> @ testing . requires . duplicate_names_in_cursor_description <EOL> def test_keyed_accessor_composite_keys_precedent ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> keyed3 = self . tables . keyed3 <EOL> row = testing . db . execute ( select ( [ keyed1 , keyed3 ] ) ) . first ( ) <EOL> eq_ ( row . q , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> getattr , row , "<STR_LIT:b>" <EOL> ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" , <EOL> getattr , row , "<STR_LIT:a>" <EOL> ) <EOL> eq_ ( row . d , "<STR_LIT>" ) <EOL> def test_keyed_accessor_composite_labeled ( self ) : <EOL> keyed1 = self . tables . keyed1 <EOL> keyed2 = self . tables . keyed2 <EOL> row = testing . db . execute ( select ( [ keyed1 , keyed2 ] ) . apply_labels ( ) ) . first ( ) <EOL> eq_ ( row . keyed1_b , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_a , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_q , "<STR_LIT>" ) <EOL> eq_ ( row . keyed1_c , "<STR_LIT>" ) <EOL> eq_ ( row . keyed2_a , "<STR_LIT>" ) <EOL> eq_ ( row . keyed2_b , "<STR_LIT>" ) <EOL> assert_raises ( KeyError , lambda : row [ '<STR_LIT>' ] ) <EOL> assert_raises ( KeyError , lambda : row [ '<STR_LIT>' ] ) <EOL> def test_column_label_overlap_fallback ( self ) : <EOL> content , bar = self . tables . content , self . tables . bar <EOL> row = testing . db . execute ( <EOL> select ( [ content . c . type . label ( "<STR_LIT>" ) ] ) ) . first ( ) <EOL> not_in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> row = testing . db . execute ( select ( [ func . now ( ) . label ( "<STR_LIT>" ) ] ) ) . first ( ) <EOL> not_in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> def test_column_label_overlap_fallback_2 ( self ) : <EOL> content , bar = self . tables . content , self . tables . bar <EOL> row = testing . db . execute ( content . select ( use_labels = True ) ) . first ( ) <EOL> in_ ( content . c . type , row ) <EOL> not_in_ ( bar . c . content_type , row ) <EOL> not_in_ ( sql . column ( '<STR_LIT>' ) , row ) <EOL> def test_columnclause_schema_column_one ( self ) : <EOL> keyed2 = self . tables . keyed2 <EOL> a , b = sql . column ( '<STR_LIT:a>' ) , sql . column ( '<STR_LIT:b>' ) <EOL> stmt = select ( [ a , b ] ) . select_from ( table ( "<STR_LIT>" ) ) <EOL> row = testing . db . execute ( stmt ) . first ( ) <EOL> in_ ( keyed2 . c . a , row ) <EOL> in_ ( keyed2 . c . b , row ) <EOL> in_ ( a , row ) <EOL> in_ ( b , row ) <EOL> def test_columnclause_schema_column_two ( self ) : <EOL> keyed2 = self . tables . keyed2 <EOL> a , b = sql . column ( '<STR_LIT:a>' ) , sql . column ( '<STR_LIT:b>' ) <EOL> stmt = select ( [ keyed2 . c . a , keyed2 . c . b ] ) <EOL> row = testing . db . execute ( stmt ) . first ( ) <EOL> in_ ( keyed2 . c . a , row ) <EOL> in_ ( keyed2 . c . b , row ) <EOL> in_ ( a , row ) <EOL> in_ ( b , row ) <EOL> def test_columnclause_schema_column_three ( self ) : <EOL> keyed2 = self . tables . keyed2 <EOL> a , b = sql . column ( '<STR_LIT:a>' ) , sql . column ( '<STR_LIT:b>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( a = CHAR , b = CHAR ) <EOL> row = testing . db . execute ( stmt ) . first ( ) <EOL> in_ ( keyed2 . c . a , row ) <EOL> in_ ( keyed2 . c . b , row ) <EOL> in_ ( a , row ) <EOL> in_ ( b , row ) <EOL> in_ ( stmt . c . a , row ) <EOL> in_ ( stmt . c . b , row ) <EOL> def test_columnclause_schema_column_four ( self ) : <EOL> keyed2 = self . tables . keyed2 <EOL> a , b = sql . column ( '<STR_LIT>' ) , sql . column ( '<STR_LIT>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( <EOL> a , b ) <EOL> row = testing . db . execute ( stmt ) . first ( ) <EOL> in_ ( keyed2 . c . a , row ) <EOL> in_ ( keyed2 . c . b , row ) <EOL> in_ ( a , row ) <EOL> in_ ( b , row ) <EOL> in_ ( stmt . c . keyed2_a , row ) <EOL> in_ ( stmt . c . keyed2_b , row ) <EOL> def test_columnclause_schema_column_five ( self ) : <EOL> keyed2 = self . tables . keyed2 <EOL> stmt = text ( "<STR_LIT>" ) . columns ( <EOL> keyed2_a = CHAR , keyed2_b = CHAR ) <EOL> row = testing . db . execute ( stmt ) . first ( ) <EOL> in_ ( keyed2 . c . a , row ) <EOL> in_ ( keyed2 . c . b , row ) <EOL> in_ ( stmt . c . keyed2_a , row ) <EOL> in_ ( stmt . c . keyed2_b , row ) <EOL> class PositionalTextTest ( fixtures . TablesTest ) : <EOL> run_inserts = '<STR_LIT>' <EOL> run_deletes = None <EOL> __backend__ = True <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT>' , <EOL> metadata , <EOL> Column ( "<STR_LIT:a>" , CHAR ( <NUM_LIT:2> ) ) , <EOL> Column ( "<STR_LIT:b>" , CHAR ( <NUM_LIT:2> ) ) , <EOL> Column ( "<STR_LIT:c>" , CHAR ( <NUM_LIT:2> ) ) , <EOL> Column ( "<STR_LIT:d>" , CHAR ( <NUM_LIT:2> ) ) <EOL> ) <EOL> @ classmethod <EOL> def insert_data ( cls ) : <EOL> cls . tables . text1 . insert ( ) . execute ( [ <EOL> dict ( a = "<STR_LIT>" , b = "<STR_LIT>" , c = "<STR_LIT>" , d = "<STR_LIT>" ) , <EOL> ] ) <EOL> def test_via_column ( self ) : <EOL> c1 , c2 , c3 , c4 = column ( '<STR_LIT:q>' ) , column ( '<STR_LIT:p>' ) , column ( '<STR_LIT:r>' ) , column ( '<STR_LIT:d>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 , c3 , c4 ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c2 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c4 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ <NUM_LIT:1> ] , "<STR_LIT>" ) <EOL> eq_ ( row [ "<STR_LIT:b>" ] , "<STR_LIT>" ) <EOL> eq_ ( row . keys ( ) , [ "<STR_LIT:a>" , "<STR_LIT:b>" , "<STR_LIT:c>" , "<STR_LIT:d>" ] ) <EOL> eq_ ( row [ "<STR_LIT:r>" ] , "<STR_LIT>" ) <EOL> eq_ ( row [ "<STR_LIT:d>" ] , "<STR_LIT>" ) <EOL> def test_fewer_cols_than_sql_positional ( self ) : <EOL> c1 , c2 = column ( '<STR_LIT:q>' ) , column ( '<STR_LIT:p>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c1 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ "<STR_LIT:c>" ] , "<STR_LIT>" ) <EOL> def test_fewer_cols_than_sql_non_positional ( self ) : <EOL> c1 , c2 = column ( '<STR_LIT:a>' ) , column ( '<STR_LIT:p>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c2 , c1 , d = CHAR ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c1 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ "<STR_LIT:c>" ] , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ c2 ] <EOL> ) <EOL> def test_more_cols_than_sql ( self ) : <EOL> c1 , c2 , c3 , c4 = column ( '<STR_LIT:q>' ) , column ( '<STR_LIT:p>' ) , column ( '<STR_LIT:r>' ) , column ( '<STR_LIT:d>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 , c3 , c4 ) <EOL> with assertions . expect_warnings ( <EOL> r"<STR_LIT>" <EOL> "<STR_LIT>" ) : <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c2 ] , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ c3 ] <EOL> ) <EOL> def test_dupe_col_obj ( self ) : <EOL> c1 , c2 , c3 = column ( '<STR_LIT:q>' ) , column ( '<STR_LIT:p>' ) , column ( '<STR_LIT:r>' ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 , c3 , c2 ) <EOL> assert_raises_message ( <EOL> exc . InvalidRequestError , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> testing . db . execute , stmt <EOL> ) <EOL> def test_anon_aliased_unique ( self ) : <EOL> text1 = self . tables . text1 <EOL> c1 = text1 . c . a . label ( None ) <EOL> c2 = text1 . alias ( ) . c . c <EOL> c3 = text1 . alias ( ) . c . b <EOL> c4 = text1 . alias ( ) . c . d . label ( None ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 , c3 , c4 ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c1 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c2 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c3 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c4 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ text1 . c . a ] , "<STR_LIT>" ) <EOL> eq_ ( row [ text1 . c . d ] , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ text1 . c . b ] <EOL> ) <EOL> def test_anon_aliased_overlapping ( self ) : <EOL> text1 = self . tables . text1 <EOL> c1 = text1 . c . a . label ( None ) <EOL> c2 = text1 . alias ( ) . c . a <EOL> c3 = text1 . alias ( ) . c . a . label ( None ) <EOL> c4 = text1 . c . a . label ( None ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( c1 , c2 , c3 , c4 ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c1 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c2 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c3 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c4 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ text1 . c . a ] , "<STR_LIT>" ) <EOL> def test_anon_aliased_name_conflict ( self ) : <EOL> text1 = self . tables . text1 <EOL> c1 = text1 . c . a . label ( "<STR_LIT:a>" ) <EOL> c2 = text1 . alias ( ) . c . a <EOL> c3 = text1 . alias ( ) . c . a . label ( "<STR_LIT:a>" ) <EOL> c4 = text1 . c . a . label ( "<STR_LIT:a>" ) <EOL> stmt = text ( "<STR_LIT>" ) . columns ( <EOL> c1 , c2 , c3 , c4 ) <EOL> result = testing . db . execute ( stmt ) <EOL> row = result . first ( ) <EOL> eq_ ( row [ c1 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c2 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c3 ] , "<STR_LIT>" ) <EOL> eq_ ( row [ c4 ] , "<STR_LIT>" ) <EOL> assert_raises_message ( <EOL> exc . NoSuchColumnError , <EOL> "<STR_LIT>" , <EOL> lambda : row [ text1 . c . a ] <EOL> ) <EOL> class AlternateResultProxyTest ( fixtures . TablesTest ) : <EOL> __requires__ = ( '<STR_LIT>' , ) <EOL> @ classmethod <EOL> def setup_bind ( cls ) : <EOL> cls . engine = engine = engines . testing_engine ( '<STR_LIT>' ) <EOL> return engine <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT:test>' , metadata , <EOL> Column ( '<STR_LIT:x>' , Integer , primary_key = True ) , <EOL> Column ( '<STR_LIT:y>' , String ( <NUM_LIT:50> , convert_unicode = '<STR_LIT>' ) ) <EOL> ) <EOL> @ classmethod <EOL> def insert_data ( cls ) : <EOL> cls . engine . execute ( cls . tables . test . insert ( ) , [ <EOL> { '<STR_LIT:x>' : i , '<STR_LIT:y>' : "<STR_LIT>" % i } for i in range ( <NUM_LIT:1> , <NUM_LIT:12> ) <EOL> ] ) <EOL> @ contextmanager <EOL> def _proxy_fixture ( self , cls ) : <EOL> self . table = self . tables . test <EOL> class ExcCtx ( default . DefaultExecutionContext ) : <EOL> def get_result_proxy ( self ) : <EOL> return cls ( self ) <EOL> self . patcher = patch . object ( <EOL> self . engine . dialect , "<STR_LIT>" , ExcCtx ) <EOL> with self . patcher : <EOL> yield <EOL> def _test_proxy ( self , cls ) : <EOL> with self . _proxy_fixture ( cls ) : <EOL> rows = [ ] <EOL> r = self . engine . execute ( select ( [ self . table ] ) ) <EOL> assert isinstance ( r , cls ) <EOL> for i in range ( <NUM_LIT:5> ) : <EOL> rows . append ( r . fetchone ( ) ) <EOL> eq_ ( rows , [ ( i , "<STR_LIT>" % i ) for i in range ( <NUM_LIT:1> , <NUM_LIT:6> ) ] ) <EOL> rows = r . fetchmany ( <NUM_LIT:3> ) <EOL> eq_ ( rows , [ ( i , "<STR_LIT>" % i ) for i in range ( <NUM_LIT:6> , <NUM_LIT:9> ) ] ) <EOL> rows = r . fetchall ( ) <EOL> eq_ ( rows , [ ( i , "<STR_LIT>" % i ) for i in range ( <NUM_LIT:9> , <NUM_LIT:12> ) ] ) <EOL> r = self . engine . execute ( select ( [ self . table ] ) ) <EOL> rows = r . fetchmany ( None ) <EOL> eq_ ( rows [ <NUM_LIT:0> ] , ( <NUM_LIT:1> , "<STR_LIT>" ) ) <EOL> assert len ( rows ) == <NUM_LIT:1> or len ( rows ) == <NUM_LIT:11> <EOL> r = self . engine . execute ( select ( [ self . table ] ) . limit ( <NUM_LIT:1> ) ) <EOL> r . fetchone ( ) <EOL> eq_ ( r . fetchone ( ) , None ) <EOL> r = self . engine . execute ( select ( [ self . table ] ) . limit ( <NUM_LIT:5> ) ) <EOL> rows = r . fetchmany ( <NUM_LIT:6> ) <EOL> eq_ ( rows , [ ( i , "<STR_LIT>" % i ) for i in range ( <NUM_LIT:1> , <NUM_LIT:6> ) ] ) <EOL> eq_ ( r . fetchmany ( <NUM_LIT:2> ) , [ ] ) <EOL> eq_ ( r . fetchmany ( <NUM_LIT:2> ) , [ ] ) <EOL> eq_ ( r . fetchall ( ) , [ ] ) <EOL> eq_ ( r . fetchone ( ) , None ) <EOL> r . close ( ) <EOL> self . _assert_result_closed ( r ) <EOL> r = self . engine . execute ( select ( [ self . table ] ) . limit ( <NUM_LIT:5> ) ) <EOL> eq_ ( r . first ( ) , ( <NUM_LIT:1> , "<STR_LIT>" ) ) <EOL> self . _assert_result_closed ( r ) <EOL> r = self . engine . execute ( select ( [ self . table ] ) . limit ( <NUM_LIT:5> ) ) <EOL> eq_ ( r . scalar ( ) , <NUM_LIT:1> ) <EOL> self . _assert_result_closed ( r ) <EOL> def _assert_result_closed ( self , r ) : <EOL> assert_raises_message ( <EOL> sa_exc . ResourceClosedError , <EOL> "<STR_LIT>" , <EOL> r . fetchone <EOL> ) <EOL> assert_raises_message ( <EOL> sa_exc . ResourceClosedError , <EOL> "<STR_LIT>" , <EOL> r . fetchmany , <NUM_LIT:2> <EOL> ) <EOL> assert_raises_message ( <EOL> sa_exc . ResourceClosedError , <EOL> "<STR_LIT>" , <EOL> r . fetchall <EOL> ) <EOL> def test_basic_plain ( self ) : <EOL> self . _test_proxy ( _result . ResultProxy ) <EOL> def test_basic_buffered_row_result_proxy ( self ) : <EOL> self . _test_proxy ( _result . BufferedRowResultProxy ) <EOL> def test_basic_fully_buffered_result_proxy ( self ) : <EOL> self . _test_proxy ( _result . FullyBufferedResultProxy ) <EOL> def test_basic_buffered_column_result_proxy ( self ) : <EOL> self . _test_proxy ( _result . BufferedColumnResultProxy ) <EOL> def test_resultprocessor_plain ( self ) : <EOL> self . _test_result_processor ( _result . ResultProxy , False ) <EOL> def test_resultprocessor_plain_cached ( self ) : <EOL> self . _test_result_processor ( _result . ResultProxy , True ) <EOL> def test_resultprocessor_buffered_column ( self ) : <EOL> self . _test_result_processor ( _result . BufferedColumnResultProxy , False ) <EOL> def test_resultprocessor_buffered_column_cached ( self ) : <EOL> self . _test_result_processor ( _result . BufferedColumnResultProxy , True ) <EOL> def test_resultprocessor_buffered_row ( self ) : <EOL> self . _test_result_processor ( _result . BufferedRowResultProxy , False ) <EOL> def test_resultprocessor_buffered_row_cached ( self ) : <EOL> self . _test_result_processor ( _result . BufferedRowResultProxy , True ) <EOL> def test_resultprocessor_fully_buffered ( self ) : <EOL> self . _test_result_processor ( _result . FullyBufferedResultProxy , False ) <EOL> def test_resultprocessor_fully_buffered_cached ( self ) : <EOL> self . _test_result_processor ( _result . FullyBufferedResultProxy , True ) <EOL> def _test_result_processor ( self , cls , use_cache ) : <EOL> class MyType ( TypeDecorator ) : <EOL> impl = String ( ) <EOL> def process_result_value ( self , value , dialect ) : <EOL> return "<STR_LIT>" + value <EOL> with self . _proxy_fixture ( cls ) : <EOL> with self . engine . connect ( ) as conn : <EOL> if use_cache : <EOL> cache = { } <EOL> conn = conn . execution_options ( compiled_cache = cache ) <EOL> stmt = select ( [ literal ( "<STR_LIT>" , type_ = MyType ( ) ) ] ) <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> r = conn . execute ( stmt ) <EOL> eq_ ( r . scalar ( ) , "<STR_LIT>" ) <EOL> def test_buffered_row_growth ( self ) : <EOL> with self . _proxy_fixture ( _result . BufferedRowResultProxy ) : <EOL> with self . engine . connect ( ) as conn : <EOL> conn . execute ( self . table . insert ( ) , [ <EOL> { '<STR_LIT:x>' : i , '<STR_LIT:y>' : "<STR_LIT>" % i } for i in range ( <NUM_LIT:15> , <NUM_LIT> ) <EOL> ] ) <EOL> result = conn . execute ( self . table . select ( ) ) <EOL> checks = { <EOL> <NUM_LIT:0> : <NUM_LIT:5> , <NUM_LIT:1> : <NUM_LIT:10> , <NUM_LIT:9> : <NUM_LIT:20> , <NUM_LIT> : <NUM_LIT> , <NUM_LIT> : <NUM_LIT> , <EOL> <NUM_LIT> : <NUM_LIT:1000> <EOL> } <EOL> for idx , row in enumerate ( result , <NUM_LIT:0> ) : <EOL> if idx in checks : <EOL> eq_ ( result . _bufsize , checks [ idx ] ) <EOL> le_ ( <EOL> len ( result . _BufferedRowResultProxy__rowbuffer ) , <EOL> <NUM_LIT:1000> <EOL> ) <EOL> def test_max_row_buffer_option ( self ) : <EOL> with self . _proxy_fixture ( _result . BufferedRowResultProxy ) : <EOL> with self . engine . connect ( ) as conn : <EOL> conn . execute ( self . table . insert ( ) , [ <EOL> { '<STR_LIT:x>' : i , '<STR_LIT:y>' : "<STR_LIT>" % i } for i in range ( <NUM_LIT:15> , <NUM_LIT> ) <EOL> ] ) <EOL> result = conn . execution_options ( max_row_buffer = <NUM_LIT> ) . execute ( <EOL> self . table . select ( ) <EOL> ) <EOL> for idx , row in enumerate ( result , <NUM_LIT:0> ) : <EOL> if idx in ( <NUM_LIT:16> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ) : <EOL> eq_ ( result . _bufsize , <NUM_LIT> ) <EOL> le_ ( <EOL> len ( result . _BufferedRowResultProxy__rowbuffer ) , <EOL> <NUM_LIT> <EOL> ) </s>
