feat: 添加任务历史记录、持久化会话、WebSocket支持和数据持久化功能

This commit is contained in:
2026-03-10 18:58:03 +08:00
parent f56ba5559d
commit 7fdd31b07b
22 changed files with 2006 additions and 4 deletions

763
docs/pytest-guide.md Normal file
View File

@@ -0,0 +1,763 @@
# Pytest 完整使用指南
Pytest 是 Python 最流行的测试框架,以其简洁的 API 和强大的功能被广泛应用于各种项目中。
## 目录
1. [快速开始](#1-快速开始)
2. [编写测试](#2-编写测试)
3. [Fixtures](#3-fixtures)
4. [异步测试](#4-异步测试)
5. [Mocking](#5-mocking)
6. [参数化测试](#6-参数化测试)
7. [跳过测试](#7-跳过测试)
8. [测试夹具作用域](#8-测试夹具作用域)
9. [测试组织](#9-测试组织)
10. [常用命令](#10-常用命令)
11. [配置文件](#11-配置文件)
---
## 1. 快速开始
### 安装
```bash
pip install pytest pytest-asyncio
```
### 运行测试
```bash
# 运行所有测试
pytest
# 运行指定文件
pytest tests/test_api.py
# 运行指定测试函数
pytest tests/test_api.py::test_health_check
# 显示详细输出
pytest -v
# 显示打印内容
pytest -s
# 在第一个失败处停止
pytest -x
```
---
## 2. 编写测试
### 基本结构
```python
# test_example.py
def test_basic_assertion():
"""简单的断言测试"""
assert 1 + 1 == 2
def test_string_operations():
"""字符串操作测试"""
text = "hello world"
assert text.upper() == "HELLO WORLD"
assert text.startswith("hello")
assert len(text) == 11
def test_list_operations():
"""列表操作测试"""
items = [1, 2, 3]
assert len(items) == 3
assert items.pop() == 3
assert items == [1, 2]
```
### 测试类
```python
class TestMathOperations:
def test_addition(self):
assert 2 + 3 == 5
def test_subtraction(self):
assert 5 - 3 == 2
def test_multiplication(self):
assert 3 * 4 == 12
class TestStringOperations:
def test_reverse(self):
assert "hello"[::-1] == "olleh"
```
---
## 3. Fixtures
Fixtures 是 pytest 最强大的特性之一,用于提供测试所需的依赖和数据。
### 基本 Fixture
```python
import pytest
@pytest.fixture
def sample_data():
"""返回一个简单的数据字典"""
return {"name": "test", "value": 100}
def test_using_fixture(sample_data):
"""使用 fixture 的测试"""
assert sample_data["value"] > 50
```
### Fixture with Setup/Teardown
```python
import pytest
import tempfile
import shutil
@pytest.fixture
def temp_directory():
"""创建临时目录,测试后清理"""
temp_path = tempfile.mkdtemp()
yield temp_path # 测试在此运行
shutil.rmtree(temp_path, ignore_errors=True) # 清理
def test_write_file(temp_directory):
"""使用临时目录的测试"""
import os
file_path = os.path.join(temp_directory, "test.txt")
with open(file_path, "w") as f:
f.write("hello")
assert os.path.exists(file_path)
# 测试结束后 temp_path 会被自动清理
```
### Fixture 参数化
```python
@pytest.fixture
def user_data():
return {"username": "testuser", "email": "test@example.com"}
def test_user_creation(user_data):
assert user_data["username"] == "testuser"
```
---
## 4. 异步测试
### 安装异步支持
```bash
pip install pytest-asyncio
```
### 配置 pytest.ini
```ini
[pytest]
asyncio_mode = auto
```
### 编写异步测试
```python
import pytest
@pytest.mark.asyncio
async def test_async_operation():
"""异步测试函数"""
import asyncio
async def fetch_data():
await asyncio.sleep(0.1)
return {"data": "result"}
result = await fetch_data()
assert result["data"] == "result"
@pytest.mark.asyncio
async def test_async_with_db():
"""模拟异步数据库操作"""
import asyncio
async def connect_db():
await asyncio.sleep(0.1)
return "database_connected"
result = await connect_db()
assert "connected" in result
```
### Fixture 返回协程
```python
import pytest
import asyncio
@pytest.fixture
async def async_database():
"""异步 fixture - 连接数据库"""
await asyncio.sleep(0.1) # 模拟连接
db = {"connected": True}
yield db
# Teardown
await asyncio.sleep(0.05)
@pytest.mark.asyncio
async def test_database_query(async_database):
assert async_database["connected"] is True
```
---
## 5. Mocking
### 使用 unittest.mock
```python
from unittest.mock import Mock, patch, AsyncMock
import pytest
def test_mock_function():
"""Mock 函数调用"""
mock = Mock(return_value=42)
result = mock()
assert result == 42
mock.assert_called_once()
def test_patch_environment():
"""Patch 环境变量"""
with patch("os.environ", {"TEST": "value"}):
import os
assert os.environ.get("TEST") == "value"
def test_mock_class():
"""Mock 类"""
mock_class = Mock()
mock_class.method.return_value = "mocked"
result = mock_class.method()
assert result == "mocked"
@pytest.mark.asyncio
async def test_async_mock():
"""Mock 异步函数"""
mock_async = AsyncMock(return_value="async result")
result = await mock_async()
assert result == "async result"
@pytest.mark.asyncio
async def test_patch_async():
"""Patch 异步函数"""
async def real_async_func():
return "real"
with patch("__main__.real_async_func", new_callable=AsyncMock) as mock_func:
mock_func.return_value = "mocked"
result = await mock_func()
assert result == "mocked"
```
### Mocking 第三方库
```python
from unittest.mock import patch, MagicMock
import pytest
# Mock HTTP 请求
@pytest.mark.asyncio
async def test_api_call():
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {"status": "ok"}
with patch("httpx.AsyncClient") as mock_client:
mock_instance = AsyncMock()
mock_instance.get = AsyncMock(return_value=mock_response)
mock_client.return_value = mock_instance
# 实际测试代码会使用 httpx.AsyncClient
client = mock_client.get("http://api.example.com")
# ...
```
---
## 6. 参数化测试
一个测试函数使用不同参数多次运行。
### @pytest.mark.parametrize
```python
import pytest
@pytest.mark.parametrize("input,expected", [
(1, 2),
(2, 4),
(3, 6),
])
def test_double(input, expected):
"""测试倍数函数"""
assert input * 2 == expected
@pytest.mark.parametrize("a,b,result", [
(1, 1, 2),
(2, 3, 5),
(10, 20, 30),
])
def test_addition(a, b, result):
assert a + b == result
# 字符串参数
@pytest.mark.parametrize("text,upper", [
("hello", "HELLO"),
("World", "WORLD"),
("pytest", "PYTEST"),
])
def test_uppercase(text, upper):
assert text.upper() == upper
```
### 多参数组合
```python
import pytest
@pytest.mark.parametrize("x", [1, 2, 3])
@pytest.mark.parametrize("y", [10, 20])
def test_combinations(x, y):
"""x=1,2,3 与 y=10,20 的所有组合"""
assert x * y > 0
# 运行 6 次: (1,10), (1,20), (2,10), (2,20), (3,10), (3,20)
```
### 使用 pytest 对象的 param
```python
import pytest
@pytest.fixture
def pytest_param_user():
return pytest.param("user1", marks=pytest.mark.slow)
def test_user_stuff():
pass
```
---
## 7. 跳过测试
### 跳过条件
```python
import pytest
import sys
@pytest.mark.skip(reason="功能未实现")
def test_not_implemented():
pass
@pytest.mark.skipif(sys.version_info < (3, 8), reason="需要 Python 3.8+")
def test_new_feature():
pass
# 动态跳过
def test_conditionally():
if some_condition:
pytest.skip("条件不满足")
assert True
```
### 预期失败
```python
@pytest.mark.xfail(reason="已知 bug")
def test_known_bug():
assert False # 预期失败
```
---
## 8. 测试夹具作用域
Fixture 可以指定不同的作用域:
```python
# function: 每个测试函数执行一次(默认)
@pytest.fixture
def function_fixture():
print("function scope")
return "function"
# class: 每个测试类执行一次
@pytest.fixture(scope="class")
def class_fixture():
print("class scope")
return "class"
# module: 每个模块执行一次
@pytest.fixture(scope="module")
def module_fixture():
print("module scope")
return "module"
# session: 整个测试会话执行一次
@pytest.fixture(scope="session")
def session_fixture():
print("session scope")
return "session"
```
---
## 9. 测试组织
### 目录结构建议
```
project/
├── tests/
│ ├── __init__.py
│ ├── conftest.py # 共享 fixtures
│ ├── test_api.py # API 测试
│ ├── test_models.py # 模型测试
│ ├── test_services/ # 服务测试
│ │ ├── __init__.py
│ │ ├── test_session.py
│ │ └── test_scheduler.py
│ └── test_integration/ # 集成测试
│ └── test_workflow.py
├── src/
│ └── your_code.py
└── pytest.ini
```
### conftest.py 共享配置
```python
# tests/conftest.py
import pytest
@pytest.fixture
def common_setup():
"""所有测试共享的设置"""
print("Setting up test environment")
yield
print("Tearing down test environment")
# 自动使用某些 fixtures
@pytest.fixture(autouse=True)
def reset_state():
"""每个测试自动重置状态"""
yield
```
---
## 10. 常用命令
```bash
# 基本运行
pytest # 运行所有测试
pytest tests/ # 运行指定目录
pytest test_file.py # 运行指定文件
pytest -k "test_name" # 按名称过滤
pytest -k "test_api" # 运行包含 "test_api" 的测试
# 输出控制
pytest -v # 详细输出
pytest -s # 显示 print 输出
pytest --tb=short # 简短 traceback
pytest --tb=long # 详细 traceback
# 调试
pytest -x # 第一个失败后停止
pytest --pdb # 失败时进入 debugger
pytest -l # 显示失败的局部变量
# 覆盖率
pytest --cov=src # 运行并计算覆盖率
pytest --cov-report=html # 生成 HTML 报告
# 标记
pytest -m "slow" # 运行带 slow 标记的测试
pytest -m "not slow" # 排除 slow 标记
# 异步
pytest --asyncio-mode=auto # 自动检测异步测试
```
---
## 11. 配置文件
### pytest.ini
```ini
[pytest]
minversion = 6.0
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
# 异步配置
asyncio_mode = auto
# 标记定义
markers =
slow: 耗时较长的测试
integration: 集成测试
unit: 单元测试
# 忽略路径
norecursedirs = .git .tox build dist
# 自定义选项
addopts = -v --tb=short
```
### pyproject.toml
```toml
[tool.pytest.ini_options]
minversion = "6.0"
testpaths = ["tests"]
pythonpath = ["."]
[tool.pytest.markers]
slow = "marks tests as slow"
integration = "marks tests as integration tests"
[tool.coverage.run]
source = ["src"]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"raise AssertionError",
]
```
---
## 高级技巧
### 1. 自定义标记
```python
# 在代码中使用
@pytest.mark.slow
def test_slow_operation():
import time
time.sleep(5)
assert True
@pytest.mark.integration
def test_api_workflow():
pass
# 在 pytest.ini 中注册
[pytest]
markers =
slow: marks tests as slow (deselect with '-m "not slow"')
integration: marks tests as integration tests
```
### 2. 条件跳过
```python
import pytest
# 基于环境变量
import os
@pytest.mark.skipif(
os.getenv("CI") == "true",
reason="跳过 CI 环境"
)
def test_dev_only():
pass
# 基于平台
@pytest.mark.skipif(
sys.platform == "win32",
reason="不适用于 Windows"
)
def test_linux_only():
pass
```
### 3. Fixture 依赖
```python
@pytest.fixture
def database():
return {"connected": False}
@pytest.fixture
def db_session(database):
"""依赖另一个 fixture"""
session = {"db": database, "active": True}
return session
def test_database_connection(db_session):
assert db_session["active"] is True
```
### 4. 工厂 Fixture
```python
@pytest.fixture
def user_factory():
"""工厂 fixture - 每次返回新实例"""
created = []
def _create_user(name):
user = {"id": len(created) + 1, "name": name}
created.append(user)
return user
return _create_user
def test_create_users(user_factory):
user1 = user_factory("Alice")
user2 = user_factory("Bob")
assert user1["id"] == 1
assert user2["id"] == 2
```
### 5. 临时文件
```python
import pytest
from pathlib import Path
import tempfile
import shutil
@pytest.fixture
def temp_dir():
temp_path = Path(tempfile.mkdtemp())
yield temp_path
shutil.rmtree(temp_path, ignore_errors=True)
def test_file_operations(temp_dir):
file_path = temp_dir / "test.txt"
file_path.write_text("Hello, World!")
assert file_path.read_text() == "Hello, World!"
```
---
## 测试最佳实践
1. **测试名称要描述性**: `test_user_can_login``test_login` 更好
2. **AAA 模式**: Arrange (准备) -> Act (执行) -> Assert (断言)
3. **每个测试一个关注点**: 保持测试简短和专注
4. **使用描述性断言消息**:
```python
assert result == expected, f"Expected {expected}, got {result}"
```
5. **避免测试间的依赖**: 每个测试应该独立运行
6. **保持测试快速**: 单元测试应该毫秒级完成
7. **合理使用 Mock**: 不要过度 Mock保持测试接近真实场景
8. **定期运行测试**: 建议配置 CI/CD 自动运行
---
## 常见问题
### Q: 如何测试异常?
```python
import pytest
def test_exception():
with pytest.raises(ValueError) as exc_info:
raise ValueError("error message")
assert str(exc_info.value) == "error message"
```
### Q: 如何测试警告?
```python
import warnings
import pytest
def test_warning():
with warnings.catch_warnings(record=True) as w:
warnings.warn("deprecated", DeprecationWarning)
assert len(w) == 1
assert str(w[0].message) == "deprecated"
```
### Q: 如何测试日志输出?
```python
import logging
import pytest
def test_logging(caplog):
logger = logging.getLogger("test")
logger.warning("test message")
assert "test message" in caplog.text
```
---
## 资源链接
- 官方文档: https://docs.pytest.org/
- Pytest 插件索引: https://plugincompat.herokuapp.com/
- Awesome Pytest: https://github.com/pytest-dev/pytest
---
> 提示: 在本项目中,所有测试文件位于 `tests/` 目录,运行 `pytest` 即可执行全部测试。