pytest.mock - 图1

  1. __init__：
  3.     name: mock 对象的标识
  5.     spec: 设置对象属性
  7.     return_value: 对象调用时的返回值
  9.     side_effect: 覆盖return_value, 当对象被调用时返回
  11. Assert_method:
  13.     assert_called_with: 断言 mock 对象的参数是否正确
  15.     assert_called_once_with: 检查某个对象如果被调用多次抛出异常，只允许一次
  17.     assert_any_call: 检查对象在全局过程中是否调用了该方法
  19.     assert_has_calls: 检查调用的参数和顺序是否正确
  21. Management:
  23.     attach_mock: 添加对象到另一个mock对象中
  25.     configure_mock: 重新设置对象的返回值
  27.     mock_add_spec: 新增对象属性
  29.     reset_mock: 重置对象
  31. Count:
  33.     called: 对象调用的访问器
  35.     call_count: 对象调用次数
  37.     call_args: 对象调用时的参数（最近）
  39.     call_args_list: 获取对用时所有的参数list
  41.     method_calls: 统计对象调用的所有方法，返回list
  43.     mock_calls: 统计工厂调用、方法调用

1. 前言

微服务架构下，由于各类服务开发进度的不一致，导致联调工作经常会存在不确定性，进而导致项目延期
在实际工作中，为了保证项目进度，我们经常需要针对部分未完成模块及不稳定模块采用 Mock 方式，以验证已开发完的模块
本篇文章将介绍 Python 实现 Mock 的几种常见方式

2. Mock 介绍

Mock 测试：在测试验证过程中，对于那些尚未完成或不稳定的对象，用一个虚拟对象来替代，以便测试的测试方法
因此，这个虚拟的对象是 Mock 对象，Mock 对象是真实对象在调试期间的代替品
它的优势包含：

  • 前、后端并行开发
  • 模拟无法访问的资源
  • 隔离系统，避免脏数据干扰测试结果

    3.1 mock

    在 Python 3.3 之前使用 mock，需要先安装依赖

# 安装mock依赖<br />pip3 install mock<br />
项目地址：
https://github.com/testing-cabal/mock
假设 Product 类中有 2 个方法

  • get_product_status_by_id
  • buy_product

其中，get_product_status_by_id 方法还没有实现；buy_product 方法依赖于 get_product_status_by_id 方法的返回值
`# product_impl.py

class Product(object):

  1. def __init__(self):<br />        pass
  3. def get_product_status_by_id(self, product_id):<br />        '''<br />        通过商品id获取产品信息（Mock）<br />        :return:<br />        '''<br />        # 待实现查询数据库的业务逻辑<br />        pass
  5. def buy_product(self, product_id):<br />        '''<br />        购买产品（真实逻辑）<br />        :return:<br />        '''<br />        # 产品信息<br />        # {'id':1,'name':'苹果','num':23}<br />        product = self.get_product_status_by_id(product_id)
  7.     if product.get('num') >= 1:<br />            result = {'status': 0, 'msg': '购买成功！'}<br />        else:<br />            result = {'status': 1, 'msg': '购买失败，库存不足！'}
  9.     return result`<br />Mock 的步骤如下：
  • 导入使用 mock 中的 patch 方法
  • 作为测试方法的装饰器，对 get_product_status_by_id 方法进行 Mock，方法参数为 Mock 对象
  • 测试方法中，对该 Mock 对象设置一个返回值
  • 调用并断言
    from mock import patch
from mock_.product_impl import Product
@patch('mock_.product_impl.Product.get_product_status_by_id')
def test_succuse(mock_get_product_status_by_id):
  # Mock方法，指定一个返回值
  mock_get_product_status_by_id.return_value = {'id': 1, 'name': '苹果', 'num': 23}
  product = Product()
  assert product.buy_product(1).get('status') == 0
    需要注意的是，Mock 此方法的时候，必须制定该方法的完整路径
    使用 @patch.object 同样能完成 Mock，不同的是，@patch.object 包含 2 个参数
    第一个参数为该方法所在的类；第二个参数为方法名
    `from mock import patch

from mock_.product_impl import Product

Mock一个方法
# @patch.object：对象、方法名
@patch.object(Product, ‘get_product_status_by_id’)
def test_succuse(mock_get_product_status_by_id):
# Mock方法，指定一个返回值
mock_get_product_status_by_id.return_value = {‘id’: 1, ‘name’: ‘苹果’, ‘num’: 23}

product = Product()

assert product.buy_product(1).get('status') == 0`

3.2 unittest.mock

Python 3.3 之后，mock 作为标准库，已经内置到 unittest 中了
还是以 3.1 的场景为例，使用 unittest 编写一个测试用例
Mock 步骤如下：

  • 导入 unittest 框架中的 mock 文件
  • 实例化 Product 对象
  • mock.Mock(return_value=*) 方法
    对 get_product_status_by_id 方法进行 Mock

  • 调用并断言

    import unittest
from unittest import mock
from unittest_mock.product_impl import Product
class TestProduct(unittest.TestCase):
  def test_success(self):
      # 成功结果
      mock_success_value = {'id': 1, 'name': '苹果', 'num': 23}
      product = Product()
      product.get_product_status_by_id = mock.Mock(return_value=mock_success_value)
      # 调用实际函数
      assert product.buy_product(1).get('status') == 0
if __name__ == '__main__':
  unittest.main()

    3.3 pytest.mock

    相比 unittest，pytest 由于强大的插件支持，用户群体可能更大！
    如果项目本身使用的框架是 pytest，则 Mock 更建议使用 pytest-mock 这个插件
    # pytest依赖<br />pip3 install pytest<br />
    Mock 步骤如下：

  • 使用 pytest 编写测试方法，参数为 mocker

  • 实例化 Product 对象
  • 使用 mocker.patch() 方法对 get_product_status_by_id 方法进行 Mock，并设置返回值
  • 调用并断言
    import pytest
from pytest_mock_.product_impl import Product
def test_buy_product_success(mocker):
  '''
  购买成功Mock
  :param mocker:
  :return:
  '''
  # 实例化一个产品对象
  product = Product()
  # 对Product中的方法的返回值进行Mock
  mock_value = {'id': 1, 'name': '苹果', 'num': 23}
  # Mock方法
  # 注意：需要指定方法的完整路径
  # mocker.patch 的第一个参数必须是模拟对象的具体路径，第二个参数用来指定返回值
  product.get_product_status_by_id = mocker.patch('product_impl.Product.get_product_status_by_id',
                                                  return_value=mock_value)
  # 调用购买产品的方法
  result = product.buy_product(1)
  assert result.get('status') == 0
    需要注意的是，mocker.patch 方法第一个参数必须是 Mock 对象的完整路径

    实例

    ```bash

@fixture def users(mocker): users = [] for x in range(1, 5): user = mocker.Mock() user.username = ‘user{}’.format(x) users.append(user) return users

@fixture def mmrepquotaout(): outpath = resource_filename( __name, ‘data/mmrepquota.output’ ) with open(out_path)as f: data = f.read() return data.encode(‘utf-8’)

@mark.django_db def test_billing_storage(mocker, today, bill_group, storage_policy, user_billgroup_map1, user_discount, settings, users, mmrepquota_out): mocker.patch(‘lico.core.accounting.charge_storage.check_call’) ## 覆盖了 里面的 check_call, return_value 是 None mocker.patch(‘lico.core.accounting.charge_storage.check_output’, ## 覆盖了 里面的 check_output, return_value 是 mmrepquota_out return_value=mmrepquota_out)

mock_client = mocker.patch('lico.core.contrib.client.Client')
mork_user = mock_client.return_value. \
    auth_client.return_value.fetch_passwd
mork_user.return_value.uid = 1141

mork_userlist = mock_client.return_value. \
    user_client.return_value.get_user_list
mork_userlist.return_value = users

StorageBilling().billing(today)
storage_state = StorageBillingStatement.objects.get(
    username='user1', billing_date=today, path='/root')
deposit = Deposit.objects.get(user='user1', billing_id=storage_state.id)

record = StorageBillingRecord.objects.filter(
    username='user1', billing_date=today).exists()
assert record is True
assert storage_state.bill_group_name == 'test_bill_group'
assert storage_state.storage_cost == 29.18
assert storage_state.billing_cost == 23.34
assert float(storage_state.discount) == 0.8
assert deposit.credits == -23.34
assert deposit.balance == 76.66
bill_group.refresh_from_db()
assert round(bill_group.balance, 2) == 53.32

`lico.core.accounting.charge_storage`    文件如下<br />虽然  `check_call, check_output`   来自 `subprocess`  ， 但是 在 文件里使用了， 我们 用 mock是 覆盖了 文件的调用。 调用的时候，      `check_call`  返回 `None` .    `check_output` 返回 `mmrepquota_out` 
```python
# Copyright 2020-present Lenovo
# Confidential and Proprietary

import logging
import os
import re
from subprocess import (
    CalledProcessError, check_call, check_output, list2cmdline,
)

from dateutil.tz import tzutc
from django.conf import settings
from django.db import IntegrityError, transaction
from django.utils.timezone import now

from lico.core.accounting.exceptions import (
    CreateDepositException, CreateStorageBillingRecordException,
    CreateStorageBillingStatementException, GetStorageQuotaFailedException,
)
from lico.core.accounting.utils import get_user_discount

from .models import (
    BillGroup, BillGroupStoragePolicy, Deposit, StorageBillingRecord,
    StorageBillingStatement, UserBillGroupMapping,
)

logger = logging.getLogger(__name__)


class StorageBilling(object):
    def __init__(self):
        self.global_path_data = {}

    def billing(self, local_date):
        billing_date = local_date.astimezone(tzutc())
        with open(os.devnull) as f:
            check_call(
                ['bash', '--login', '-c',
                 list2cmdline(
                     ['which', settings.ACCOUNTING.STORAGE.GPFS_STORAGE_CMD]
                 )],
                stdout=f, stderr=f
            )

        billing_users = []
        from lico.core.contrib.client import Client
        user_list = Client().user_client().get_user_list(
            date_joined__lte=billing_date)
        username_list = [x.username for x in user_list]
        record_user_name = StorageBillingRecord.objects.filter(
            billing_date=billing_date).values_list('username', flat=True)
        for username in username_list:
            if username in record_user_name:
                logger.info(
                    'storage was already charged for user %s on billing'
                    ' date %s.', username,
                    billing_date.strftime("%Y-%m-%d"))
            else:
                ret_flag = self._storage_charge(billing_date, username)
                if ret_flag:
                    billing_users.append(username)

        return billing_users