2
from twisted.trial import unittest
3
from twisted.internet import task
4
from twisted.internet.interfaces import IReactorTime
5
from zope.interface import implements
7
from txtorcon.addrmap import AddrMap
8
from txtorcon.interface import IAddrListener
11
class AddrMapTests(unittest.TestCase):
12
implements(IAddrListener)
14
fmt = '%Y-%m-%d %H:%M:%S'
18
Make sure it's parsing things properly.
21
now = datetime.datetime.now() + datetime.timedelta(seconds=10)
22
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
23
## we need to not-barf on extra args as per control-spec.txt
24
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s" FOO=bar BAR=baz' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
27
addr = am.find('www.example.com')
29
self.assertTrue(addr.ip == '72.30.2.43' or addr.ip.exploded == '72.30.2.43')
30
## maybe not the most robust, should convert to
31
## seconds-since-epoch instead? the net result of the parsing
32
## is we've rounded to seconds...
33
self.assertEqual(addr.expires.ctime(), nowutc.ctime())
35
## this will have resulted in an expiry call, which we need to
36
## cancel to keep the reactor clean. for consistency, we use
37
## the IReactorTime interface from AddrMap
38
am.scheduler.getDelayedCalls()[0].cancel()
40
def test_expires(self):
42
Test simply expiry case
47
am.scheduler = IReactorTime(clock)
49
now = datetime.datetime.now() + datetime.timedelta(seconds=10)
50
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
51
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s"' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
55
self.assertTrue('www.example.com' in am.addr)
56
## advance time past when the expiry should have occurred
58
self.assertTrue('www.example.com' not in am.addr)
60
def test_expires_never(self):
62
Test a NEVER expires line, as in what we'd get a startup for a
63
configured address-mapping.
68
am.scheduler = IReactorTime(clock)
70
line = 'www.example.com 72.30.2.43 "NEVER"'
73
self.assertTrue('www.example.com' in am.addr)
74
self.assertEqual(len(clock.getDelayedCalls()), 0)
76
def test_expires_old(self):
78
Test something that expires before "now"
83
am.scheduler = IReactorTime(clock)
85
now = datetime.datetime.now() + datetime.timedelta(seconds=-10)
86
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=-10)
87
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s"' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
90
self.assertTrue('www.example.com' in am.addr)
91
## arguably we shouldn't even have put this in the map maybe,
92
## but the reactor needs to iterate before our expiry callback
93
## gets called (right away) which is simulated by the
96
self.assertTrue('www.example.com' not in am.addr)
98
def test_expires_with_update(self):
100
This test updates the expiry time and checks that we properly
101
delay our expiry callback.
105
am.scheduler = IReactorTime(clock)
107
## now do an actual update to an existing Addr entry.
108
now = datetime.datetime.now() + datetime.timedelta(seconds=10)
109
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
110
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s"' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
112
self.assertTrue(am.find('www.example.com'))
115
now = datetime.datetime.now() + datetime.timedelta(seconds=20)
116
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=20)
117
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s"' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
119
self.assertTrue('www.example.com' in am.addr)
121
## advance time by the old expiry value and we should still
124
self.assertTrue('www.example.com' in am.addr)
126
## ...but advance past the new expiry (another 10 seconds) and
129
self.assertTrue('www.example.com' not in am.addr)
131
def test_8596_cached_1(self):
134
am.scheduler = IReactorTime(clock)
136
line = 'example.com 192.0.2.1 NEVER CACHED="YES"'
139
self.assertTrue('example.com' in am.addr)
140
self.assertEqual(len(clock.getDelayedCalls()), 0)
142
def test_8596_cached_2(self):
145
am.scheduler = IReactorTime(clock)
147
line = 'example.com 192.0.43.10 "2013-04-03 22:29:11" EXPIRES="2013-04-03 20:29:11" CACHED="NO"'
150
self.assertTrue('example.com' in am.addr)
151
self.assertEqual(len(clock.getDelayedCalls()), 1)
153
def test_8596_cached_3(self):
156
am.scheduler = IReactorTime(clock)
158
line = 'example.invalid <error> "2013-04-03 08:28:52" error=yes EXPIRES="2013-04-03 06:28:52" CACHE="NO"'
161
self.assertTrue('example.invalid' not in am.addr)
162
self.assertEqual(len(clock.getDelayedCalls()), 0)
164
def addrmap_expired(self, name):
165
self.expires.append(name)
167
def addrmap_added(self, addr):
168
self.addrmap.append(addr)
170
def test_listeners(self):
176
am.scheduler = IReactorTime(clock)
177
am.add_listener(self)
179
now = datetime.datetime.now() + datetime.timedelta(seconds=10)
180
nowutc = datetime.datetime.utcnow() + datetime.timedelta(seconds=10)
181
line = 'www.example.com 72.30.2.43 "%s" EXPIRES="%s"' % (now.strftime(self.fmt), nowutc.strftime(self.fmt))
185
## see if our listener got an update
186
a = am.find('www.example.com')
187
self.assertEqual(self.addrmap, [a])
189
## advance time past when the expiry should have occurred
192
## check that our listener got an expires event
193
self.assertEqual(self.expires, ['www.example.com'])
added
removed
test/test_addrmap.py
