/usr/share/pyshared/tftp/test/test_util.py is in python-txtftp 0.1~bzr38-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | '''
@author: shylent
'''
from tftp.util import SequentialCall, Spent, Cancelled
from twisted.internet.task import Clock
from twisted.trial import unittest
class CallCounter(object):
call_num = 0
def __call__(self):
self.call_num += 1
class SequentialCalling(unittest.TestCase):
def setUp(self):
self.f = CallCounter()
self.t = CallCounter()
self.clock = Clock()
def test_empty(self):
SequentialCall.run((), self.f, on_timeout=self.t, _clock=self.clock)
self.clock.pump((1,))
self.assertEqual(self.f.call_num, 0)
self.assertEqual(self.t.call_num, 1)
def test_empty_now(self):
SequentialCall.run((), self.f, on_timeout=self.t, run_now=True, _clock=self.clock)
self.clock.pump((1,))
self.assertEqual(self.f.call_num, 1)
self.assertEqual(self.t.call_num, 1)
def test_non_empty(self):
c = SequentialCall.run((1, 3, 5), self.f, run_now=True, on_timeout=self.t, _clock=self.clock)
self.clock.advance(0.1)
self.failUnless(c.active())
self.assertEqual(self.f.call_num, 1)
self.clock.pump((1,)*2)
self.failUnless(c.active())
self.assertEqual(self.f.call_num, 2)
self.clock.pump((1,)*3)
self.failUnless(c.active())
self.assertEqual(self.f.call_num, 3)
self.clock.pump((1,)*5)
self.failIf(c.active())
self.assertEqual(self.f.call_num, 4)
self.assertEqual(self.t.call_num, 1)
self.assertRaises(Spent, c.reschedule)
self.assertRaises(Spent, c.cancel)
def test_cancel(self):
c = SequentialCall.run((1, 3, 5), self.f, on_timeout=self.t, _clock=self.clock)
self.clock.pump((1,)*2)
self.assertEqual(self.f.call_num, 1)
c.cancel()
self.assertRaises(Cancelled, c.cancel)
self.assertEqual(self.t.call_num, 0)
self.assertRaises(Cancelled, c.reschedule)
def test_cancel_immediately(self):
c = SequentialCall.run((1, 3, 5), lambda: c.cancel(), run_now=True,
on_timeout=self.t, _clock=self.clock)
self.clock.pump((1,)*2)
self.assertRaises(Cancelled, c.cancel)
self.assertEqual(self.t.call_num, 0)
self.assertRaises(Cancelled, c.reschedule)
|