1 from calculus.client_2 import RemoteCalculationClient, ClientTimeoutError
3 from twisted.internet import task
4 from twisted.trial import unittest
5 from twisted.test import proto_helpers
9 class ClientCalculationTestCase(unittest.TestCase):
11 self.tr = proto_helpers.StringTransportWithDisconnection()
12 self.clock = task.Clock()
13 self.proto = RemoteCalculationClient()
14 self.tr.protocol = self.proto
15 self.proto.callLater = self.clock.callLater
16 self.proto.makeConnection(self.tr)
19 def _test(self, operation, a, b, expected):
20 d = getattr(self.proto, operation)(a, b)
21 self.assertEqual(self.tr.value(), '%s %d %d\r\n' % (operation, a, b))
23 d.addCallback(self.assertEqual, expected)
24 self.proto.dataReceived("%d\r\n" % (expected,))
29 return self._test('add', 7, 6, 13)
32 def test_subtract(self):
33 return self._test('subtract', 82, 78, 4)
36 def test_multiply(self):
37 return self._test('multiply', 2, 8, 16)
40 def test_divide(self):
41 return self._test('divide', 14, 3, 4)
44 def test_timeout(self):
45 d = self.proto.add(9, 4)
46 self.assertEqual(self.tr.value(), 'add 9 4\r\n')
47 self.clock.advance(self.proto.timeOut)
48 return self.assertFailure(d, ClientTimeoutError)