<s> import datetime <EOL> from twisted . trial import unittest <EOL> from twisted . internet import task <EOL> from twisted . internet . interfaces import IReactorTime <EOL> from zope . interface import implements <EOL> from txtorcon . addrmap import AddrMap <EOL> from txtorcon . interface import IAddrListener <EOL> class AddrMapTests ( unittest . TestCase ) : <EOL> implements ( IAddrListener ) <EOL> fmt = '<STR_LIT>' <EOL> def test_parse ( self ) : <EOL> """<STR_LIT>""" <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am = AddrMap ( ) <EOL> am . update ( line ) <EOL> addr = am . find ( '<STR_LIT>' ) <EOL> self . assertTrue ( addr . ip == '<STR_LIT>' or addr . ip . exploded == '<STR_LIT>' ) <EOL> self . assertEqual ( addr . expires . ctime ( ) , nowutc . ctime ( ) ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> self . assertEqual ( addr . expires . ctime ( ) , nowutc . ctime ( ) ) <EOL> am . scheduler . getDelayedCalls ( ) [ <NUM_LIT:0> ] . cancel ( ) <EOL> def test_expires ( self ) : <EOL> """<STR_LIT>""" <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> clock . advance ( <NUM_LIT:10> ) <EOL> self . assertTrue ( '<STR_LIT>' not in am . addr ) <EOL> def test_expires_never ( self ) : <EOL> """<STR_LIT>""" <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> line = '<STR_LIT>' <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> self . assertEqual ( len ( clock . getDelayedCalls ( ) ) , <NUM_LIT:0> ) <EOL> def test_expires_old ( self ) : <EOL> """<STR_LIT>""" <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = - <NUM_LIT:10> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = - <NUM_LIT:10> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> clock . advance ( <NUM_LIT:0> ) <EOL> self . assertTrue ( '<STR_LIT>' not in am . addr ) <EOL> def test_expires_with_update ( self ) : <EOL> """<STR_LIT>""" <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> self . assertTrue ( am . find ( '<STR_LIT>' ) ) <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:20> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:20> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> clock . advance ( <NUM_LIT:10> ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> clock . advance ( <NUM_LIT:10> ) <EOL> self . assertTrue ( '<STR_LIT>' not in am . addr ) <EOL> def test_8596_cached_1 ( self ) : <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> line = '<STR_LIT>' <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> self . assertEqual ( len ( clock . getDelayedCalls ( ) ) , <NUM_LIT:0> ) <EOL> def test_8596_cached_2 ( self ) : <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> line = '<STR_LIT>' <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' in am . addr ) <EOL> self . assertEqual ( len ( clock . getDelayedCalls ( ) ) , <NUM_LIT:1> ) <EOL> def test_8596_cached_3 ( self ) : <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> line = '<STR_LIT>' <EOL> am . update ( line ) <EOL> self . assertTrue ( '<STR_LIT>' not in am . addr ) <EOL> self . assertEqual ( len ( clock . getDelayedCalls ( ) ) , <NUM_LIT:0> ) <EOL> def addrmap_expired ( self , name ) : <EOL> self . expires . append ( name ) <EOL> def addrmap_added ( self , addr ) : <EOL> self . addrmap . append ( addr ) <EOL> def test_double_add_listener ( self ) : <EOL> am = AddrMap ( ) <EOL> am . add_listener ( self ) <EOL> am . add_listener ( self ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( am . listeners ) ) <EOL> def test_listeners ( self ) : <EOL> self . expires = [ ] <EOL> self . addrmap = [ ] <EOL> clock = task . Clock ( ) <EOL> am = AddrMap ( ) <EOL> am . scheduler = IReactorTime ( clock ) <EOL> am . add_listener ( self ) <EOL> now = datetime . datetime . now ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> nowutc = datetime . datetime . utcnow ( ) + datetime . timedelta ( seconds = <NUM_LIT:10> ) <EOL> line = '<STR_LIT>' % ( now . strftime ( self . fmt ) , nowutc . strftime ( self . fmt ) ) <EOL> am . update ( line ) <EOL> a = am . find ( '<STR_LIT>' ) <EOL> self . assertEqual ( self . addrmap , [ a ] ) <EOL> clock . advance ( <NUM_LIT:10> ) <EOL> self . assertEqual ( self . expires , [ '<STR_LIT>' ] ) </s>
