2

我正在为基于龙卷风的 Web 应用程序编写一个测试模块。该应用程序使用电机作为 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

目前我无法让它工作，我想知道我想要做的事情是否可以使用当前版本的电机（1.2.1），mongomock（3.8.0）和龙卷风（4.5）。 3），或者我错过了什么？

感谢您的所有建议。

4

1 回答 1

3

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

我发现了以下问题：

  1. 电机似乎__delegate_class__有时会忽略，并实例化实际的 pymongoDatabase或Collection代替（例如delegate = _delegate or Collection(database.delegate, name)）
  2. Motor 包装了委托类并希望每个预期的属性都存在，以便它可以“异步”它们。
  3. 电机“不可知”游标依赖于 pymongo 游标的半私密细节，例如_refreshor __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
  1. 是你遇到的问题。这可以通过扫描所需属性并在缺失时定义假属性来避免：
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 必须保持原样。

  1. 我通过修补 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

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

于 2018-10-26T13:23:09.080 回答