<s> from sqlalchemy import Integer , ForeignKey , String <EOL> from sqlalchemy . types import PickleType , TypeDecorator , VARCHAR <EOL> from sqlalchemy . orm import mapper , Session , composite <EOL> from sqlalchemy . orm . mapper import Mapper <EOL> from sqlalchemy . orm . instrumentation import ClassManager <EOL> from sqlalchemy . testing . schema import Table , Column <EOL> from sqlalchemy . testing import eq_ , assert_raises_message , assert_raises <EOL> from sqlalchemy . testing . util import picklers <EOL> from sqlalchemy . testing import fixtures <EOL> from sqlalchemy . ext . mutable import MutableComposite <EOL> from sqlalchemy . ext . mutable import MutableDict , MutableList , MutableSet <EOL> class Foo ( fixtures . BasicEntity ) : <EOL> pass <EOL> class SubFoo ( Foo ) : <EOL> pass <EOL> class FooWithEq ( object ) : <EOL> def __init__ ( self , ** kw ) : <EOL> for k in kw : <EOL> setattr ( self , k , kw [ k ] ) <EOL> def __hash__ ( self ) : <EOL> return hash ( self . id ) <EOL> def __eq__ ( self , other ) : <EOL> return self . id == other . id <EOL> class Point ( MutableComposite ) : <EOL> def __init__ ( self , x , y ) : <EOL> self . x = x <EOL> self . y = y <EOL> def __setattr__ ( self , key , value ) : <EOL> object . __setattr__ ( self , key , value ) <EOL> self . changed ( ) <EOL> def __composite_values__ ( self ) : <EOL> return self . x , self . y <EOL> def __getstate__ ( self ) : <EOL> return self . x , self . y <EOL> def __setstate__ ( self , state ) : <EOL> self . x , self . y = state <EOL> def __eq__ ( self , other ) : <EOL> return isinstance ( other , Point ) and other . x == self . x and other . y == self . y <EOL> class MyPoint ( Point ) : <EOL> @ classmethod <EOL> def coerce ( cls , key , value ) : <EOL> if isinstance ( value , tuple ) : <EOL> value = Point ( * value ) <EOL> return value <EOL> class _MutableDictTestFixture ( object ) : <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> return MutableDict <EOL> def teardown ( self ) : <EOL> Mapper . dispatch . _clear ( ) <EOL> ClassManager . dispatch . _clear ( ) <EOL> super ( _MutableDictTestFixture , self ) . teardown ( ) <EOL> class _MutableDictTestBase ( _MutableDictTestFixture ) : <EOL> run_define_tables = '<STR_LIT>' <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> mapper ( Foo , foo ) <EOL> def test_coerce_none ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = None ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , None ) <EOL> def test_coerce_raise ( self ) : <EOL> assert_raises_message ( <EOL> ValueError , <EOL> "<STR_LIT>" , <EOL> Foo , data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> ) <EOL> def test_in_place_mutation ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data [ '<STR_LIT:a>' ] = '<STR_LIT:c>' <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:c>' } ) <EOL> def test_clear ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . clear ( ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { } ) <EOL> def test_update ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . update ( { '<STR_LIT:a>' : '<STR_LIT:z>' } ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:z>' } ) <EOL> def test_pop ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . pop ( '<STR_LIT:a>' ) , '<STR_LIT:b>' ) <EOL> sess . commit ( ) <EOL> assert_raises ( KeyError , f1 . data . pop , '<STR_LIT:g>' ) <EOL> eq_ ( f1 . data , { '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> def test_pop_default ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . pop ( '<STR_LIT:a>' , '<STR_LIT:q>' ) , '<STR_LIT:b>' ) <EOL> eq_ ( f1 . data . pop ( '<STR_LIT:a>' , '<STR_LIT:q>' ) , '<STR_LIT:q>' ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> def test_popitem ( self ) : <EOL> sess = Session ( ) <EOL> orig = { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } <EOL> data = dict ( orig ) <EOL> f1 = Foo ( data = data ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> k , v = f1 . data . popitem ( ) <EOL> assert k in ( '<STR_LIT:a>' , '<STR_LIT:c>' ) <EOL> orig . pop ( k ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , orig ) <EOL> def test_setdefault ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . setdefault ( '<STR_LIT:c>' , '<STR_LIT:d>' ) , '<STR_LIT:d>' ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> eq_ ( f1 . data . setdefault ( '<STR_LIT:c>' , '<STR_LIT:q>' ) , '<STR_LIT:d>' ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> def test_replace ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . data = { '<STR_LIT:b>' : '<STR_LIT:c>' } <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:b>' : '<STR_LIT:c>' } ) <EOL> def test_replace_itself_still_ok ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . data = f1 . data <EOL> f1 . data [ '<STR_LIT:b>' ] = '<STR_LIT:c>' <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:b>' : '<STR_LIT:c>' } ) <EOL> def test_pickle_parent ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data <EOL> sess . close ( ) <EOL> for loads , dumps in picklers ( ) : <EOL> sess = Session ( ) <EOL> f2 = loads ( dumps ( f1 ) ) <EOL> sess . add ( f2 ) <EOL> f2 . data [ '<STR_LIT:a>' ] = '<STR_LIT:c>' <EOL> assert f2 in sess . dirty <EOL> def test_unrelated_flush ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { "<STR_LIT:a>" : "<STR_LIT:b>" } , unrelated_data = "<STR_LIT>" ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . unrelated_data = "<STR_LIT>" <EOL> sess . flush ( ) <EOL> f1 . data [ "<STR_LIT:a>" ] = "<STR_LIT:c>" <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data [ "<STR_LIT:a>" ] , "<STR_LIT:c>" ) <EOL> def _test_non_mutable ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( non_mutable_data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . non_mutable_data [ '<STR_LIT:a>' ] = '<STR_LIT:c>' <EOL> sess . commit ( ) <EOL> eq_ ( f1 . non_mutable_data , { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> class _MutableListTestFixture ( object ) : <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> return MutableList <EOL> def teardown ( self ) : <EOL> Mapper . dispatch . _clear ( ) <EOL> ClassManager . dispatch . _clear ( ) <EOL> super ( _MutableListTestFixture , self ) . teardown ( ) <EOL> class _MutableListTestBase ( _MutableListTestFixture ) : <EOL> run_define_tables = '<STR_LIT>' <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> mapper ( Foo , foo ) <EOL> def test_coerce_none ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = None ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , None ) <EOL> def test_coerce_raise ( self ) : <EOL> assert_raises_message ( <EOL> ValueError , <EOL> "<STR_LIT>" , <EOL> Foo , data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> ) <EOL> def test_in_place_mutation ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data [ <NUM_LIT:0> ] = <NUM_LIT:3> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:3> , <NUM_LIT:2> ] ) <EOL> def test_in_place_slice_mutation ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data [ <NUM_LIT:1> : <NUM_LIT:3> ] = <NUM_LIT:5> , <NUM_LIT:6> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:6> , <NUM_LIT:4> ] ) <EOL> def test_del_slice ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> del f1 . data [ <NUM_LIT:1> : <NUM_LIT:3> ] <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:4> ] ) <EOL> def test_clear ( self ) : <EOL> if not hasattr ( list , '<STR_LIT>' ) : <EOL> return <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . clear ( ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ ] ) <EOL> def test_pop ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . pop ( ) , <NUM_LIT:3> ) <EOL> eq_ ( f1 . data . pop ( <NUM_LIT:0> ) , <NUM_LIT:1> ) <EOL> sess . commit ( ) <EOL> assert_raises ( IndexError , f1 . data . pop , <NUM_LIT:5> ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:2> ] ) <EOL> def test_append ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . append ( <NUM_LIT:5> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:5> ] ) <EOL> def test_extend ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . extend ( [ <NUM_LIT:5> ] ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:5> ] ) <EOL> def test_insert ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . insert ( <NUM_LIT:1> , <NUM_LIT:5> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:5> , <NUM_LIT:2> ] ) <EOL> def test_remove ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . remove ( <NUM_LIT:2> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:3> ] ) <EOL> def test_sort ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . sort ( ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> def test_reverse ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . reverse ( ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , [ <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:1> ] ) <EOL> def test_pickle_parent ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data <EOL> sess . close ( ) <EOL> for loads , dumps in picklers ( ) : <EOL> sess = Session ( ) <EOL> f2 = loads ( dumps ( f1 ) ) <EOL> sess . add ( f2 ) <EOL> f2 . data [ <NUM_LIT:0> ] = <NUM_LIT:3> <EOL> assert f2 in sess . dirty <EOL> def test_unrelated_flush ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = [ <NUM_LIT:1> , <NUM_LIT:2> ] , unrelated_data = "<STR_LIT>" ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . unrelated_data = "<STR_LIT>" <EOL> sess . flush ( ) <EOL> f1 . data [ <NUM_LIT:0> ] = <NUM_LIT:3> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data [ <NUM_LIT:0> ] , <NUM_LIT:3> ) <EOL> class _MutableSetTestFixture ( object ) : <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> return MutableSet <EOL> def teardown ( self ) : <EOL> Mapper . dispatch . _clear ( ) <EOL> ClassManager . dispatch . _clear ( ) <EOL> super ( _MutableSetTestFixture , self ) . teardown ( ) <EOL> class _MutableSetTestBase ( _MutableSetTestFixture ) : <EOL> run_define_tables = '<STR_LIT>' <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> mapper ( Foo , foo ) <EOL> def test_coerce_none ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = None ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , None ) <EOL> def test_coerce_raise ( self ) : <EOL> assert_raises_message ( <EOL> ValueError , <EOL> "<STR_LIT>" , <EOL> Foo , data = [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] <EOL> ) <EOL> def test_clear ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . clear ( ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( ) ) <EOL> def test_pop ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . pop ( ) , <NUM_LIT:1> ) <EOL> sess . commit ( ) <EOL> assert_raises ( KeyError , f1 . data . pop ) <EOL> eq_ ( f1 . data , set ( ) ) <EOL> def test_add ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . add ( <NUM_LIT:5> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> def test_update ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . update ( set ( [ <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> def test_intersection_update ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . intersection_update ( set ( [ <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:2> ] ) ) <EOL> def test_difference_update ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . difference_update ( set ( [ <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> ] ) ) <EOL> def test_symmetric_difference_update ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . symmetric_difference_update ( set ( [ <NUM_LIT:2> , <NUM_LIT:5> ] ) ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:5> ] ) ) <EOL> def test_remove ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . remove ( <NUM_LIT:2> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:3> ] ) ) <EOL> def test_discard ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . discard ( <NUM_LIT:2> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:3> ] ) ) <EOL> f1 . data . discard ( <NUM_LIT:2> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:3> ] ) ) <EOL> def test_pickle_parent ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data <EOL> sess . close ( ) <EOL> for loads , dumps in picklers ( ) : <EOL> sess = Session ( ) <EOL> f2 = loads ( dumps ( f1 ) ) <EOL> sess . add ( f2 ) <EOL> f2 . data . add ( <NUM_LIT:3> ) <EOL> assert f2 in sess . dirty <EOL> def test_unrelated_flush ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = set ( [ <NUM_LIT:1> , <NUM_LIT:2> ] ) , unrelated_data = "<STR_LIT>" ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . unrelated_data = "<STR_LIT>" <EOL> sess . flush ( ) <EOL> f1 . data . add ( <NUM_LIT:3> ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , set ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) ) <EOL> class MutableColumnDefaultTest ( _MutableDictTestFixture , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> MutableDict = cls . _type_fixture ( ) <EOL> mutable_pickle = MutableDict . as_mutable ( PickleType ) <EOL> Table ( <EOL> '<STR_LIT:foo>' , metadata , <EOL> Column ( <EOL> '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , mutable_pickle , default = { } ) , <EOL> ) <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> mapper ( Foo , foo ) <EOL> def test_evt_on_flush_refresh ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> assert isinstance ( f1 . data , self . _type_fixture ( ) ) <EOL> assert f1 not in sess . dirty <EOL> f1 . data [ '<STR_LIT:foo>' ] = '<STR_LIT:bar>' <EOL> assert f1 in sess . dirty <EOL> class MutableWithScalarPickleTest ( _MutableDictTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> MutableDict = cls . _type_fixture ( ) <EOL> mutable_pickle = MutableDict . as_mutable ( PickleType ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT:data>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> def test_non_mutable ( self ) : <EOL> self . _test_non_mutable ( ) <EOL> class MutableWithScalarJSONTest ( _MutableDictTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> import json <EOL> class JSONEncodedDict ( TypeDecorator ) : <EOL> impl = VARCHAR ( <NUM_LIT:50> ) <EOL> def process_bind_param ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . dumps ( value ) <EOL> return value <EOL> def process_result_value ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . loads ( value ) <EOL> return value <EOL> MutableDict = cls . _type_fixture ( ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , MutableDict . as_mutable ( JSONEncodedDict ) ) , <EOL> Column ( '<STR_LIT>' , JSONEncodedDict ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> def test_non_mutable ( self ) : <EOL> self . _test_non_mutable ( ) <EOL> class MutableListWithScalarPickleTest ( _MutableListTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> MutableList = cls . _type_fixture ( ) <EOL> mutable_pickle = MutableList . as_mutable ( PickleType ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT:data>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> class MutableSetWithScalarPickleTest ( _MutableSetTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> MutableSet = cls . _type_fixture ( ) <EOL> mutable_pickle = MutableSet . as_mutable ( PickleType ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT:data>' , mutable_pickle ) , <EOL> Column ( '<STR_LIT>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> class MutableAssocWithAttrInheritTest ( _MutableDictTestBase , <EOL> fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> ) <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> subfoo = cls . tables . subfoo <EOL> mapper ( Foo , foo ) <EOL> mapper ( SubFoo , subfoo , inherits = Foo ) <EOL> MutableDict . associate_with_attribute ( Foo . data ) <EOL> def test_in_place_mutation ( self ) : <EOL> sess = Session ( ) <EOL> f1 = SubFoo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data [ '<STR_LIT:a>' ] = '<STR_LIT:c>' <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:a>' : '<STR_LIT:c>' } ) <EOL> def test_replace ( self ) : <EOL> sess = Session ( ) <EOL> f1 = SubFoo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . data = { '<STR_LIT:b>' : '<STR_LIT:c>' } <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , { '<STR_LIT:b>' : '<STR_LIT:c>' } ) <EOL> class MutableAssociationScalarPickleTest ( _MutableDictTestBase , <EOL> fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> MutableDict = cls . _type_fixture ( ) <EOL> MutableDict . associate_with ( PickleType ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT>' , PickleType ) , <EOL> Column ( '<STR_LIT:data>' , PickleType ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> class MutableAssociationScalarJSONTest ( _MutableDictTestBase , <EOL> fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> import json <EOL> class JSONEncodedDict ( TypeDecorator ) : <EOL> impl = VARCHAR ( <NUM_LIT:50> ) <EOL> def process_bind_param ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . dumps ( value ) <EOL> return value <EOL> def process_result_value ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . loads ( value ) <EOL> return value <EOL> MutableDict = cls . _type_fixture ( ) <EOL> MutableDict . associate_with ( JSONEncodedDict ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , JSONEncodedDict ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> class CustomMutableAssociationScalarJSONTest ( _MutableDictTestBase , <EOL> fixtures . MappedTest ) : <EOL> CustomMutableDict = None <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> if not ( getattr ( cls , '<STR_LIT>' ) ) : <EOL> MutableDict = super ( <EOL> CustomMutableAssociationScalarJSONTest , cls ) . _type_fixture ( ) <EOL> class CustomMutableDict ( MutableDict ) : <EOL> pass <EOL> cls . CustomMutableDict = CustomMutableDict <EOL> return cls . CustomMutableDict <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> import json <EOL> class JSONEncodedDict ( TypeDecorator ) : <EOL> impl = VARCHAR ( <NUM_LIT:50> ) <EOL> def process_bind_param ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . dumps ( value ) <EOL> return value <EOL> def process_result_value ( self , value , dialect ) : <EOL> if value is not None : <EOL> value = json . loads ( value ) <EOL> return value <EOL> CustomMutableDict = cls . _type_fixture ( ) <EOL> CustomMutableDict . associate_with ( JSONEncodedDict ) <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:data>' , JSONEncodedDict ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> def test_pickle_parent ( self ) : <EOL> pass <EOL> def test_coerce ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> eq_ ( type ( f1 . data ) , self . _type_fixture ( ) ) <EOL> class _CompositeTestBase ( object ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:x>' , Integer ) , <EOL> Column ( '<STR_LIT:y>' , Integer ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> def setup ( self ) : <EOL> from sqlalchemy . ext import mutable <EOL> mutable . _setup_composite_listener ( ) <EOL> super ( _CompositeTestBase , self ) . setup ( ) <EOL> def teardown ( self ) : <EOL> Mapper . dispatch . _clear ( ) <EOL> ClassManager . dispatch . _clear ( ) <EOL> super ( _CompositeTestBase , self ) . teardown ( ) <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> return Point <EOL> class MutableCompositeColumnDefaultTest ( _CompositeTestBase , <EOL> fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( <EOL> '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:x>' , Integer , default = <NUM_LIT:5> ) , <EOL> Column ( '<STR_LIT:y>' , Integer , default = <NUM_LIT:9> ) , <EOL> Column ( '<STR_LIT>' , String ( <NUM_LIT:50> ) ) <EOL> ) <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> cls . Point = cls . _type_fixture ( ) <EOL> mapper ( Foo , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( cls . Point , foo . c . x , foo . c . y ) <EOL> } ) <EOL> def test_evt_on_flush_refresh ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = self . Point ( None , None ) ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> eq_ ( f1 . data , self . Point ( <NUM_LIT:5> , <NUM_LIT:9> ) ) <EOL> assert f1 not in sess . dirty <EOL> f1 . data . x = <NUM_LIT:10> <EOL> assert f1 in sess . dirty <EOL> class MutableCompositesUnpickleTest ( _CompositeTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> cls . Point = cls . _type_fixture ( ) <EOL> mapper ( FooWithEq , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( cls . Point , foo . c . x , foo . c . y ) <EOL> } ) <EOL> def test_unpickle_modified_eq ( self ) : <EOL> u1 = FooWithEq ( data = self . Point ( <NUM_LIT:3> , <NUM_LIT:5> ) ) <EOL> for loads , dumps in picklers ( ) : <EOL> loads ( dumps ( u1 ) ) <EOL> class MutableCompositesTest ( _CompositeTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> Point = cls . _type_fixture ( ) <EOL> mapper ( Foo , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( Point , foo . c . x , foo . c . y ) <EOL> } ) <EOL> def test_in_place_mutation ( self ) : <EOL> sess = Session ( ) <EOL> d = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> f1 = Foo ( data = d ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . y = <NUM_LIT:5> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , Point ( <NUM_LIT:3> , <NUM_LIT:5> ) ) <EOL> def test_pickle_of_parent ( self ) : <EOL> sess = Session ( ) <EOL> d = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> f1 = Foo ( data = d ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data <EOL> assert '<STR_LIT:data>' in f1 . __dict__ <EOL> sess . close ( ) <EOL> for loads , dumps in picklers ( ) : <EOL> sess = Session ( ) <EOL> f2 = loads ( dumps ( f1 ) ) <EOL> sess . add ( f2 ) <EOL> f2 . data . y = <NUM_LIT:12> <EOL> assert f2 in sess . dirty <EOL> def test_set_none ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = None ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , Point ( None , None ) ) <EOL> f1 . data . y = <NUM_LIT:5> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , Point ( None , <NUM_LIT:5> ) ) <EOL> def test_set_illegal ( self ) : <EOL> f1 = Foo ( ) <EOL> assert_raises_message ( <EOL> ValueError , <EOL> "<STR_LIT>" , <EOL> setattr , f1 , '<STR_LIT:data>' , '<STR_LIT:foo>' <EOL> ) <EOL> def test_unrelated_flush ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) , unrelated_data = "<STR_LIT>" ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . unrelated_data = "<STR_LIT>" <EOL> sess . flush ( ) <EOL> f1 . data . x = <NUM_LIT:5> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . x , <NUM_LIT:5> ) <EOL> class MutableCompositeCallableTest ( _CompositeTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> Point = cls . _type_fixture ( ) <EOL> mapper ( Foo , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( lambda x , y : Point ( x , y ) , foo . c . x , foo . c . y ) <EOL> } ) <EOL> def test_basic ( self ) : <EOL> sess = Session ( ) <EOL> f1 = Foo ( data = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> sess . add ( f1 ) <EOL> sess . flush ( ) <EOL> f1 . data . x = <NUM_LIT:5> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data . x , <NUM_LIT:3> ) <EOL> class MutableCompositeCustomCoerceTest ( _CompositeTestBase , <EOL> fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def _type_fixture ( cls ) : <EOL> return MyPoint <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> Point = cls . _type_fixture ( ) <EOL> mapper ( Foo , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( Point , foo . c . x , foo . c . y ) <EOL> } ) <EOL> def test_custom_coerce ( self ) : <EOL> f = Foo ( ) <EOL> f . data = ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> eq_ ( f . data , Point ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> def test_round_trip_ok ( self ) : <EOL> sess = Session ( ) <EOL> f = Foo ( ) <EOL> f . data = ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> sess . add ( f ) <EOL> sess . commit ( ) <EOL> eq_ ( f . data , Point ( <NUM_LIT:3> , <NUM_LIT:4> ) ) <EOL> class MutableInheritedCompositesTest ( _CompositeTestBase , fixtures . MappedTest ) : <EOL> @ classmethod <EOL> def define_tables ( cls , metadata ) : <EOL> Table ( '<STR_LIT:foo>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , primary_key = True , <EOL> test_needs_autoincrement = True ) , <EOL> Column ( '<STR_LIT:x>' , Integer ) , <EOL> Column ( '<STR_LIT:y>' , Integer ) <EOL> ) <EOL> Table ( '<STR_LIT>' , metadata , <EOL> Column ( '<STR_LIT:id>' , Integer , ForeignKey ( '<STR_LIT>' ) , primary_key = True ) , <EOL> ) <EOL> @ classmethod <EOL> def setup_mappers ( cls ) : <EOL> foo = cls . tables . foo <EOL> subfoo = cls . tables . subfoo <EOL> Point = cls . _type_fixture ( ) <EOL> mapper ( Foo , foo , properties = { <EOL> '<STR_LIT:data>' : composite ( Point , foo . c . x , foo . c . y ) <EOL> } ) <EOL> mapper ( SubFoo , subfoo , inherits = Foo ) <EOL> def test_in_place_mutation_subclass ( self ) : <EOL> sess = Session ( ) <EOL> d = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> f1 = SubFoo ( data = d ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data . y = <NUM_LIT:5> <EOL> sess . commit ( ) <EOL> eq_ ( f1 . data , Point ( <NUM_LIT:3> , <NUM_LIT:5> ) ) <EOL> def test_pickle_of_parent_subclass ( self ) : <EOL> sess = Session ( ) <EOL> d = Point ( <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> f1 = SubFoo ( data = d ) <EOL> sess . add ( f1 ) <EOL> sess . commit ( ) <EOL> f1 . data <EOL> assert '<STR_LIT:data>' in f1 . __dict__ <EOL> sess . close ( ) <EOL> for loads , dumps in picklers ( ) : <EOL> sess = Session ( ) <EOL> f2 = loads ( dumps ( f1 ) ) <EOL> sess . add ( f2 ) <EOL> f2 . data . y = <NUM_LIT:12> <EOL> assert f2 in sess . dirty </s>
