banishment

rollbar/test/twisted_tests/test_twisted.py

@@ -17,15 +17,63 @@
 try:
  from twisted.test import proto_helpers
  from twisted.trial import unittest
-
- from twisted_test_remote import RemoteCalculationFactory
+ from twisted.protocols import basic
+ from twisted.internet import protocol
+ from twisted.python import log
 
  TWISTED_INSTALLED = True
 except ImportError:
  TWISTED_INSTALLED = False
 
 
 if ALLOWED_PYTHON_VERSION and TWISTED_INSTALLED:
+ class Calculation(object):
+ def _make_ints(self, *args):
+ try:
+ return map(int, args)
+ except ValueError:
+ raise TypeError("Couldn't coerce arguments to integers: %s" % args)
+
+ def add(self, a, b):
+ a, b = self._make_ints(a, b)
+ return a + b
+
+ def subtract(self, a, b):
+ a, b = self._make_ints(a, b)
+ return a - b
+
+ def multiply(self, a, b):
+ a, b = self._make_ints(a, b)
+ return a * b
+
+ def divide(self, a, b):
+ a, b = self._make_ints(a, b)
+ return a / b
+
+ class CalculationProxy(object):
+ def __init__(self):
+ self.calc = Calculation()
+ for m in ['add', 'subtract', 'multiply', 'divide']:
+ setattr(self, 'remote_%s' % m, getattr(self.calc, m))
+
+ class RemoteCalculationProtocol(basic.LineReceiver):
+ def __init__(self):
+ self.proxy = CalculationProxy()
+
+ def lineReceived(self, line):
+ op, a, b = line.split()
+ op = getattr(self.proxy, 'remote_%s' % (op,))
+ try:
+ result = op(a, b)
+ except TypeError:
+ log.err()
+ self.sendLine("error")
+ else:
+ self.sendLine(str(result))
+
+ class RemoteCalculationFactory(protocol.Factory):
+ protocol = RemoteCalculationProtocol
+
  class TwistedFactoryTest(unittest.TestCase):
  def setUp(self):
  rollbar.init(TOKEN, 'twisted-test')