Использование Nose для тестирования зависимого от txmongo кода - PullRequest
Новая
       20

Использование Nose для тестирования зависимого от txmongo кода

1 голос
/ 04 ноября 2010

Я хочу использовать нос, чтобы протестировать приложение, которое я пишу, используя twisted и txmongo.Я даже не могу получить простые варианты использования, такие как следующие рабочие: из механизма переноса импорта nose.twistedtools, отложенный, запись по протоколу импорта нитей / нитей из twisted.internet, отложенный импорт импорт txmongo 

log = logging.getLogger("common.test.test_db")

conn = txmongo.lazyMongoConnectionPool('localhost', 27017, 4)

@deferred()
def test_mongo():
    tdb = conn.test

    @defer.inlineCallbacks
    def cb(oid):
        assert oid
        obj = yield tdb.test.find({"_id":oid})
        log.error("In callback")
        assert obj 

    d = tdb.test.save({"s":1, "b":2})
    d.addCallback(cb)

    return d

Однако это всегда возвращает: 

E
======================================================================
ERROR: common.test.test_db.test_mongo
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Volumes/Users/jce/.pyenv/celery/lib/python2.6/site-packages/nose/case.py", line 186, in runTest
    self.test(*self.arg)
  File "/Volumes/Users/jce/.pyenv/celery/lib/python2.6/site-packages/nose/twistedtools.py", line 138, in errback
    failure.raiseException()
  File "/Volumes/Users/jce/.pyenv/celery/lib/python2.6/site-packages/twisted/python/failure.py", line 326, in raiseException
    raise self.type, self.value, self.tb
RuntimeWarning: not connected

----------------------------------------------------------------------
Ran 1 test in 0.006s

FAILED (errors=1)

Я пытался вручную добавить вызов threadaded_reactor (), но это не помогло.

edit

Я удалил ""ленивые" соединения и модифицированный код, и теперь он работает ... Мне все еще любопытно, почему "ленивый" не работает.Рабочий код выглядит следующим образом: 

dbconn = txmongo.MongoConnectionPool('localhost', 27017, 4)

@deferred()
def test_mongo():
    @defer.inlineCallbacks
    def cb(conn):
        tdb = conn.test
        oid = yield tdb.test.save({"s":1, "b":2})
        assert oid
        log.error(str(oid))
        obj = yield tdb.test.find({"_id":oid})
        assert obj 
        log.error(str(obj))
    dbconn.addCallback(cb)
    return dbconn

1 Ответ

1 голос
/ 05 ноября 2010

MongoConnectionPool вернет отложенное, которое запускается, когда соединение установлено, передавая обработчик соединения в качестве аргумента для обратного вызова. Вы должны conn = yield MongoConnectionPool().

lazyMongoConnectionPool вернет обработчик соединения напрямую, не дожидаясь установления соединения.

Lazy обычно используется веб-серверами и другими службами, которым не требуется немедленное подключение при запуске службы. Если вы хотите сделать это, не используйте ленивый метод.

...