DjangoBB LoFi version

Полная версия: Тестирование приложение использующих twisted

Начало » Python для экспертов » Тестирование приложение использующих twisted

svas

Янв. 24, 2012 18:42:26

Читал главу документации 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()

Тесты здесь простые, в реальном коде тестируются классы. Кто знает как лучше это тестировать?

s0rg

Янв. 24, 2012 20:13:24

http://twistedmatrix.com/documents/current/core/howto/testing.html

svas

Янв. 25, 2012 10:32:39

Спасибо, не видел, посмотрю.

Все таки, что-то не получается

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'])

Второй тест должен не пройти, но проходят оба теста

svas

Янв. 25, 2012 12:09:53

Еще один вариант

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)

В функции ProcessProtocol::processEnded возбуждается исключение. Но походу до этой функции дело не доходит

svas

Янв. 25, 2012 12:22:56

Забыл, что нужно поменять стандартный 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)