Форум сайта python.su
Читал главу документации twisted про тестирование. Там, как я понял, в основном про тестирование протоколов.
Имеются классы, которые работают c потоками, c базой данных, с процессами, используя функции из twisted. Для того чтобы они работали, нужно запустить reactor.
Не знаю как лучше написать тесты для этих классов, используя unittest.
Последний мой неработающий вариант
import unittest
from twisted.internet import reactor
from twisted.internet import protocol
from twisted.python import log
class ProcessProtocol(protocol.ProcessProtocol):
def __init__(self, callback):
self._callback = callback
self._out_buffer = ''
def outReceived(self, data):
self._out_buffer += data
def processEnded(self, reason):
self._callback(self._out_buffer)
class SimpleTest(unittest.TestCase):
def test_test1(self):
def callback(data):
self.assertEqual(data, 'Linux\n')
reactor.spawnProcess(ProcessProtocol(callback), 'uname', args=['uname'])
def test_test2(self):
def callback(data):
self.assertEqual(data, 'Linux')
reactor.spawnProcess(ProcessProtocol(callback), 'uname', args=['uname'])
reactor.callLater(1, unittest.main)
reactor.callLater(5, reactor.stop)
reactor.run()
Офлайн
Офлайн
Спасибо, не видел, посмотрю.
Все таки, что-то не получается
import unittest
from twisted.internet import protocol
from twisted.internet import reactor
class ProcessProtocol(protocol.ProcessProtocol):
def __init__(self, callback):
self._callback = callback
self._out_buffer = ''
def outReceived(self, data):
self._out_buffer += data
def processEnded(self, reason):
self._callback(self._out_buffer)
class SimpleTest(unittest.TestCase):
def test_test1(self):
def callback(data):
self.assertEqual(data, 'Linux\n')
reactor.spawnProcess(ProcessProtocol(callback), 'uname', args=['uname'])
def test_test2(self):
def callback(data):
self.assertEqual(data, 'Linux')
reactor.spawnProcess(ProcessProtocol(callback), 'uname', args=['uname'])
Отредактировано (Янв. 25, 2012 10:48:07)
Офлайн
Еще один вариант
import unittest
from twisted.internet import protocol
from twisted.internet import reactor
from twisted.internet import defer
class ProcessProtocol(protocol.ProcessProtocol):
def __init__(self):
self._out_buffer = ''
self.endedDeferred = None
def outReceived(self, data):
self._out_buffer += data
def processEnded(self, reason):
raise Exception('process ended')
self.endedDeferred.callback(self)
class SimpleTest(unittest.TestCase):
def test_test1(self):
p = ProcessProtocol()
d = p.endedDeferred = defer.Deferred()
reactor.spawnProcess(p, 'uname', args=['uname'])
def processEnded(p):
self.assertEqual(p._out_buffer, 'Linux\n')
return d.addCallback(processEnded)
def test_test2(self):
p = ProcessProtocol()
d = p.endedDeferred = defer.Deferred()
reactor.spawnProcess(p, 'uname', args=['uname'])
def processEnded(p):
self.assertEqual(p._out_buffer, 'Linux')
return d.addCallback(processEnded)
Офлайн
Забыл, что нужно поменять стандартный unittest на тот, что в twisted
Рабочий код
from twisted.internet import protocol
from twisted.internet import reactor
from twisted.internet import defer
from twisted.trial import unittest
class ProcessProtocol(protocol.ProcessProtocol):
def __init__(self):
self._out_buffer = ''
self.endedDeferred = None
def outReceived(self, data):
self._out_buffer += data
def processEnded(self, reason):
self.endedDeferred.callback(self)
class SimpleTest(unittest.TestCase):
def test_test1(self):
p = ProcessProtocol()
d = p.endedDeferred = defer.Deferred()
reactor.spawnProcess(p, 'uname', args=['uname'])
def processEnded(p):
self.assertEqual(p._out_buffer, 'Linux\n')
return d.addCallback(processEnded)
def test_test2(self):
p = ProcessProtocol()
d = p.endedDeferred = defer.Deferred()
reactor.spawnProcess(p, 'uname', args=['uname'])
def processEnded(p):
self.assertEqual(p._out_buffer, 'Linux')
return d.addCallback(processEnded)
Офлайн