（二十一）核对并发行为与实际经过时间

1. 固定数据、真实连接与失败后的状态

第六章的 pytest 方法继续使用。异步函数可以在普通测试函数里由 asyncio.run 执行；测试本身不必一开始就增加异步测试插件：

# 项目文件：network_project/tests/conftest.py
# 每次测试使用独立服务状态与随机空闲端口，退出后关闭监听与处理收尾。
import pytest

from labserver import running_server


@pytest.fixture
def server():
    with running_server() as opened:
        yield opened

# 项目文件：network_project/tests/test_import.py
# 正常数据按输入保存，失败响应和坏字段都不能把已有状态改成半批结果。

import pytest

from bookdata.errors import RecordError
from bookdata.library import Library
from bookdata.models import Book
from netreading.service import import_remote


def test_import_order_and_persistence(server, tmp_path):
    path = tmp_path / "books.sqlite3"
    paths = [
        "/api/books/1?delay=0.06",
        "/api/books/2?delay=0.02",
        "/api/books/3?delay=0.04",
    ]
    assert import_remote(server.base_url, paths, path, concurrency=2) == 3
    with Library(path) as library:
        assert [row["title"] for row in library.search()] == [
            "Python",
            "NumPy",
            "Pygame",
        ]
    assert server.state.snapshot()["peak"] <= 2


@pytest.mark.parametrize(
    "bad_path, error_type",
    [
        ("/api/status/503", ExceptionGroup),
        ("/api/malformed", ExceptionGroup),
        ("/api/bad-record", RecordError),
    ],
)
def test_failure_preserves_existing_database(server, tmp_path, bad_path, error_type):
    path = tmp_path / "books.sqlite3"
    with Library(path) as library:
        library.import_books([Book("原有书籍", True, ["保留"])])
        before = library.export_books()
    with pytest.raises(error_type):
        import_remote(server.base_url, ["/api/books/1", bad_path], path)
    with Library(path) as library:
        assert library.export_books() == before

本地服务用锁记录处理中的数量，peak 观察服务处理段有没有超过两个；结果顺序另外根据数据库内容核对。失败测试比较原有记录，检查的不只是有没有异常。

配套测试还分别检查请求超时、整体时间、取消后的客户端关闭、同组任务收尾、Cookie、重定向、连接复用与命令行退出状态。取消与任务组的检查用事件通知协调开始位置，不把“多睡一会儿”当成同步规则。

2. 计时只比较经过的那段工作

配套 bench_waits.py 把同样三条延迟请求分别交给顺序代码、线程池与异步客户端：

# 在项目根目录比较相同输入；打印结果、处理峰值与本次实际经过时间。
uv run python bench_waits.py
# 运行完整的网络、取消、校验和导入检查。
uv run python -m pytest

每轮都使用新的本地服务状态，比较三本书是否相同，再看各自的等待重叠。AsyncClient 的建立、连接、调度、解析和退出也在计时范围中，因此实际时间不会恰好等于第一节的等待模型。小任务的额外开销可能盖过重叠收益，先保留正确性检查，再调整批量与并发数。