烧瓶单元测试-模拟Aerospike DB

Question

我有以下端点，

    @developer_blueprint.route("/init_db", methods=["POST"])
    def initialize_database():
       try:
          upload_data(current_app)
          logger.debug("Database entries upload.")
          return jsonify({"result": "Database entries uploaded."}), 201
       except Exception as e:
          return jsonify({"error": str(e)})

    def upload_data(app):
       with open("src/core/data/data.json") as data_file:
          data = json.load(data_file)
          try:
             current_app.db.put(("somenamespace", "test", "default"), data, None)
          except Exception as e:
             raise e

我正在尝试弄清楚如何对它进行单元测试（我们需要覆盖我们的代码）。我只是模拟app.db吗？我该怎么办？

任何建议将不胜感激。

Answer 1

[使用unittest.mock之类的工具来模拟数据库调用进行单元测试，然后在容器或VM中运行Aerospike进行端到端测试是很常见的。

但是，请记住，Aerospike Python客户端库是用C编写的，以获得更好的性能，因此进行部分修补（又称“猴子修补”）并不容易。例如，如果您尝试简单地修补TypeError: can't set attributes of built-in/extension type，则将获得aerospike.Client.put。

一种方法是创建一个模拟客户端对象，以替换或细分Aerospike客户端对象。该模拟对象的实现取决于您的代码和要测试的案例。

[使用下面的示例代码，其中app.db是Aerospike客户端库的实例：

# example.py
import aerospike
import json

class App(object):
    db = None
    def __init__(self):
        config = {'hosts': [('127.0.0.1', 3000)]}
        self.db = aerospike.client(config).connect()

def upload_data(app):
    with open("data.json") as data_file:
        data = json.load(data_file)

    try:
        app.db.put(("ns1", "test", "default"), data, None)
    except Exception as e:
            raise e

if __name__ == "__main__":
    app = App()
    upload_data(app)

[为upload_data函数编写单元测试，让我们假设您要测试一个成功案例，该案例被确定为意味着调用了put方法并且没有引发异常：
# test.py
from unittest import TestCase, main
from unittest.mock import PropertyMock, patch
from example import App, upload_data
from aerospike import Client, exception


class MockClient(Client):
    def __init__(self, *args, **kwargs):
        pass

    def put(self, *args, **kwargs):
        return 0


class ExampleTestCase(TestCase):
    def test_upload_data_success(self):
        with patch.object(App, 'db', new_callable=PropertyMock) as db_mock:
            db_mock.return_value = client = MockClient()
            app = App()
            with patch.object(client, 'put') as put_mock:
                upload_data(app)
                put_mock.assert_called()


if __name__ == '__main__':
    main()

在test_upload_data_success方法中，App.db属性使用MockClient类而不是aerospike.Client类进行了修补。还修补了put实例的MockClient方法，以便可以断言在调用put之后将调用upload_data方法。
为了测试从upload_data函数重新引发Aerospike客户端引发的异常，可以修改MockClient类以显式引发异常：
# test.py
from unittest import TestCase, main
from unittest.mock import PropertyMock, patch
from example import App, upload_data
from aerospike import Client, exception


class MockClient(Client):
    def __init__(self, *args, **kwargs):
        self.put_err = None
        if 'put_err' in kwargs:
            self.put_err = kwargs['put_err']

    def put(self, *args, **kwargs):
        if self.put_err:
            raise self.put_err
        else:
            return 0


class ExampleTestCase(TestCase):
    def test_upload_data_success(self):
        with patch.object(App, 'db', new_callable=PropertyMock) as db_mock:
            db_mock.return_value = client = MockClient()
            app = App()
            with patch.object(client, 'put') as put_mock:
                upload_data(app)
                put_mock.assert_called()

    def test_upload_data_error(self):
        with patch.object(App, 'db', new_callable=PropertyMock) as db_mock:
            db_mock.return_value = MockClient(put_err=exception.AerospikeError)
            app = App()
            with self.assertRaises(exception.AerospikeError):
                upload_data(app)


if __name__ == '__main__':
    main()