python - Tornado ，带 mongmock 的电机进行测试

Question

我正在为基于 Tornado 的 Web 应用程序编写一个测试模块。该应用程序使用 motor 作为 mongodb 连接器，我希望我的测试在临时数据库上运行。我在连接器客户端的 delegate_class 上使用模拟技术，如下所示:

import json
import mock
import motor
import tornado.ioloop
import tornado.testing
import mongomock
import myapp


patch_motor_client = mock.patch('motor.motor_tornado.MotorClient.__delegate_class__', new=mongomock.MongoClient)
patch_motor_database = mock.patch('motor.motor_tornado.MotorDatabase.__delegate_class__', new=mock.MagicMock)

patch_motor_client.start()
patch_motor_database.start()


class TestHandlerBase(tornado.testing.AsyncHTTPTestCase):
    """
    Base test handler
    """

    def setUp(self):
        # Create your Application for testing
        self.application = myapp.app.Application()
        super(TestHandlerBase, self).setUp()

    def get_app(self):
        return self.application

    def get_new_ioloop(self):
        return tornado.ioloop.IOLoop.instance()


class TestMyHandler(TestHandlerBase):
    def test_post_ok(self):
        """
        POST a resource is OK
        """
        post_args = {
            'data': 'some data here..'
        }

        response = self.fetch('myapi/v1/scripts', method='POST', body=json.dumps(post_args))

        # assert status is 201
        self.assertEqual(response.code, 201)

当我启动测试时，我收到此错误:

  File "/data/.virtualenvs/myapp/lib/python3.5/site-packages/motor/core.py", line 162, in __getitem__
    return db_class(self, name)
  File "/data/.virtualenvs/myapp/lib/python3.5/site-packages/motor/core.py", line 217, in __init__
    client.delegate, name, **kwargs)
  File "/data/.virtualenvs/myapp/lib/python3.5/site-packages/pymongo/database.py", line 102, in __init__
    read_concern or client.read_concern)
  File "/data/.virtualenvs/myapp/lib/python3.5/site-packages/pymongo/common.py", line 614, in __init__
    raise TypeError("codec_options must be an instance of "
TypeError: codec_options must be an instance of bson.codec_options.CodecOptions

目前我无法让它工作，我想知道我想要做的事情是否可以用当前版本的 motor (1.2.1)、mongomock (3.8.0) 和tornado 来实现(4.5.3)，或者我错过了什么？

感谢您的所有建议。

Answer 1

我只能让它与大量的猴子修补一起工作(我猜mock.patch-ing会是类似的，但我对恢复更改不感兴趣)。

我发现了以下问题:

1. Motor 有时似乎会忽略 __delegate_class__，并实例化实际的 pymongo Database 或 Collection (例如 delegate = _delegate 或 Collection) (数据库.委托(delegate)，名称))
2. Motor 包装委托(delegate)类并希望每个预期属性都存在，以便它可以“异步”它们。
3. 电机“不可知”游标依赖于 pymongo 游标的半私有(private)细节，例如 _refresh 或 __data，因为它们需要干预其操作(使它们异步)再说一次，因为他们是 IO)。 Mongomock游标要简单得多并且没有这样的属性

并像这样解决它们:

1. 可能可以在电机中修复，但这给我带来了麻烦，所以我不得不仅在集合级别启动“电机包装”(它在客户端或数据库中不起作用)，并破解以便 mongomock 数据库实例化电机包装的集合:

db = mongomock.Database(mongomock.MongoClient(), 'db_name')

# Monkeypatch get_collection so that collections are motor-wrapped
def create_motor_wrapped_mock_collection(
        name, codec_options=None, read_preference=None,
        write_concern=None, read_concern=None):
    if read_concern:
        raise NotImplementedError('Mongomock does not handle read_concern yet')
    collection = db._collections.get(name)
    if collection is None:
        delegate = mongomock.Collection(db, name, write_concern=write_concern)

        # wont be used, as we patch get_io_loop, but the MotorCollection ctor checks type
        fake_client = motor.motor_tornado.MotorClient()
        fake_db = motor.motor_tornado.MotorDatabase(fake_client, 'db_name')

        motor_collection = motor.motor_tornado.MotorCollection(fake_db, name, _delegate=delegate)
        collection = db._collections[name] = motor_collection
        collection.get_io_loop = lambda: tornado.ioloop.IOLoop.current()
    return collection

db.get_collection = create_motor_wrapped_mock_collection 

# Then use db in your code or patch it in

这是您遇到的问题吗？通过扫描所需的属性并在丢失时定义假属性可以避免这种情况:

def _prepare_for_motor_wrapping(cls, wrapper_cls):

    # Motor expects all attributes to exist on a delegate, to generate wrapped methods/attributes, even the ones we
    # won't need. This patches in dummy attributes/methods so that Motor wrapping can succeed

    def gen_fake_method(name, on):
        def fake_method(*args, **kwargs):
            raise NotImplementedError(name + ' on ' + on)
        return fake_method

    attrs = list(wrapper_cls.__dict__.items()) + list(motor.core.AgnosticBaseProperties.__dict__.items())
    for k, v in attrs:
        attr_name = getattr(v, 'attr_name', None) or k
        if not hasattr(cls, attr_name) and isinstance(v, motor.metaprogramming.MotorAttributeFactory):
            if isinstance(v, motor.metaprogramming.ReadOnlyProperty):
                setattr(cls, attr_name, None)
            elif isinstance(v, motor.metaprogramming.Async) or isinstance(v, motor.metaprogramming.Unwrap):
                setattr(cls, attr_name, gen_fake_method(attr_name, cls.__name__))
            else:
                raise RuntimeError('Dont know how to fake %s' % v)

# We must clear the cache, as classes might have been generated already during some previous import
motor.metaprogramming._class_cache = {}

_prepare_for_motor_wrapping(mongomock.Database, motor.core.AgnosticDatabase)
motor.motor_tornado.MotorDatabase = motor.motor_tornado.create_motor_class(motor.core.AgnosticDatabase)

_prepare_for_motor_wrapping(mongomock.Collection, motor.core.AgnosticCollection) 
motor.motor_tornado.MotorCollection = motor.motor_tornado.create_motor_class(motor.core.AgnosticCollection)

出于某种原因，MotorClient 必须保持完整。

我通过修补 to_list() 摆脱了困境，因为这是我在 coll.aggregate() 和 coll.find() 的结果上使用的唯一东西

def _patch_aggregate_cursor():

    def curs_to_docs(docs_future, curs_future):
        curs = curs_future.result()
        docs_future.set_result(list(curs))

    def to_list(self, *args):
        mock_cursor_future = self.collection._async_aggregate(self.pipeline)
        docs_future = self._framework.get_future(self.get_io_loop())
        self._framework.add_future(
            self.get_io_loop(),
            mock_cursor_future,
            curs_to_docs, docs_future)
        return docs_future

    motor.core.AgnosticAggregationCursor.to_list = to_list


def _patch_generic_cursor():

    def to_list(self, *args):
        docs = list(self.delegate)
        docs_future = self._framework.get_future(self.get_io_loop())
        docs_future.set_result(docs)
        return docs_future

    motor.core.AgnosticCursor.to_list = to_list

这一切可能是不完整且脆弱的，所以我让你判断是否值得付出努力。