<s> import unittest <EOL> from pyramid import testing <EOL> class Test_root_factory ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . config = testing . setUp ( ) <EOL> def tearDown ( self ) : <EOL> testing . tearDown ( ) <EOL> def _callFUT ( self , request , transaction , get_connection , evolve_packages ) : <EOL> from . . import root_factory <EOL> return root_factory ( request , transaction , get_connection , <EOL> evolve_packages ) <EOL> def _makeRequest ( self , app_root = None ) : <EOL> request = Dummy ( ) <EOL> request . registry = DummyRegistry ( ) <EOL> request . registry . content = Dummy ( ) <EOL> request . registry . content . create = lambda * arg : app_root <EOL> return request <EOL> def test_without_app_root ( self ) : <EOL> txn = DummyTransaction ( ) <EOL> root = { } <EOL> gc = Dummy_get_connection ( root ) <EOL> ep = DummyFunction ( True ) <EOL> app_root = object ( ) <EOL> request = self . _makeRequest ( app_root ) <EOL> result = self . _callFUT ( request , txn , gc , ep ) <EOL> self . assertEqual ( result , app_root ) <EOL> self . assertTrue ( txn . committed ) <EOL> self . assertTrue ( txn . savepointed ) <EOL> self . assertTrue ( ep . called ) <EOL> def test_with_app_root ( self ) : <EOL> txn = DummyTransaction ( ) <EOL> app_root = object ( ) <EOL> root = { '<STR_LIT>' : app_root } <EOL> gc = Dummy_get_connection ( root ) <EOL> ep = DummyFunction ( True ) <EOL> request = testing . DummyRequest ( ) <EOL> result = self . _callFUT ( request , txn , gc , ep ) <EOL> self . assertEqual ( result , app_root ) <EOL> self . assertFalse ( txn . committed ) <EOL> class Test_includeme ( unittest . TestCase ) : <EOL> def test_it ( self ) : <EOL> from . . import ( <EOL> includeme , <EOL> connection_opened , <EOL> connection_will_close , <EOL> ZODBConnectionOpened , <EOL> ZODBConnectionWillClose , <EOL> ) <EOL> config = DummyConfig ( ) <EOL> includeme ( config ) <EOL> self . assertEqual ( <EOL> config . subscriptions , <EOL> [ ( connection_opened , ZODBConnectionOpened ) , <EOL> ( connection_will_close , ZODBConnectionWillClose ) , <EOL> ] <EOL> ) <EOL> class Test_connection_opened ( unittest . TestCase ) : <EOL> def test_it ( self ) : <EOL> from . . import connection_opened <EOL> event = DummyEvent ( ) <EOL> connection_opened ( event ) <EOL> self . assertEqual ( event . request . _zodb_tx_counts , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> class Test_connection_will_close ( unittest . TestCase ) : <EOL> def _callFUT ( self , event , statsd_incr ) : <EOL> from . . import connection_will_close <EOL> return connection_will_close ( event , statsd_incr ) <EOL> def test_no_tx_counts ( self ) : <EOL> event = DummyEvent ( ) <EOL> result = self . _callFUT ( event , None ) <EOL> self . assertEqual ( result , None ) <EOL> def test_with_postitive_tx_counts ( self ) : <EOL> event = DummyEvent ( <NUM_LIT:5> , <NUM_LIT:5> ) <EOL> event . request . _zodb_tx_counts = ( <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> L = [ ] <EOL> def statsd_incr ( name , num , registry = None ) : <EOL> L . append ( ( name , num ) ) <EOL> self . _callFUT ( event , statsd_incr ) <EOL> self . assertEqual ( <EOL> L , <EOL> [ ( '<STR_LIT>' , <NUM_LIT:4> ) , ( '<STR_LIT>' , <NUM_LIT:4> ) ] <EOL> ) <EOL> def test_with_zero_tx_counts ( self ) : <EOL> event = DummyEvent ( <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> event . request . _zodb_tx_counts = ( <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> L = [ ] <EOL> self . _callFUT ( event , None ) <EOL> self . assertEqual ( <EOL> L , <EOL> [ ] <EOL> ) <EOL> class DummyTransaction ( object ) : <EOL> committed = False <EOL> savepointed = False <EOL> def commit ( self ) : <EOL> self . committed = True <EOL> def savepoint ( self ) : <EOL> self . savepointed = True <EOL> class Dummy_get_connection ( object ) : <EOL> def __init__ ( self , root ) : <EOL> self . _root = root <EOL> def root ( self ) : <EOL> return self . _root <EOL> def __call__ ( self , request ) : <EOL> return self <EOL> class DummyFunction ( object ) : <EOL> called = False <EOL> def __init__ ( self , result ) : <EOL> self . result = result <EOL> def __call__ ( self , * args , ** kw ) : <EOL> self . called = True <EOL> self . args = args <EOL> self . kw = kw <EOL> return self . result <EOL> class Dummy ( object ) : <EOL> pass <EOL> class DummyRegistry ( object ) : <EOL> def notify ( self , event ) : <EOL> self . event = event <EOL> class DummyConfig ( object ) : <EOL> def __init__ ( self ) : <EOL> self . subscriptions = [ ] <EOL> def add_subscriber ( self , fn , event_type ) : <EOL> self . subscriptions . append ( ( fn , event_type ) ) <EOL> class DummyConnection ( object ) : <EOL> def __init__ ( self , loads , stores ) : <EOL> self . loads = loads <EOL> self . stores = stores <EOL> def getTransferCounts ( self ) : <EOL> return ( self . loads , self . stores ) <EOL> class DummyEvent ( object ) : <EOL> def __init__ ( self , loads = <NUM_LIT:0> , stores = <NUM_LIT:0> ) : <EOL> self . request = testing . DummyRequest ( ) <EOL> self . conn = DummyConnection ( loads , stores ) </s>
