代码库
单元测试
Xunit风格
import pytest
def setup_module(module):
print("每个模块之前执行一次")
def teardown_module(module):
print("每个模块之后执行一次")
class TestCase:
@classmethod
def setup_class(cls):
print("父类的setup_class")
class TestMaShang(TestCase):
workage = 8
def setup_class(self):
"""也支持方法的形式实现setupClass"""
super().setup_class()
print("每个类之前执行一次")
def teardown_class(self):
print("每个类之后执行一次")
def setup_method(self):
print("每个用例之前执行一次")
def teardown_method(self):
print("每个用例之后执行一次")
@pytest.mark.smoke # 标记
def test_baili(self):
print("测试百里老师")
def test_xingyao(self):
print("测试星瑶老师")
@pytest.mark.smoke
def test_yiran(self):
print("测试依然老师")
@pytest.mark.skipif(reason="无理由跳过")
def test_skip_if(self):
"""跳过测试
使用场景: 反例测试用例
"""
@pytest.mark.skipif(workage < 10, reason="有条件跳过")
def test_skip_if_condition(self):
pass
if __name__ == '__main__':
pytest.main()
# 钩子函数
# ==================================
# 只执行一次
# https://docs.pytest.org/en/7.4.x/reference/reference.html#pytest.hookspec.pytest_runtestloop
def pytest_runtestloop(session):
print("初始化......")
package yours
import (
"fmt"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
"testing"
)
// TestMain https://pkg.go.dev/testing#hdr-Main
// 每个目录下只允许定义一个TestMain, 类似main包的main()函数
func TestMain(m *testing.M) {
// 开始测试前的初始化动作
fmt.Println("启动mysql、redis容器")
m.Run()
// 结束测试后的清理动作
fmt.Println("关闭mysql、redis容器")
}
// Define the suite, and absorb the built-in basic suite
// functionality from testify - including a T() method which
// returns the current testing context
type MaShangeTestSuite struct {
suite.Suite
WorkAge int
}
// SetupSuite
func (sute *MaShangeTestSuite) SetupSuite() {
fmt.Println("每个测试集之前执行一次")
}
// TearDownSuite
func (suite *MaShangeTestSuite) TearDownSuite() {
fmt.Println("每个测试集之后执行一次")
}
// Make sure that WorkAge is set to five
// before each test
func (suite *MaShangeTestSuite) SetupTest() {
suite.WorkAge = 8
}
// TearDownTest
func (suite *MaShangeTestSuite) TearDownTest() {
fmt.Println("每条用例之后执行一次")
}
// All methods that begin with "Test" are run as tests within a
// suite.
func (suite *MaShangeTestSuite) TestBaiLi() {
assert.Equal(suite.T(), 8, suite.WorkAge)
// 或者
suite.Equal(suite.WorkAge, 8)
}
// In order for 'go test' to run this suite, we need to create
// a normal test function and pass our suite to suite.Run
func TestEMaShangeTestSuite(t *testing.T) {
suite.Run(t, new(MaShangeTestSuite))
}
func TestSomething(t *testing.T) {
// assert equality
assert.Equal(t, 123, 123, "they should be equal")
// assert inequality
assert.NotEqual(t, 123, 456, "they should not be equal")
// assert for nil (good for errors)
assert.Nil(t, object)
// assert for not nil (good when you expect something)
if assert.NotNil(t, object) {
// now we know that object isn't nil, we are safe to make
// further assertions without causing any errors
assert.Equal(t, "Something", object.Value)
}
}
单脚本场景下的单元测试
场景: 一个main脚本放在服务器用定时任务去跑
def main():
pass
if __name__ == '__main__':
main()
创建一个main_test.py编写测试用例
接口测试
"""https://www.django-rest-framework.org/api-guide/testing/#api-test-cases"""
from django.urls import reverse
from rest_framework import status
from rest_framework.test import APITestCase
from django.db import models
class Account(models.Model):
pass
class AccountTest(APITestCase):
def test_create_account(self):
"""确保我们可以创建一个新的账号"""
url = reverse('account-list')
data = {"name": "DabApps"}
response = self.client.post(url, data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
self.assertEqual(Account.objects.count(), 1)
self.assertEqual(AccountTest.objects.get().name, 'DabApps')
基于容器的外部依赖
https://gitee.com/luzhenxiong/bilibili-tests/tree/master/testcontainers_
import os
import pytest
import sqlalchemy
# mysql
# =======================================
@pytest.fixture()
def c():
from testcontainers.mysql import MySqlContainer
# 3.7.1的testcontainers需要设置环境变量
os.environ.setdefault('TC_HOST', 'localhost')
with MySqlContainer('mysql:8.0.35') as mysql:
engine = sqlalchemy.create_engine(mysql.get_connection_url())
print(mysql.get_connection_url())
with engine.connect() as conn:
yield conn
def test_case(c):
result = c.execute(sqlalchemy.text("select version()"))
version, = result.fetchone()
print(version)
# redis
# =======================================
from testcontainers.redis import RedisContainer
@pytest.fixture(scope='session')
def redis_container():
# 3.7.1的testcontainers需要设置环境变量
# os.environ.setdefault('TC_HOST', 'localhost')
with RedisContainer('redis:7.2.3') as redis_container:
yield redis_container
@pytest.fixture
def redis_client(redis_container):
yield redis_container.get_client()
def test_redis(redis_container):
print(redis_container.get_container_host_ip())
# 端口号
# 3.7.1
# print(redis_container.get_exposed_port(redis_container.port_to_expose))
package customer
import (
"context"
"log"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
"github.com/testcontainers/testcontainers-go-demo/testhelpers"
)
type CustomerRepoTestSuite struct {
suite.Suite
pgContainer *testhelpers.PostgresContainer
repository *Repository
ctx context.Context
}
func (suite *CustomerRepoTestSuite) SetupSuite() {
suite.ctx = context.Background()
pgContainer, err := testhelpers.CreatePostgresContainer(suite.ctx)
if err != nil {
log.Fatal(err)
}
suite.pgContainer = pgContainer
repository, err := NewRepository(suite.ctx, suite.pgContainer.ConnectionString)
if err != nil {
log.Fatal(err)
}
suite.repository = repository
}
func (suite *CustomerRepoTestSuite) TearDownSuite() {
if err := suite.pgContainer.Terminate(suite.ctx); err != nil {
log.Fatalf("error terminating postgres container: %s", err)
}
}
func (suite *CustomerRepoTestSuite) TestCreateCustomer() {
t := suite.T()
customer, err := suite.repository.CreateCustomer(suite.ctx, Customer{
Name: "Henry",
Email: "henry@gmail.com",
})
assert.NoError(t, err)
assert.NotNil(t, customer.Id)
}
func (suite *CustomerRepoTestSuite) TestGetCustomerByEmail() {
t := suite.T()
customer, err := suite.repository.GetCustomerByEmail(suite.ctx, "john@gmail.com")
assert.NoError(t, err)
assert.NotNil(t, customer)
assert.Equal(t, "John", customer.Name)
assert.Equal(t, "john@gmail.com", customer.Email)
}
func TestCustomerRepoTestSuite(t *testing.T) {
suite.Run(t, new(CustomerRepoTestSuite))
}
package testhelpers
import (
"context"
"path/filepath"
"time"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/postgres"
"github.com/testcontainers/testcontainers-go/wait"
)
type PostgresContainer struct {
*postgres.PostgresContainer
ConnectionString string
}
func CreatePostgresContainer(ctx context.Context) (*PostgresContainer, error) {
pgContainer, err := postgres.RunContainer(ctx,
testcontainers.WithImage("postgres:15.3-alpine"),
postgres.WithInitScripts(filepath.Join("..", "testdata", "init-db.sql")),
postgres.WithDatabase("test-db"),
postgres.WithUsername("postgres"),
postgres.WithPassword("postgres"),
testcontainers.WithWaitStrategy(
wait.ForLog("database system is ready to accept connections").
WithOccurrence(2).WithStartupTimeout(5*time.Second)),
)
if err != nil {
return nil, err
}
connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable")
if err != nil {
return nil, err
}
return &PostgresContainer{
PostgresContainer: pgContainer,
ConnectionString: connStr,
}, nil
}
mock技术
import datetime
import os
from unittest.mock import MagicMock
from pytest_mock import MockFixture, MockType
class UnixFS:
@staticmethod
def rm(filename):
os.remove(filename)
def test_unix_fs(mocker):
mocker.patch('os.remove')
UnixFS.rm('file')
os.remove.assert_called_once_with('file')
# Stub: https://pytest-mock.readthedocs.io/en/latest/usage.html#stub
# 等价于调用MagicMock
def test_stub(mocker: MockFixture):
def foo(on_something):
on_something('foo', 'bar')
# 【pk】MockType: 关联issue[#414](https://github.com/pytest-dev/pytest-mock/issues/414)
stub: MockType = mocker.stub(name='on_something_stub')
foo(on_something=stub)
stub.assert_called_once_with('foo', 'bar')
# 关联issue: [mocker.stub() is a coroutine function](https://github.com/pytest-dev/pytest-mock/issues/375)
import inspect
assert inspect.iscoroutinefunction(stub)
assert not inspect.iscoroutinefunction(MagicMock(name='on_something_stub'))
# mock类的方法的同时让原方法调用生效
def test_real_call(mocker: MockFixture):
class Client:
def create(self):
print('create')
c = Client()
def call(client: Client):
client.create()
stub = mocker.stub(name='create')
stub.side_effect = c.create
# new参数是关键
mocker.patch.object(Client, 'create', new=stub)
call(c) # create方法仍然生效
stub.assert_called_once()
# mock时间类型
# https://github.com/spulec/freezegun/blob/master/README.rst
@freeze_time("2012-01-14")
def test():
assert datetime.datetime.now() == datetime.datetime(2012, 1, 14)
https://github.com/agiledragon/gomonkey
package logic
import (
"errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/suite"
"go-zero-api/app/internal/svc"
"go-zero-api/app/internal/types"
"go-zero-api/app/testhelpers"
"testing"
. "github.com/agiledragon/gomonkey/v2"
)
type SendMsgSuite struct {
testhelpers.DbSuite
l *SendMsgLogic
}
func (s *SendMsgSuite) SetupSuite() {
s.Init()
s.l = NewSendMsgLogic(s.Ctx, s.NewServiceContext())
}
// TestSendMsg go test添加额外参数-gcflags=all=-l, 禁止内联优化
// Goland的运行配置在go tool添加参数
func (s *SendMsgSuite) TestSendMsg() {
ast := assert.New(s.T())
var called bool
patches := ApplyFunc(SendFeiShu, func(svcCtx *svc.ServiceContext, title string, msgBody string, receivers []string) error {
if title != "测试" {
return errors.New("传入参数不正确")
}
called = true
return nil
})
defer patches.Reset()
resp, err := s.l.SendMsg(&types.SendMsgReq{
Title: "测试",
Content: "文本内容",
Receivers: []string{"397132445@qq.om"},
})
ast.Nil(err)
ast.Equal(resp.Code, 0, resp.Msg)
// 验证是否有调用函数
ast.True(called, "SendFeiShu should have been called")
}
func TestSendMsgSuite(t *testing.T) {
suite.Run(t, new(SendMsgSuite))
}
from unittest.mock import MagicMock
thing = ProductionClass()
thing.method = MagicMock(return_value=3)
thing.method(3, 4, 5, key='value')
thing.method.assert_called_with(3, 4, 5, key='value')
# 通过 side_effect 设置副作用(side effects) ,可以是一个 mock 被调用是抛出的异常
mock = Mock(side_effect=KeyError('foo'))
mock()
values = {'a': 1, 'b': 2, 'c': 3}
def side_effect(arg):
return values[arg]
mock.side_effect = side_effect
mock('a'), mock('b'), mock('c')
mock.side_effect = [5, 4, 3, 2, 1]
mock(), mock(), mock()
保证测试中的模拟对象与要替换的对象具有相同的api
from unittest.mock import create_autospec
def function(a, b, c):
pass
mock_function = create_autospec(function, return_value='fishy')
mock_function(1, 2, 3)
mock_function.assert_called_once_with(1, 2, 3)
mock_function('wrong arguments')
命令行执行测试
pytest -v
go test ./... -v
测试异步函数
@pytest.mark.asyncio
async def test_some_asyncio_code():
res = await library.do_something()
assert b'expected result' == res
测试覆盖
# 运行
$ coverage run --source=app -m pytest
# 命令行输出报告
$ coverage report --show-missing
# 测试结果保存到html
$ coverage html --title "${@-coverage}"
参见
fastapi模板项目的 test.sh