使用 Django REST 框架进行 PyTest:从零到大师
这篇文章适合哪些人阅读?
我为什么写这篇文章
为什么使用 PyTest
如何构建测试
PyTest 设置
测试类型
我们为什么要测试
正确的测试流程
我应该在单元测试中测试什么?
常见测试实用程序
如何组织测试
测试示例
conftest.py
我们的测试套件的覆盖范围
关于不稳定的测试
进行中
这篇文章适合哪些人阅读?
适合那些想要学习成为高效测试人员所需的一切知识、没有 QA 团队支持或只想控制代码质量的开发人员。
我为什么写这篇文章
测试是开发人员代码生命周期的关键组成部分,无论对于整个团队还是孤军奋战,然而全面的测试指南却未能跟上 Django REST 框架的发展步伐。我目前还没有找到任何资源能够涵盖我认为测试标准应该为应用程序做到的所有内容。这些内容包括:
- 拥有足够快的测试速度,以便为开发人员提供高效的代码和测试流程以及 CI/CD 管道。
- 进行足够全面的测试来隔离代码片段中出现的问题以及我们正在测试的代码所依赖的外部/内部代码中出现的问题。
- 留下一套清晰实用的工具来精确定位相关的单个或多个测试,从而为开发人员提供快速的工作流程。
在本文中,我提出了一种测试方法、一个测试结构、一套用于开发过程中的测试工具,并展示了如何测试 Django 应用程序的每个常用部分。这项工作仍在进行中,我期待大家的反馈和建议,任何有意义的评论都将大有裨益。
为什么使用 PyTest
- 更少的样板(您不必记住类似 with 的断言语句
unittest
) - 可自定义的错误详细信息
- 独立于所用 IDE 的自动发现
- 跨团队标准化标记
- 很棒的终端命令
- DRY 的参数化
- 庞大的插件社区所有这些以及更多的功能使成为一名熟练的 PyTest 测试人员成为一项改变游戏规则的技能。
如何构建测试
对于防扩展应用程序来说,最好创建一个tests
文件夹,并在其中为我们的 django 项目中的每个应用程序创建一个文件夹:
...
├── tests
│ ├── __init__.py
│ ├── test_app1
│ │ ├── __init__.py
│ │ ├── conftest.py
│ │ ├── factories.py
│ │ ├── e2e_tests.py
│ │ ├── test_models.py <--
│ │ ├── test_signals.py <--
│ │ ├── test_serializers.py
│ │ ├── test_utils.py
│ │ ├── test_views.py
│ │ └── test_urls.py
│ │
│ └── ...
└── ...
PyTest 设置
pytest.ini
PyTest 设置配置可以在以下文件中[pytest]
或在setup.cfg
以下文件中设置[tool:pytest]
:
:in pytest.ini
[pytest]
DJANGO_SETTINGS_MODULE = ...
markers = ...
python_files = ...
addopts = ...
:in setup.cfg
[tool:pytest]
DJANGO_SETTINGS_MODULE = ...
markers = ...
python_files = ...
addopts = ...
在很多教程中,你会看到有人创建pytest.ini
文件。这些文件只能用于 pytest 设置,我们希望使用更多设置,例如插件设置,并且随着项目的增长,不需要为每个插件添加大量文件。因此,强烈建议使用文件setup.cfg
。
在此文件中,对您来说重要的主要配置是:
DJANGO_SETTINGS_MODULE
:表示设置位于工作目录中的什么位置。python_files
:表示 pytest 将用于匹配测试文件的模式。addopts
:表示每次运行 pytest 时 pytest 应该使用的命令行参数。markers
:在这里我们定义了我们和我们的团队以后可能同意用于对测试进行分类的标记(即:“单元”、“集成”、“e2e”、“回归”等)。
测试类型
尽管在 QA 领域中,各种测试都有不同的名称(请参见此处),但只要我们了解以下 4 种类型的测试,就可以完成经过开发人员彻底测试的应用程序:
- 单元测试:对一段相关的代码进行测试,该代码与其他代码单元的交互隔离(主要是通过模拟外部代码来测试)。这些代码可以是内部代码,例如我们用来清理代码的辅助函数,也可以是数据库调用或外部 API 调用。
- 集成测试:测试一段代码而不将其与其他单元的交互隔离的测试。
- e2e 测试:e2e 代表“端到端”,此测试是集成测试,用于测试我们正在测试的 Django 应用程序的端到端流程。
- 回归测试:一种测试,无论是集成测试还是单元测试,都是由一个错误引起的,该错误在修复后立即被测试覆盖,以期在未来再次出现。
我们为什么要测试
在逐步推荐通过测试来确保应用万无一失的方法之前,最好先明确一下我们为什么要进行测试。主要有两个原因:
- 指导我们编码:在开发过程中,我们会有这些测试来指导我们的代码,我们构建的代码的目的是通过这些测试。
- 为了确保我们没有破坏任何东西:每一段代码都像是意大利面盘上的一根意大利面条。当你拿起一根面条,然后再放回去时,破坏任何东西的几率至少可以说很高。
正确的测试流程
对于测试开发人员来说,主要有两种代码测试流程:
- TDD(测试驱动开发):在通过代码断言之前进行测试。
- 开发后测试:创建代码后立即进行测试。
除非我们非常熟悉语法和我们预期要编写的代码(当我们还不确定要编写的代码时几乎不可能),否则选择后者几乎总是一个好主意。也就是说,编写代码,一旦我们认为它完成了,就继续测试它是否按预期工作。但我建议混合使用这两种方法:
- 进行端到端测试:这应该是我们最终测试套件的 TDD 组件。这些测试将首先帮助你预先规划每个端点应该返回的内容,并对应用程序其余代码的执行方式有一个初步的设想。
- 创建单元测试:编写代码并为 Django 应用的每个部分创建独立的测试。这不仅能帮助开发人员完成开发过程,还能帮助我们精准定位问题所在。
我们测试套件中唯一的集成测试应该是端到端测试。这类测试不应该成为我们团队在开发过程中的测试流程的一部分,而且,当面对大型代码库时,考虑到集成测试需要时间,它甚至不应该成为我们设置的持续集成/持续交付 (CI/CD) 流水线的一部分。端到端测试主要应该用作开发人员在代码的所有单元测试都通过后运行的健全性检查(请参阅 PyTest 文档中关于此策略的讨论),以确保不同代码单元之间的代码内聚性正常工作。
避免在流水线和开发工作流程中使用端到端测试,而改用单元测试的另一个原因是,这可以避免不稳定的测试。不稳定的测试在套件中运行时会失败,但在单独运行时却运行良好。请参阅如何修复不稳定的测试。
我们在工作流程中应该使用的测试实际上是单元测试,我们可以使用PyTest 标记或测试函数的名称来识别它们,以便稍后使用单个命令运行它们。此命令将在我们的开发工作流程中频繁使用。
我应该在单元测试中测试什么?
读者可能会想到的一个问题是,按照这种方法,我们无法测试模型。事实上,这正是我想要实现的。虽然本文并非介绍如何在 Dango 应用的不同部分管理业务逻辑,但模型应该仅用于数据表示。我们可能需要的模型逻辑应该存储在信号或模型管理器中。
由于我们避免在主要测试中进行数据库访问,因此我们将确保正确构建模型的工作留给 e2e 测试。
虽然在其他一些编程语言和框架中,这可能是一个棘手的问题,但在 Django 中,了解什么是可测试单元很简单,因为我们可以将逻辑分为 6 个部分。因此,单元测试可以是:
- 模型(模型方法/模型管理器)
- 信号
- 序列化器
- 辅助对象又称“实用程序”(函数、类、方法等)
- 视图/视图集
- URL 配置
常见测试实用程序
标记
在介绍测试示例之前,最好先介绍一下 PyTest 提供的一些实用标记。标记只是一些装饰器,它们用@pytest.mark.<marker>
我们设置的格式包装了测试函数。
@pytest.mark.parametrize()
:此标记将用于使用不同的值多次运行相同的测试,就像 for 循环一样。@pytest.mark.django_db
:如果我们不授予测试对数据库的访问权限,默认情况下它将无法访问数据库。pytest -django插件提供的这个标记当然只适用于集成测试。
嘲讽
在进行单元测试时,我们需要模拟对外部 API、数据库以及内部代码的访问。这时,以下库将会有所帮助:
pytest-mock
:提供unittest.mock
对象和非侵入式修补功能来代替unittest.mock
上下文管理器。requests-mock
:通过装置提供请求工厂rf
和模拟请求对象的能力。django-mock-queries
:提供了一个类,让我们可以模拟查询集对象并用非持久对象实例填充它。
PyTest 命令
我们在工作目录中运行所有测试的命令是pytest
,但 PyTest 提供的命令可以帮助我们缩小测试范围,并轻松选择要运行的测试。主要命令包括:
-k <expression>
:匹配测试文件夹内包含指示表达式的文件、类或函数名称。-m <marker>
:将使用输入的标记运行所有测试。-m "not <marker>"
:将运行所有没有输入标记的测试。-x
:一旦测试失败就会停止运行测试,让我们立即停止测试运行,以便我们可以返回调试测试,而不必等待测试套件完成运行。--lf
:从上次失败的测试开始运行测试套件,完美避免在调试时连续运行我们已经知道通过的测试。-vv
:显示失败断言的更详细版本。--cov
:显示测试覆盖的百分比(取决于pytest-cov
插件)。--reruns <num_of_reruns>
:用于处理不稳定的测试,即在测试套件中运行时失败但单独运行时通过的测试。
采用
Addopts 是 PyTest 命令,每次运行该命令时都会运行,pytest
因此我们不必每次都输入它。
我们可以按照以下方式配置我们的 addopts:
DJANGO_SETTINGS_MODULE = ...
markers = ...
python_files = ...
addopts = -vv -x --lf --cov
精确测试
为了快速运行所需的测试,我们可以运行pytest
命令-k
并输入所需测试的名称。
如果我们想运行一组具有共同点的测试,我们可以pytest
使用-k
命令运行来选择所有test_*.py
文件、所有Test*
类或所有test_*
带有插入表达式的函数。
pytest.ini
分组测试的一个好方法是在/文件中设置自定义标记,setup.cfg
并在团队中共享这些标记。我们可以为每种想要用作测试过滤器的模式设置一个标记。
制作什么样的标记与实际相关取决于开发人员的标准,但我建议仅对那些难以使用命令对匹配表达式进行分组的测试使用标记-k
。一个常见的相关标记示例是unit
仅选择单元测试的标记。
我们pytest.ini
将像这样设置标记:
[tool:pytest]
markers =
# Define our new marker
unit: tests that are isolated from the db, external api calls and other mockable internal code.
因此,如果我们想要标记遵循单元测试模式的测试,我们可以像这样标记:
import pytest
@pytest.mark.unit
def test_something(self):
pass
由于每个文件中都有多个单元测试(如果您听从我的建议,大多数测试文件实际上都是单元测试文件),我们可以通过在导入之后立即在文件顶部声明 pytestmark 变量来为所有文件设置全局标记,从而避免为每个函数创建标记的繁琐,该变量将包含单个 pytest 标记或标记列表:
# (imports)
# Only one global marker (most commonly used)
pytestmark = pytest.mark.unit
# Several global markers
pytestmark = [pytest.mark.unit, pytest.mark.other_criteria]
# (tests)
如果我们想更进一步,为所有测试设置一个全局标记,pytest 会创建一个名为 的 Fixture,items
它代表包含目录中的所有 PyTest 测试对象。因此,我们可以使用它来创建一个all
标记,并标记与其同级及以下的所有测试文件,conftest.py
如下所示:
# in conftest.py
def pytest_collection_modifyitems(items):
for item in items:
item.add_marker('all')
工厂
工厂是预先填充的模型实例。无需手动创建模型实例,工厂会为我们完成这项工作。创建工厂的主要模块是factory_boy
和model_bakery
。出于效率考虑,我们几乎应该始终使用model_bakery
,它只会根据给定的 Django 模型创建一个填充有效数据的模型实例。model_bakery 的问题在于它会创建随机的无意义数据,因此,如果我们想要一个工厂能够生成非随机字符且有意义的字段,那么我们应该继续使用 factory_boy 和 faker ,例如,如果我们有一个“name”字段,则生成一个看起来像名字而不是随机字母的名字。
对于创建工厂,我们希望在进行集成测试时创建并存储它们,但在进行单元测试时则不需要存储它们,因为我们不想访问数据库。保存到数据库中的实例被称为“持久化实例”,与“非持久化实例”相对,后者用于模拟数据库调用的响应。以下是两者的示例model_bakery
:
from model_bakery import baker
from apps.my_app.models import MyModel
# create and save to the database
baker.make(MyModel) # --> One instance
baker.make(MyModel, _quantity=3) # --> Batch of 3 instances
# create and don't save
baker.prepare(MyModel) # --> One instance
baker.prepare(MyModel, _quantity=3) # --> Batch of 3 instances
如果我们想要显示非随机数据(例如,在前端显示给产品所有者),我们可以覆盖 model_bakery 的默认行为(参见此处)。或者,我们可以像下面这样使用factory_boy
和来编写工厂函数:faker
# factories.py
import factory
class MyModelFactory(factory.DjangoModelFactory):
class Meta:
model = MyModel
field1 = factory.faker.Faker('relevant_generator')
...
# test_something.py
# Save to db
MyModelFactory() # --> One instance
MyModelFactory.create_batch(3) # --> Batch of 3 instances
# Do not save to db
MyModelFactory.build() # --> One instance
MyModelFactory.build_batch() # --> Batch of 3 instances
Faker 拥有生成器,可以为许多不同的主题生成随机相关数据,每个主题都由一个 Faker 提供程序表示。Factory Boy 的默认 Faker 版本开箱即用,包含所有提供程序,因此您只需前往此处,查看所有提供程序,然后选择一个您喜欢的提供程序,并将所需的生成器名称以字符串形式传递,即可生成字段(例如faker.Faker(‘name’)
:)。
工厂可以存储在 中conftest.py
,这是一个文件中,我们还可以在该文件中为同一目录级别及以下的所有测试设置配置(例如,这里是我们定义装置的地方),或者,如果我们的应用程序有许多不同的工厂,我们可以将它们直接存储在每个应用程序的factories.py
文件中:
....
├── tests
│ ├── __init__.py
│ ├── test_app1
│ │ ├── __init__.py
│ │ ├── conftest.py <--
│ │ ├── factories.py <--
│ │ ├── e2e_tests.py
│ │ ├── test_models.py
│ │ ├── test_signals.py
│ │ ├── test_serializers.py
│ │ ├── test_utils.py
│ │ ├── test_views.py
│ │ └── test_urls.py
│ │
│ └── ...
└── ...
如何组织测试
首先,如前所述,测试应该放在一个文件中,该文件具有我们pytest.ini
/setup.cfg
文件中指示的模式。
为了在我们的文件中对测试进行排序,对于每个测试代码单元(例如,Django APIView),创建一个带有 Test 的类,使用 Camel Case,并在其中创建该单元的测试(例如,对视图接受的所有方法的测试)。
对于根据测试功能组织测试,我建议使用“AMAA”标准,这是“AAA”标准的自定义版本。也就是说,测试应遵循以下顺序:
- 安排:设置测试所需的一切
- 模拟:模拟隔离测试所需的一切
- 动作:触发你的代码单元。
- 断言:断言结果与预期完全一致,以避免以后出现任何不愉快的意外。
因此,测试结构应如下所示:
# tests/test_app/app/test_some_part.py
...
# inside test_something.py
class TestUnitName:
def test_<functionality_1>(self):
# Arrange
# Mock
# Act
# Assert
...
测试示例
对于示例,我们将使用以下交易和货币模型来围绕它们构建示例:
# inside apps/app/models.py
import string
from django.db import models
from django.utils import timezone
from hashid_field import HashidAutoField
from apps.transaction.utils import create_payment_intent, PaymentStatuses
class Currency(models.Model):
"""Currency model"""
name = models.CharField(max_length=120, null=False, blank=False, unique=True)
code = models.CharField(max_length=3, null=False, blank=False, unique=True)
symbol = models.CharField(max_length=5, null=False, blank=False, default='$')
def __str__(self) -> str:
return self.code
class Transaction(models.Model):
"""Transaction model."""
id = HashidAutoField(primary_key=True, min_length=8, alphabet=string.printable.replace('/', ''))
name = models.CharField(max_length=50, null=False, blank=False)
email = models.EmailField(max_length=50, null=False, blank=False)
creation_date = models.DateTimeField(auto_now_add=True, null=False, blank=False)
currency = models.ForeignKey(Currency, null=False, blank=False, default=1, on_delete=models.PROTECT)
payment_status = models.CharField(choices=PaymentStatuses.choices, default=PaymentStatuses.WAI, max_length=21)
payment_intent_id = models.CharField(max_length=100, null=True, blank=False, default=None)
message = models.TextField(null=True, blank=True)
@property
def link(self):
"""
Link to a payment form for the transaction
"""
return settings.ALLOWED_HOSTS[0] + f'/payment/{str(self.id)}'
在内部,tests/test_app/conftest.py
我们将工厂设置为 Fixture,以便稍后在测试函数中将它们作为参数访问。我们并不总是需要一个由用户填写所有字段的模型实例,也许我们希望在后端自动填充它们。在这种情况下,我们可以创建自定义工厂,并填充某些字段:
def utbb():
def unfilled_transaction_bakery_batch(n):
utbb = baker.make(
'transaction.Transaction',
amount_in_cents=1032000, # --> Passes min. payload restriction in every currency
_fill_optional=[
'name',
'email',
'currency',
'message'
],
_quantity=n
)
return utbb
return unfilled_transaction_bakery_batch
@pytest.fixture
def ftbb():
def filled_transaction_bakery_batch(n):
utbb = baker.make(
'transaction.Transaction',
amount_in_cents=1032000, # --> Passes min. payload restriction in every currency
_quantity=n
)
return utbb
return filled_transaction_bakery_batch
@pytest.fixture
def ftb():
def filled_transaction_bakery():
utbb = baker.make(
'transaction.Transaction',
amount_in_cents=1032000, # --> Passes min. payload restriction in every currency
currency=baker.make('transaction.Currency')
)
return utbb
return filled_transaction_bakery
E2E 测试
一旦创建了模型及其工厂,我们最好对应用中希望实现的流程进行端到端测试,这既是指导,也是健全性检查。这不应该,也肯定不会在我们创建完成后就立即通过。这些测试是相当灵活的,可以根据程序员在开发过程中的某个时刻,认为每个端点应该为每个输入返回什么。
根据我们正在处理的功能的复杂性或产品所有者带来的更改,我们可能会在此过程中完全改变它们。
对于此端到端测试,为了提供多种情况的示例,我们假设我们希望使用所有六种 http 方法为两种模型提供完整的 CRUD 功能,如下所示:
GET api/transactions List all transaction objects
POST api/transactions Create a transaction object
GET api/transactions Retrieve a transaction object
PUT api/transactions/hash Update a transaction object
PATCH api/transactions/hash Update a field of a transaction object
DELETE api/transactions/hash Delete a transaction object
为此我们将:
- 为 DRF 的 APIClient 对象创建一个装置,并将其命名为
api_client
直接从端点开始测试:```python
conftest.py
@pytest.fixture
def api_client():
返回 APIClient
A good idea for this fixture, since it is not only relevant for this app, is to define it in a `conftest.py` file above all apps so it can be shared among all of them:
....
│
...
Now, let's proceed to test all endpoints:
```python
from model_bakery import baker
import factory
import json
import pytest
from apps.transaction.models import Transaction, Currency
pytestmark = pytest.mark.django_db
class TestCurrencyEndpoints:
endpoint = '/api/currencies/'
def test_list(self, api_client):
baker.make(Currency, _quantity=3)
response = api_client().get(
self.endpoint
)
assert response.status_code == 200
assert len(json.loads(response.content)) == 3
def test_create(self, api_client):
currency = baker.prepare(Currency)
expected_json = {
'name': currency.name,
'code': currency.code,
'symbol': currency.symbol
}
response = api_client().post(
self.endpoint,
data=expected_json,
format='json'
)
assert response.status_code == 201
assert json.loads(response.content) == expected_json
def test_retrieve(self, api_client):
currency = baker.make(Currency)
expected_json = {
'name': currency.name,
'code': currency.code,
'symbol': currency.symbol
}
url = f'{self.endpoint}{currency.id}/'
response = api_client().get(url)
assert response.status_code == 200
assert json.loads(response.content) == expected_json
def test_update(self, rf, api_client):
old_currency = baker.make(Currency)
new_currency = baker.prepare(Currency)
currency_dict = {
'code': new_currency.code,
'name': new_currency.name,
'symbol': new_currency.symbol
}
url = f'{self.endpoint}{old_currency.id}/'
response = api_client().put(
url,
currency_dict,
format='json'
)
assert response.status_code == 200
assert json.loads(response.content) == currency_dict
@pytest.mark.parametrize('field',[
('code'),
('name'),
('symbol'),
])
def test_partial_update(self, mocker, rf, field, api_client):
currency = baker.make(Currency)
currency_dict = {
'code': currency.code,
'name': currency.name,
'symbol': currency.symbol
}
valid_field = currency_dict[field]
url = f'{self.endpoint}{currency.id}/'
response = api_client().patch(
url,
{field: valid_field},
format='json'
)
assert response.status_code == 200
assert json.loads(response.content)[field] == valid_field
def test_delete(self, mocker, api_client):
currency = baker.make(Currency)
url = f'{self.endpoint}{currency.id}/'
response = api_client().delete(url)
assert response.status_code == 204
assert Currency.objects.all().count() == 0
class TestTransactionEndpoints:
endpoint = '/api/transactions/'
def test_list(self, api_client, utbb):
client = api_client()
utbb(3)
url = self.endpoint
response = client.get(url)
assert response.status_code == 200
assert len(json.loads(response.content)) == 3
def test_create(self, api_client, utbb):
client = api_client()
t = utbb(1)[0]
valid_data_dict = {
'amount_in_cents': t.amount_in_cents,
'currency': t.currency.code,
'name': t.name,
'email': t.email,
'message': t.message
}
url = self.endpoint
response = client.post(
url,
valid_data_dict,
format='json'
)
assert response.status_code == 201
assert json.loads(response.content) == valid_data_dict
assert Transaction.objects.last().link
def test_retrieve(self, api_client, ftb):
t = ftb()
t = Transaction.objects.last()
expected_json = t.__dict__
expected_json['link'] = t.link
expected_json['currency'] = t.currency.code
expected_json['creation_date'] = expected_json['creation_date'].strftime(
'%Y-%m-%dT%H:%M:%S.%fZ'
)
expected_json.pop('_state')
expected_json.pop('currency_id')
url = f'{self.endpoint}{t.id}/'
response = api_client().get(url)
assert response.status_code == 200 or response.status_code == 301
assert json.loads(response.content) == expected_json
def test_update(self, api_client, utbb):
old_transaction = utbb(1)[0]
t = utbb(1)[0]
expected_json = t.__dict__
expected_json['id'] = old_transaction.id.hashid
expected_json['currency'] = old_transaction.currency.code
expected_json['link'] = Transaction.objects.first().link
expected_json['creation_date'] = old_transaction.creation_date.strftime(
'%Y-%m-%dT%H:%M:%S.%fZ'
)
expected_json.pop('_state')
expected_json.pop('currency_id')
url = f'{self.endpoint}{old_transaction.id}/'
response = api_client().put(
url,
data=expected_json,
format='json'
)
assert response.status_code == 200 or response.status_code == 301
assert json.loads(response.content) == expected_json
@pytest.mark.parametrize('field',[
('name'),
('billing_name'),
('billing_email'),
('email'),
('amount_in_cents'),
('message'),
])
def test_partial_update(self, api_client, field, utbb):
utbb(2)
old_transaction = Transaction.objects.first()
new_transaction = Transaction.objects.last()
valid_field = {
field: new_transaction.__dict__[field],
}
url = f'{self.endpoint}{old_transaction.id}/'
response = api_client().patch(
path=url,
data=valid_field,
format='json',
)
assert response.status_code == 200 or response.status_code == 301
try:
assert json.loads(response.content)[field] == valid_field[field]
except json.decoder.JSONDecodeError as e:
pass
def test_delete(self, api_client, utbb):
transaction = utbb(1)[0]
url = f'{self.endpoint}{transaction.id}/'
response = api_client().delete(
url
)
assert response.status_code == 204 or response.status_code == 301
一旦我们对应用程序端点的初步预期输出进行了测试,我们就可以继续从模型构建应用程序的其余部分。
实用程序
Utils 是辅助函数,将分布在我们的整个代码中,因此您可以按照任何顺序构建它们及其相应的测试。
我们要制作的第一个实用程序是一个fill_transaction
函数,给定一个交易模型的实例,它将填充用户不打算填写的字段。
我们可以在后端填写的一个字段是该字段。“支付意向”是Stripepayment_intent_id
(一家支付服务)表示预期发生交易的方式;而它的 ID,简单来说,就是他们在数据库中查找相关数据的方式。
因此,使用 Stripe 的 python 库来创建和检索付款意向 ID 的实用程序可能是这样的:
def fill_transaction(transaction):
payment_intent_id = stripe.PaymentIntent.create(
amount=amount,
currency=currency.code.lower(),
payment_method_types=['card'],
).id
t = transaction.__class__.objects.filter(id=transaction.id)
t.update( # We use update not to trigger a save-signal recursion Overflow
payment_intent_id=payment_intent_id,
)
此实用程序的测试应模拟 API 调用和 2 个 db 调用:
class TestFillTransaction:
def test_function_code(self, mocker):
t = FilledTransactionFactory.build()
pi = PaymentIntentFactory()
create_pi_mock = mocker.Mock(return_value=pi)
stripe.PaymentIntent.create = create_pi_mock
filter_call_mock = mocker.Mock()
Transaction.objects.filter = filter_call_mock
update_call_mock = mocker.Mock()
filter_call_mock.return_value.update = update_call_mock
utils.fill_transaction(t)
filter_call_mock.assert_called_with(id=t.id)
update_call_mock.assert_called_with(
payment_intent_id=pi.id,
stripe_response=pi.last_response.data,
billing_email=t.email,
billing_name=t.name,
)
信号
fill_transaction
对于信号,我们可以在创建交易时发出信号来运行我们的实用程序。
from django.db.models.signals import pre_save
from django.dispatch import receiver
from apps.transaction.models import Transaction
from apps.transaction.utils import fill_transaction
@receiver(pre_save, sender=Transaction)
def transaction_filler(sender, instance, *args, **kwargs):
"""Fills fields"""
if not instance.id:
fill_transaction(instance)
顺便说一下,这个信号将在端到端进行隐式测试。一个好的显式单元测试可以如下:
import pytest
from django.db.models.signals import pre_save
from apps.transaction.models import Transaction
from tests.test_transaction.factories import UnfilledTransactionFactory, FilledTransactionFactory
pytestmark = pytest.mark.unit
class TestTransactionFiller:
def test_pre_save(self, mocker):
instance = UnfilledTransactionFactory.build()
mock = mocker.patch(
'apps.transaction.signals.fill_transaction'
)
pre_save.send(Transaction, instance=instance, created=True)
mock.assert_called_with(instance)
序列化器
对于我们的应用程序,我们将为我们的货币模型提供一个序列化器,为我们的交易模型提供两个序列化器:
- 包含可由交易管理员(创建和删除交易的人)修改的字段
- 其中包含“交易管理员”或付款人可以看到的字段。
from hashid_field.rest import HashidSerializerCharField
from rest_framework import serializers
from django.conf import settings
from django.core.validators import MaxLengthValidator, ProhibitNullCharactersValidator
from rest_framework.validators import ProhibitSurrogateCharactersValidator
from apps.transaction.models import Currency, Transaction
class CurrencySerializer(serializers.ModelSerializer):
class Meta:
model = Currency
fields = ['name', 'code', 'symbol']
if settings.DEBUG == True:
extra_kwargs = {
'name': {
'validators': [MaxLengthValidator, ProhibitNullCharactersValidator]
},
'code': {
'validators': [MaxLengthValidator, ProhibitNullCharactersValidator]
}
}
class UnfilledTransactionSerializer(serializers.ModelSerializer):
currency = serializers.SlugRelatedField(
slug_field='code',
queryset=Currency.objects.all(),
)
class Meta:
model = Transaction
fields = (
'name',
'currency',
'email',
'amount_in_cents',
'message'
)
class FilledTransactionSerializer(serializers.ModelSerializer):
id = HashidSerializerCharField(source_field='transaction.Transaction.id', read_only=True)
currency = serializers.StringRelatedField(read_only=True)
link = serializers.ReadOnlyField()
class Meta:
model = Transaction
fields = '__all__'
extra_kwargs = {
"""Non editable fields"""
'id': {'read_only': True},
'creation_date': {'read_only': True},
'payment_date': {'read_only': True},
'amount_in_cents': {'read_only': True},
'payment_intent_id': {'read_only': True},
'payment_status': {'read_only': True},
}
序列化器的单元测试应该旨在测试两件事(当相关时):
- 它可以正确序列化模型实例
- 它可以正确地将有效的序列化数据转换为模型(又称“反序列化”)
import pytest
import factory
from rest_framework.fields import CharField
from apps.transaction.api.serializers import CurrencySerializer, UnfilledTransactionSerializer, FilledTransactionSerializer
from tests.test_transaction.factories import CurrencyFactory, UnfilledTransactionFactory, FilledTransactionFactory
class TestCurrencySerializer:
transaction = UnfilledTransactionFactory.build()
@pytest.mark.unit
def test_serialize_model(self):
currency = CurrencyFactory.build()
serializer = CurrencySerializer(currency)
assert serializer.data
@pytest.mark.unit
def test_serialized_data(self, mocker):
valid_serialized_data = factory.build(
dict,
FACTORY_CLASS=CurrencyFactory
)
serializer = CurrencySerializer(data=valid_serialized_data)
assert serializer.is_valid()
assert serializer.errors == {}
class TestUnfilledTransactionSerializer:
@pytest.mark.unit
def test_serialize_model(self):
t = UnfilledTransactionFactory.build()
expected_serialized_data = {
'name': t.name,
'currency': t.currency.code,
'email': t.email,
'amount_in_cents': t.amount_in_cents,
'message': t.message,
}
serializer = UnfilledTransactionSerializer(t)
assert serializer.data == expected_serialized_data
https://docs.pytest.org/en/stable/flaky.html
@pytest.mark.django_db
def test_serialized_data(self, mocker):
c = CurrencyFactory()
t = UnfilledTransactionFactory.build(currency=c)
valid_serialized_data = {
'name': t.name,
'currency': t.currency.code,
'email': t.email,
'amount_in_cents': t.amount_in_cents,
'message': t.message,
}
serializer = UnfilledTransactionSerializer(data=valid_serialized_data)
assert serializer.is_valid(raise_exception=True)
assert serializer.errors == {}
ml_model_name_max_chars = 134
@pytest.mark.parametrize("wrong_field", (
{"name": "a" * (ml_model_name_max_chars + 1)},
{"tags": "tag outside of array"},
{"tags": ["--------wrong length tag--------"]},
{"version": "wronglengthversion"},
{"is_public": 1},
{"is_public": "Nope"},
))
def test_deserialize_fails(self, wrong_field: dict):
transaction_fields = [field.name for field in UnfilledTransaction._meta.get_fields()]
invalid_serialized_data = {
k: v for (k, v) in self.transaction.__dict__.items() if k in transaction_fields and k != "id"
} | wrong_field
serializer = MLModelSerializer(data=invalid_serialized_data)
assert not serializer.is_valid()
assert serializer.errors != {}
class TestFilledTransactionSerializer:
@pytest.mark.unit
def test_serialize_model(self, ftd):
t = FilledTransactionFactory.build()
expected_serialized_data = ftd(t)
serializer = FilledTransactionSerializer(t)
assert serializer.data == expected_serialized_data
@pytest.mark.unit
def test_serialized_data(self):
t = FilledTransactionFactory.build()
valid_serialized_data = {
'id': t.id.hashid,
'name': t.name,
'currency': t.currency.code,
'creation_date': t.creation_date.strftime('%Y-%m-%dT%H:%M:%SZ'),
'payment_date': t.payment_date.strftime('%Y-%m-%dT%H:%M:%SZ'),
'stripe_response': t.stripe_response,
'payment_intent_id': t.payment_intent_id,
'billing_name': t.billing_name,
'billing_email': t.billing_email,
'payment_status': t.payment_status,
'link': t.link,
'email': t.email,
'amount_in_cents': t.amount_in_cents,
'message': t.message,
}
serializer = FilledTransactionSerializer(data=valid_serialized_data)
assert serializer.is_valid(raise_exception=True)
assert serializer.errors == {}
视图集
我们将在模型中使用 DRF 视图集进行 CRUD 操作,从而无需测试 url 配置:
# in urls.py
route_lists = [
transaction_urls.route_list,
]
router = routers.DefaultRouter()
for route_list in route_lists:
for route in route_list:
router.register(route[0], route[1])
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', include(router.urls)),
]
# in views.py
from rest_framework.viewsets import ModelViewSet
from rest_framework.permissions import IsAuthenticated
from apps.transaction.api.serializers import CurrencySerializer, UnfilledTransactionSerializer, FilledTransactionSerializer
from apps.transaction.models import Currency, Transaction
class CurrencyViewSet(ModelViewSet):
queryset = Currency.objects.all()
serializer_class = CurrencySerializer
class TransactionViewset(ModelViewSet):
"""Transaction Viewset"""
queryset = Transaction.objects.all()
permission_classes = [IsAuthenticated]
def get_serializer_class(self):
if self.action == 'create':
return UnfilledTransactionSerializer
else:
return FilledTransactionSerializer
第一步是模拟我们要测试的视图中使用的所有权限。稍后会单独测试这些权限:
from rest_framework.permissions import IsAuthenticated
@pytest.fixture(scope="session", autouse=True)
def mock_views_permissions():
# little util I use for testing for DRY when patching multiple objects
patch_perm = lambda perm: mock.patch.multiple(
perm,
has_permission=mock.Mock(return_value=True),
has_object_permission=mock.Mock(return_value=True),
)
with (
patch_perm(IsAuthenticated),
# ...add other permissions you may have below
):
yield
使用简单的基于类的视图或基于函数的视图时,我们会从视图本身开始测试,并将视图与触发该视图的 urlconf 隔离开来。但由于我们为此使用了路由器,因此我们开始使用 API 客户端从端点本身测试视图集。
import factory
import json
import pytest
from django.urls import reverse
from django_mock_queries.mocks import MockSet
from rest_framework.relations import RelatedField, SlugRelatedField
from apps.transaction.api.serializers import UnfilledTransactionSerializer, CurrencySerializer
from apps.transaction.api.views import CurrencyViewSet, TransactionViewset
from apps.transaction.models import Currency, Transaction
from tests.test_transaction.factories import CurrencyFactory, FilledTransactionFactory, UnfilledTransactionFactory
pytestmark = [pytest.mark.urls('config.urls'), pytest.mark.unit]
class TestCurrencyViewset:
def test_list(self, mocker, rf):
# Arrange
url = reverse('currency-list')
request = rf.get(url)
qs = MockSet(
CurrencyFactory.build(),
CurrencyFactory.build(),
CurrencyFactory.build()
)
view = CurrencyViewSet.as_view(
{'get': 'list'}
)
#Mcking
mocker.patch.object(
CurrencyViewSet, 'get_queryset', return_value=qs
)
# Act
response = view(request).render()
#Assert
assert response.status_code == 200
assert len(json.loads(response.content)) == 3
def test_retrieve(self, mocker, rf):
currency = CurrencyFactory.build()
expected_json = {
'name': currency.name,
'code': currency.code,
'symbol': currency.symbol
}
url = reverse('currency-detail', kwargs={'pk': currency.id})
request = rf.get(url)
mocker.patch.object(
CurrencyViewSet, 'get_queryset', return_value=MockSet(currency)
)
view = CurrencyViewSet.as_view(
{'get': 'retrieve'}
)
response = view(request, pk=currency.id).render()
assert response.status_code == 200
assert json.loads(response.content) == expected_json
def test_create(self, mocker, rf):
valid_data_dict = factory.build(
dict,
FACTORY_CLASS=CurrencyFactory
)
url = reverse('currency-list')
request = rf.post(
url,
content_type='application/json',
data=json.dumps(valid_data_dict)
)
mocker.patch.object(
Currency, 'save'
)
view = CurrencyViewSet.as_view(
{'post': 'create'}
)
response = view(request).render()
assert response.status_code == 201
assert json.loads(response.content) == valid_data_dict
def test_update(self, mocker, rf):
old_currency = CurrencyFactory.build()
new_currency = CurrencyFactory.build()
currency_dict = {
'code': new_currency.code,
'name': new_currency.name,
'symbol': new_currency.symbol
}
url = reverse('currency-detail', kwargs={'pk': old_currency.id})
request = rf.put(
url,
content_type='application/json',
data=json.dumps(currency_dict)
)
mocker.patch.object(
CurrencyViewSet, 'get_object', return_value=old_currency
)
mocker.patch.object(
Currency, 'save'
)
view = CurrencyViewSet.as_view(
{'put': 'update'}
)
response = view(request, pk=old_currency.id).render()
assert response.status_code == 200
assert json.loads(response.content) == currency_dict
@pytest.mark.parametrize('field',[
('code'),
('name'),
('symbol'),
])
def test_partial_update(self, mocker, rf, field):
currency = CurrencyFactory.build()
currency_dict = {
'code': currency.code,
'name': currency.name,
'symbol': currency.symbol
}
valid_field = currency_dict[field]
url = reverse('currency-detail', kwargs={'pk': currency.id})
request = rf.patch(
url,
content_type='application/json',
data=json.dumps({field: valid_field})
)
mocker.patch.object(
CurrencyViewSet, 'get_object', return_value=currency
)
mocker.patch.object(
Currency, 'save'
)
view = CurrencyViewSet.as_view(
{'patch': 'partial_update'}
)
response = view(request).render()
assert response.status_code == 200
assert json.loads(response.content)[field] == valid_field
def test_delete(self, mocker, rf):
currency = CurrencyFactory.build()
url = reverse('currency-detail', kwargs={'pk': currency.id})
request = rf.delete(url)
mocker.patch.object(
CurrencyViewSet, 'get_object', return_value=currency
)
del_mock = mocker.patch.object(
Currency, 'delete'
)
view = CurrencyViewSet.as_view(
{'delete': 'destroy'}
)
response = view(request).render()
assert response.status_code == 204
assert del_mock.assert_called
class TestTransactionViewset:
def test_list(self, mocker, rf):
url = reverse('transaction-list')
request = rf.get(url)
qs = MockSet(
FilledTransactionFactory.build(),
FilledTransactionFactory.build(),
FilledTransactionFactory.build()
)
mocker.patch.object(
TransactionViewset, 'get_queryset', return_value=qs
)
view = TransactionViewset.as_view(
{'get': 'list'}
)
response = view(request).render()
assert response.status_code == 200
assert len(json.loads(response.content)) == 3
def test_create(self, mocker, rf):
valid_data_dict = factory.build(
dict,
FACTORY_CLASS=UnfilledTransactionFactory
)
currency = valid_data_dict['currency']
valid_data_dict['currency'] = currency.code
url = reverse('transaction-list')
request = rf.post(
url,
content_type='application/json',
data=json.dumps(valid_data_dict)
)
retrieve_currency = mocker.Mock(return_value=currency)
SlugRelatedField.to_internal_value = retrieve_currency
mocker.patch.object(
Transaction, 'save'
)
view = TransactionViewset.as_view(
{'post': 'create'}
)
response = view(request).render()
assert response.status_code == 201
assert json.loads(response.content) == valid_data_dict
def test_retrieve(self, api_client, mocker, ftd):
transaction = FilledTransactionFactory.build()
expected_json = ftd(transaction)
url = reverse(
'transaction-detail', kwargs={'pk': transaction.id}
)
TransactionViewset.get_queryset = mocker.Mock(
return_value=MockSet(transaction)
)
response = api_client().get(url)
assert response.status_code == 200
assert json.loads(response.content) == expected_json
def test_update(self, mocker, api_client, ftd):
old_transaction = FilledTransactionFactory.build()
new_transaction = FilledTransactionFactory.build()
transaction_json = ftd(new_transaction, old_transaction)
url = reverse(
'transaction-detail',
kwargs={'pk': old_transaction.id}
)
retrieve_currency = mocker.Mock(
return_value=old_transaction.currency
)
SlugRelatedField.to_internal_value = retrieve_currency
mocker.patch.object(
TransactionViewset,
'get_object',
return_value=old_transaction
)
Transaction.save = mocker.Mock()
response = api_client().put(
url,
data=transaction_json,
format='json'
)
assert response.status_code == 200
assert json.loads(response.content) == transaction_json
@pytest.mark.parametrize('field',[
('name'),
('billing_name'),
('billing_email'),
('email'),
('amount_in_cents'),
('message'),
])
def test_partial_update(self, mocker, api_client, field):
old_transaction = FilledTransactionFactory.build()
new_transaction = FilledTransactionFactory.build()
valid_field = {
field: new_transaction.__dict__[field]
}
url = reverse(
'transaction-detail',
kwargs={'pk': old_transaction.id}
)
SlugRelatedField.to_internal_value = mocker.Mock(
return_value=old_transaction.currency
)
mocker.patch.object(
TransactionViewset,
'get_object',
return_value=old_transaction
)
Transaction.save = mocker.Mock()
response = api_client().patch(
url,
data=valid_field,
format='json'
)
assert response.status_code == 200
assert json.loads(response.content)[field] == valid_field[field]
def test_delete(self, mocker, api_client):
transaction = FilledTransactionFactory.build()
url = reverse('transaction-detail', kwargs={'pk': transaction.id})
mocker.patch.object(
TransactionViewset, 'get_object', return_value=transaction
)
del_mock = mocker.patch.object(
Transaction, 'delete'
)
response = api_client().delete(
url
)
assert response.status_code == 204
assert del_mock.assert_called
我们的测试套件的覆盖范围
考虑到我们的代码将具有不同的逻辑分支,我们可以使用该插件以百分比形式测试测试覆盖的代码量(即测试套件的“覆盖率”)pytest-cov
。为了查看测试的覆盖率,我们需要使用 --cov 命令。
覆盖率不会告诉您您的代码是否会中断,它只会告诉您测试覆盖了多少内容,而这与您所做的测试是否相关无关。
最好在setup.cfg
under 目录下手动修改覆盖率设置[coverage:run]
(或者,如果你想要一个独立的文件,也可以在.coveragerc
[run] 目录下的一个文件中修改),并设置要测试覆盖率的目录以及要排除的文件。我自己把所有应用都放在一个apps
目录中,所以我的setup.cfg
设置看起来像这样:
[tool:pytest]
...
[coverage:run]
source=apps
omit=*/migrations/*,
如果我们运行pytest --cov --cov-config=setup.cfg
(你可以把它包含在你的 addopts 中),我们可能会得到这样的输出: 我得到了 100% 的覆盖率并不意味着什么。同样,如果你因为遗漏了一段不相关的代码而得到了 90% 的覆盖率,你的覆盖率也和 100% 一样好。
免责声明:pytest-cov
据报道,插件与 VSCode 调试器不兼容,因此您可能需要从 addopts 中删除它的命令,有时甚至将其从项目中完全删除。
关于不稳定的测试
不稳定的测试是由于测试之间没有得到适当的隔离而导致的。在运行端到端测试时,这种情况是可以接受的,但在单元测试中发生时,应该会发出红色警告。
进行中
- 普通视图测试
- URL 配置测试
- 资源列表
- 导入清理