<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import sys , warnings <EOL> from unittest import TestResult <EOL> from twisted . python . compat import NativeStringIO as StringIO <EOL> from twisted . python . filepath import FilePath <EOL> from twisted . trial . unittest import ( <EOL> SynchronousTestCase , _collectWarnings , _setWarningRegistryToNone ) <EOL> class Mask ( object ) : <EOL> """<STR_LIT>""" <EOL> class MockTests ( SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> message = "<STR_LIT>" <EOL> category = UserWarning <EOL> def test_unflushed ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( self . message , self . category ) <EOL> def test_flushed ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( self . message , self . category ) <EOL> self . assertEqual ( len ( self . flushWarnings ( ) ) , <NUM_LIT:1> ) <EOL> class FlushWarningsTests ( SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def assertDictSubset ( self , set , subset ) : <EOL> """<STR_LIT>""" <EOL> for k , v in subset . items ( ) : <EOL> self . assertEqual ( set [ k ] , v ) <EOL> def assertDictSubsets ( self , sets , subsets ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( len ( sets ) , len ( subsets ) ) <EOL> for a , b in zip ( sets , subsets ) : <EOL> self . assertDictSubset ( a , b ) <EOL> def test_none ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( self . flushWarnings ( ) , [ ] ) <EOL> def test_several ( self ) : <EOL> """<STR_LIT>""" <EOL> firstMessage = "<STR_LIT>" <EOL> firstCategory = UserWarning <EOL> warnings . warn ( message = firstMessage , category = firstCategory ) <EOL> secondMessage = "<STR_LIT>" <EOL> secondCategory = RuntimeWarning <EOL> warnings . warn ( message = secondMessage , category = secondCategory ) <EOL> self . assertDictSubsets ( <EOL> self . flushWarnings ( ) , <EOL> [ { '<STR_LIT>' : firstCategory , '<STR_LIT:message>' : firstMessage } , <EOL> { '<STR_LIT>' : secondCategory , '<STR_LIT:message>' : secondMessage } ] ) <EOL> def test_repeated ( self ) : <EOL> """<STR_LIT>""" <EOL> message = "<STR_LIT>" <EOL> category = RuntimeWarning <EOL> for i in range ( <NUM_LIT:2> ) : <EOL> warnings . warn ( message = message , category = category ) <EOL> self . assertDictSubsets ( <EOL> self . flushWarnings ( ) , <EOL> [ { '<STR_LIT>' : category , '<STR_LIT:message>' : message } ] * <NUM_LIT:2> ) <EOL> def test_cleared ( self ) : <EOL> """<STR_LIT>""" <EOL> message = "<STR_LIT>" <EOL> category = RuntimeWarning <EOL> warnings . warn ( message = message , category = category ) <EOL> self . assertDictSubsets ( <EOL> self . flushWarnings ( ) , <EOL> [ { '<STR_LIT>' : category , '<STR_LIT:message>' : message } ] ) <EOL> self . assertEqual ( self . flushWarnings ( ) , [ ] ) <EOL> def test_unflushed ( self ) : <EOL> """<STR_LIT>""" <EOL> result = TestResult ( ) <EOL> case = Mask . MockTests ( '<STR_LIT>' ) <EOL> case . run ( result ) <EOL> warningsShown = self . flushWarnings ( [ Mask . MockTests . test_unflushed ] ) <EOL> self . assertEqual ( warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT:message>' ] , '<STR_LIT>' ) <EOL> self . assertIdentical ( warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , UserWarning ) <EOL> where = type ( case ) . test_unflushed . __code__ <EOL> filename = where . co_filename <EOL> lineno = where . co_firstlineno + <NUM_LIT:4> <EOL> self . assertEqual ( warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT:filename>' ] , filename ) <EOL> self . assertEqual ( warningsShown [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , lineno ) <EOL> self . assertEqual ( len ( warningsShown ) , <NUM_LIT:1> ) <EOL> def test_flushed ( self ) : <EOL> """<STR_LIT>""" <EOL> result = TestResult ( ) <EOL> case = Mask . MockTests ( '<STR_LIT>' ) <EOL> output = StringIO ( ) <EOL> monkey = self . patch ( sys , '<STR_LIT>' , output ) <EOL> case . run ( result ) <EOL> monkey . restore ( ) <EOL> self . assertEqual ( output . getvalue ( ) , "<STR_LIT>" ) <EOL> def test_warningsConfiguredAsErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> class CustomWarning ( Warning ) : <EOL> pass <EOL> result = TestResult ( ) <EOL> case = Mask . MockTests ( '<STR_LIT>' ) <EOL> case . category = CustomWarning <EOL> originalWarnings = warnings . filters [ : ] <EOL> try : <EOL> warnings . simplefilter ( '<STR_LIT:error>' ) <EOL> case . run ( result ) <EOL> self . assertEqual ( len ( result . errors ) , <NUM_LIT:1> ) <EOL> self . assertIdentical ( result . errors [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , case ) <EOL> self . assertTrue ( <EOL> result . errors [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . splitlines ( ) [ - <NUM_LIT:1> ] . endswith ( <EOL> "<STR_LIT>" ) ) <EOL> finally : <EOL> warnings . filters [ : ] = originalWarnings <EOL> def test_flushedWarningsConfiguredAsErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> class CustomWarning ( Warning ) : <EOL> pass <EOL> result = TestResult ( ) <EOL> case = Mask . MockTests ( '<STR_LIT>' ) <EOL> case . category = CustomWarning <EOL> originalWarnings = warnings . filters [ : ] <EOL> try : <EOL> warnings . simplefilter ( '<STR_LIT:error>' ) <EOL> case . run ( result ) <EOL> self . assertEqual ( result . errors , [ ] ) <EOL> finally : <EOL> warnings . filters [ : ] = originalWarnings <EOL> def test_multipleFlushes ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> self . assertEqual ( len ( self . flushWarnings ( ) ) , <NUM_LIT:1> ) <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> self . assertEqual ( len ( self . flushWarnings ( ) ) , <NUM_LIT:1> ) <EOL> def test_filterOnOffendingFunction ( self ) : <EOL> """<STR_LIT>""" <EOL> firstMessage = "<STR_LIT>" <EOL> firstCategory = UserWarning <EOL> def one ( ) : <EOL> warnings . warn ( firstMessage , firstCategory , stacklevel = <NUM_LIT:1> ) <EOL> secondMessage = "<STR_LIT>" <EOL> secondCategory = RuntimeWarning <EOL> def two ( ) : <EOL> warnings . warn ( secondMessage , secondCategory , stacklevel = <NUM_LIT:1> ) <EOL> one ( ) <EOL> two ( ) <EOL> self . assertDictSubsets ( <EOL> self . flushWarnings ( offendingFunctions = [ one ] ) , <EOL> [ { '<STR_LIT>' : firstCategory , '<STR_LIT:message>' : firstMessage } ] ) <EOL> self . assertDictSubsets ( <EOL> self . flushWarnings ( offendingFunctions = [ two ] ) , <EOL> [ { '<STR_LIT>' : secondCategory , '<STR_LIT:message>' : secondMessage } ] ) <EOL> def test_functionBoundaries ( self ) : <EOL> """<STR_LIT>""" <EOL> def warner ( ) : <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> warner ( ) <EOL> self . assertEqual ( <EOL> len ( self . flushWarnings ( offendingFunctions = [ warner ] ) ) , <NUM_LIT:3> ) <EOL> def test_invalidFilter ( self ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> self . assertRaises ( ValueError , self . flushWarnings , [ None ] ) <EOL> self . assertEqual ( len ( self . flushWarnings ( ) ) , <NUM_LIT:1> ) <EOL> def test_missingSource ( self ) : <EOL> """<STR_LIT>""" <EOL> package = FilePath ( self . mktemp ( ) . encode ( '<STR_LIT:utf-8>' ) ) . child ( b'<STR_LIT>' ) <EOL> package . makedirs ( ) <EOL> package . child ( b'<STR_LIT>' ) . setContent ( b'<STR_LIT>' ) <EOL> package . child ( b'<STR_LIT>' ) . setContent ( b'''<STR_LIT>''' ) <EOL> pathEntry = package . parent ( ) . path . decode ( '<STR_LIT:utf-8>' ) <EOL> sys . path . insert ( <NUM_LIT:0> , pathEntry ) <EOL> self . addCleanup ( sys . path . remove , pathEntry ) <EOL> from twisted_private_helper import missingsourcefile <EOL> self . addCleanup ( sys . modules . pop , '<STR_LIT>' ) <EOL> self . addCleanup ( sys . modules . pop , missingsourcefile . __name__ ) <EOL> package . child ( b'<STR_LIT>' ) . remove ( ) <EOL> missingsourcefile . foo ( ) <EOL> self . assertEqual ( len ( self . flushWarnings ( [ missingsourcefile . foo ] ) ) , <NUM_LIT:1> ) <EOL> def test_renamedSource ( self ) : <EOL> """<STR_LIT>""" <EOL> package = FilePath ( self . mktemp ( ) . encode ( '<STR_LIT:utf-8>' ) ) . child ( b'<STR_LIT>' ) <EOL> package . makedirs ( ) <EOL> package . child ( b'<STR_LIT>' ) . setContent ( b'<STR_LIT>' ) <EOL> package . child ( b'<STR_LIT>' ) . setContent ( b'''<STR_LIT>''' ) <EOL> pathEntry = package . parent ( ) . path . decode ( '<STR_LIT:utf-8>' ) <EOL> sys . path . insert ( <NUM_LIT:0> , pathEntry ) <EOL> self . addCleanup ( sys . path . remove , pathEntry ) <EOL> from twisted_private_helper import module <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> del sys . modules [ module . __name__ ] <EOL> try : <EOL> from importlib import invalidate_caches <EOL> except ImportError : <EOL> pass <EOL> else : <EOL> invalidate_caches ( ) <EOL> package . moveTo ( package . sibling ( b'<STR_LIT>' ) ) <EOL> from twisted_renamed_helper import module <EOL> self . addCleanup ( sys . modules . pop , '<STR_LIT>' ) <EOL> self . addCleanup ( sys . modules . pop , module . __name__ ) <EOL> module . foo ( ) <EOL> self . assertEqual ( len ( self . flushWarnings ( [ module . foo ] ) ) , <NUM_LIT:1> ) <EOL> class FakeWarning ( Warning ) : <EOL> pass <EOL> class CollectWarningsTests ( SynchronousTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_callsObserver ( self ) : <EOL> """<STR_LIT>""" <EOL> firstMessage = "<STR_LIT>" <EOL> secondMessage = firstMessage [ : : - <NUM_LIT:1> ] <EOL> events = [ ] <EOL> def f ( ) : <EOL> events . append ( '<STR_LIT>' ) <EOL> warnings . warn ( firstMessage ) <EOL> warnings . warn ( secondMessage ) <EOL> events . append ( '<STR_LIT>' ) <EOL> _collectWarnings ( events . append , f ) <EOL> self . assertEqual ( events [ <NUM_LIT:0> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( events [ <NUM_LIT:1> ] . message , firstMessage ) <EOL> self . assertEqual ( events [ <NUM_LIT:2> ] . message , secondMessage ) <EOL> self . assertEqual ( events [ <NUM_LIT:3> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( len ( events ) , <NUM_LIT:4> ) <EOL> def test_suppresses ( self ) : <EOL> """<STR_LIT>""" <EOL> output = StringIO ( ) <EOL> self . patch ( sys , '<STR_LIT>' , output ) <EOL> _collectWarnings ( lambda x : None , warnings . warn , "<STR_LIT:text>" ) <EOL> self . assertEqual ( output . getvalue ( ) , "<STR_LIT>" ) <EOL> def test_callsFunction ( self ) : <EOL> """<STR_LIT>""" <EOL> arguments = [ ] <EOL> value = object ( ) <EOL> def f ( * args , ** kwargs ) : <EOL> arguments . append ( ( args , kwargs ) ) <EOL> return value <EOL> result = _collectWarnings ( lambda x : None , f , <NUM_LIT:1> , '<STR_LIT:a>' , b = <NUM_LIT:2> , c = '<STR_LIT:d>' ) <EOL> self . assertEqual ( arguments , [ ( ( <NUM_LIT:1> , '<STR_LIT:a>' ) , { '<STR_LIT:b>' : <NUM_LIT:2> , '<STR_LIT:c>' : '<STR_LIT:d>' } ) ] ) <EOL> self . assertIdentical ( result , value ) <EOL> def test_duplicateWarningCollected ( self ) : <EOL> """<STR_LIT>""" <EOL> global __warningregistry__ <EOL> del __warningregistry__ <EOL> def f ( ) : <EOL> warnings . warn ( "<STR_LIT:foo>" ) <EOL> warnings . simplefilter ( '<STR_LIT:default>' ) <EOL> f ( ) <EOL> events = [ ] <EOL> _collectWarnings ( events . append , f ) <EOL> self . assertEqual ( len ( events ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( events [ <NUM_LIT:0> ] . message , "<STR_LIT:foo>" ) <EOL> self . assertEqual ( len ( self . flushWarnings ( ) ) , <NUM_LIT:1> ) <EOL> def test_immutableObject ( self ) : <EOL> """<STR_LIT>""" <EOL> key = object ( ) <EOL> sys . modules [ key ] = key <EOL> self . addCleanup ( sys . modules . pop , key ) <EOL> self . test_duplicateWarningCollected ( ) <EOL> def test_setWarningRegistryChangeWhileIterating ( self ) : <EOL> """<STR_LIT>""" <EOL> d = { } <EOL> class A ( object ) : <EOL> def __init__ ( self , key ) : <EOL> self . __dict__ [ '<STR_LIT>' ] = key <EOL> def __setattr__ ( self , value , item ) : <EOL> d [ self . _key ] = None <EOL> key1 = object ( ) <EOL> key2 = object ( ) <EOL> d [ key1 ] = A ( key2 ) <EOL> key3 = object ( ) <EOL> key4 = object ( ) <EOL> d [ key3 ] = A ( key4 ) <EOL> _setWarningRegistryToNone ( d ) <EOL> self . assertEqual ( set ( [ key1 , key2 , key3 , key4 ] ) , set ( d . keys ( ) ) ) </s>
