<s> import warnings <EOL> from pytest import mark <EOL> class DummyVenusian ( object ) : <EOL> def __init__ ( self ) : <EOL> self . attached = [ ] <EOL> def attach ( self , wrapped , fn , category = None ) : <EOL> self . attached . append ( ( wrapped , fn , category ) ) <EOL> class TestEvents : <EOL> @ mark . user ( '<STR_LIT>' ) <EOL> def test_owner ( self , root , db_session , events , dummy_request ) : <EOL> from kotti . resources import Content <EOL> from kotti . security import list_groups <EOL> from kotti . security import list_groups_raw <EOL> from kotti . util import clear_cache <EOL> child = root [ u'<STR_LIT>' ] = Content ( ) <EOL> db_session . flush ( ) <EOL> assert child . owner == u'<STR_LIT>' <EOL> assert list_groups ( u'<STR_LIT>' , child ) == [ u'<STR_LIT>' ] <EOL> clear_cache ( ) <EOL> grandchild = child [ u'<STR_LIT>' ] = Content ( ) <EOL> db_session . flush ( ) <EOL> assert grandchild . owner == u'<STR_LIT>' <EOL> assert list_groups ( u'<STR_LIT>' , grandchild ) == [ u'<STR_LIT>' ] <EOL> assert len ( list_groups_raw ( u'<STR_LIT>' , grandchild ) ) == <NUM_LIT:0> <EOL> def test_sqlalchemy_events ( self , root , db_session , events ) : <EOL> from kotti import events <EOL> from kotti . resources import Content <EOL> insert_events = [ ] <EOL> def insert ( event ) : <EOL> insert_events . append ( event ) <EOL> update_events = [ ] <EOL> def update ( event ) : <EOL> update_events . append ( event ) <EOL> delete_events = [ ] <EOL> def delete ( event ) : <EOL> delete_events . append ( event ) <EOL> after_delete_events = [ ] <EOL> def after_delete ( event ) : <EOL> after_delete_events . append ( event ) <EOL> def lengths ( ) : <EOL> return ( len ( insert_events ) , len ( update_events ) , <EOL> len ( delete_events ) , len ( after_delete_events ) ) <EOL> lis = events . objectevent_listeners <EOL> lis [ ( events . ObjectInsert , None ) ] . append ( insert ) <EOL> lis [ ( events . ObjectUpdate , None ) ] . append ( update ) <EOL> lis [ ( events . ObjectDelete , None ) ] . append ( delete ) <EOL> with warnings . catch_warnings ( record = True ) : <EOL> lis [ ( events . ObjectAfterDelete , None ) ] . append ( after_delete ) <EOL> child = root [ u'<STR_LIT>' ] = Content ( ) <EOL> db_session . flush ( ) <EOL> assert lengths ( ) == ( <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> assert insert_events [ <NUM_LIT:0> ] . object == child <EOL> child . title = u"<STR_LIT>" <EOL> db_session . flush ( ) <EOL> assert lengths ( ) == ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> assert update_events [ <NUM_LIT:0> ] . object == child <EOL> db_session . delete ( child ) <EOL> db_session . flush ( ) <EOL> assert lengths ( ) == ( <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> assert delete_events [ <NUM_LIT:0> ] . object == child <EOL> assert after_delete_events [ <NUM_LIT:0> ] . object == child <EOL> def test_subscribe ( self , root , db_session ) : <EOL> from kotti . events import ObjectEvent <EOL> from kotti . events import clear <EOL> from kotti . events import listeners <EOL> from kotti . events import objectevent_listeners <EOL> from kotti . events import subscribe <EOL> from kotti . resources import Document <EOL> def handler ( event ) : <EOL> pass <EOL> dec = subscribe ( ) <EOL> dec . venusian = DummyVenusian ( ) <EOL> decorated = dec ( handler ) <EOL> dec . register ( None , None , handler ) <EOL> assert dec . evttype is object <EOL> assert dec . objtype is None <EOL> assert decorated == handler <EOL> assert ( handler , dec . register , '<STR_LIT>' ) in dec . venusian . attached <EOL> assert handler in listeners [ object ] <EOL> assert handler not in objectevent_listeners [ object ] <EOL> clear ( ) <EOL> dec = subscribe ( ObjectEvent ) <EOL> dec . venusian = DummyVenusian ( ) <EOL> decorated = dec ( handler ) <EOL> dec . register ( None , None , handler ) <EOL> assert dec . evttype is ObjectEvent <EOL> assert dec . objtype is None <EOL> assert decorated == handler <EOL> assert ( handler , dec . register , '<STR_LIT>' ) in dec . venusian . attached <EOL> assert handler not in listeners [ ObjectEvent ] <EOL> assert handler in objectevent_listeners [ ( ObjectEvent , None ) ] <EOL> clear ( ) <EOL> dec = subscribe ( ObjectEvent , Document ) <EOL> dec . venusian = DummyVenusian ( ) <EOL> decorated = dec ( handler ) <EOL> dec . register ( None , None , handler ) <EOL> assert dec . evttype is ObjectEvent <EOL> assert dec . objtype is Document <EOL> assert decorated == handler <EOL> assert ( handler , dec . register , '<STR_LIT>' ) in dec . venusian . attached <EOL> assert handler not in listeners [ ObjectEvent ] <EOL> assert handler in objectevent_listeners [ ( ObjectEvent , Document ) ] <EOL> def test_set_modification_date ( self , root , db_session , events ) : <EOL> from time import sleep <EOL> from kotti . resources import Document <EOL> d1 = root [ '<STR_LIT>' ] = Document ( title = u'<STR_LIT>' ) <EOL> d2 = root [ '<STR_LIT>' ] = Document ( title = u'<STR_LIT>' ) <EOL> assert d1 . position == <NUM_LIT:0> <EOL> assert d2 . position == <NUM_LIT:1> <EOL> db_session . flush ( ) <EOL> md1 = d1 . modification_date <EOL> md2 = d2 . modification_date <EOL> sleep ( <NUM_LIT:1> ) <EOL> d1 . position = <NUM_LIT:1> <EOL> d2 . position = <NUM_LIT:0> <EOL> db_session . flush ( ) <EOL> assert d1 . position == <NUM_LIT:1> <EOL> assert d2 . position == <NUM_LIT:0> <EOL> assert d1 . modification_date == md1 <EOL> assert d2 . modification_date == md2 <EOL> d1 . title = u'<STR_LIT>' <EOL> d2 . title = u'<STR_LIT>' <EOL> db_session . flush ( ) <EOL> assert d1 . modification_date != md1 <EOL> assert d2 . modification_date != md2 </s>
