Flask 单元测试——模拟 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 语言编写的，因此进行部分修补（又名 "monkey patching"）并不容易。例如，如果您尝试简单地修补 aerospike.Client.put.

，您将得到一个 TypeError: can't set attributes of built-in/extension type

一种方法是创建一个模拟客户端对象来替换或子class 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 class 而不是 aerospike.Client class. MockClient 实例的 put 方法也被修补，因此可以断言 put 方法在调用 upload_data 之后被调用。

为了测试 Aerospike 客户端引发的异常是否从 upload_data 函数中重新引发，可以修改 MockClient class 以显式引发异常：

# 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()

Flask 单元测试——模拟 Aerospike DB

Flask unit tests -- mocking up Aerospike DB

python

unit-testing

flask

aerospike