（十四）观察输出、日志与服务协作

1. 服务层协调读取与事务

把“从文件导入到数据库”组合成一个服务函数。它调用第五章的格式读写器和 Library，不参与终端参数解析：

# 项目文件：quality_project/src/readingapp/services.py
# 先读出并校验整份书目，再交给数据库；成功提交后记录导入完成。
import logging
from pathlib import Path
from typing import Literal

from bookdata.csv_io import read_csv
from bookdata.json_io import read_json
from bookdata.library import Library

logger = logging.getLogger(__name__)


def import_file(source: Path, database: Path, kind: Literal["json", "csv"]) -> int:
    reader = read_json if kind == "json" else read_csv
    books = reader(source)
    with Library(database) as library:
        count = library.import_books(books)
    logger.info("导入完成 count=%d source=%s", count, source.name)
    return count

Literal["json", "csv"] 来自 typing，把注解中的可选值写成两个文本值；它沿用第三章的静态注解，不会在运行时自行检查。CLI 的动作分支负责只传这两个值。文件格式错误发生在打开数据库前，因此失败导入不会先建立一份空库。

2. caplog 检查记录，替身检查传入的数据

# 项目文件：quality_project/tests/integration/test_services.py
# caplog 收集业务模块发出的日志；这里不重新配置根记录器。
import logging

from bookdata.json_io import write_json
from bookdata.library import Library
from bookdata.models import Book
from readingapp import services


def test_import_records_log(tmp_path, caplog):
    source = tmp_path / "books.json"
    write_json(source, [Book("Python")])
    database = tmp_path / "library.sqlite3"
    with caplog.at_level(logging.INFO, logger="readingapp.services"):
        assert services.import_file(source, database, "json") == 1
    messages = [
        record.getMessage() for record in caplog.records if record.name == "readingapp.services"
    ]
    assert messages == ["导入完成 count=1 source=books.json"]
    with Library(database) as library:
        assert library.search()[0]["title"] == "Python"

caplog.at_level 临时接收指定等级，caplog.records 保存 LogRecord 对象。比较 getMessage() 得到的业务文本，不比较每次变化的时间戳。pytest 日志检查

接着只检查服务层协调，用内存对象替代打开数据库的动作：

# 续写同一文件：quality_project/tests/integration/test_services.py
# FakeLibrary 记录收到的数据；补丁目标是 services 查找 Library 的位置。
def test_import_passes_validated_books(tmp_path, monkeypatch):
    source = tmp_path / "books.json"
    write_json(source, [Book("Python", True, ["基础"])])
    database = tmp_path / "unused.sqlite3"
    received = []

    class FakeLibrary:
        def __init__(self, path):
            assert path == database

        def __enter__(self):
            return self

        def __exit__(self, exc_type, exc, traceback):
            received.append("关闭")
            return False

        def import_books(self, books):
            received.extend(books)
            return len(books)

    monkeypatch.setattr(services, "Library", FakeLibrary)
    assert services.import_file(source, database, "json") == 1
    assert received == [Book("Python", True, ["基础"]), "关闭"]
    assert not database.exists()

这份测试证明解析结果被传给存储器，并且服务离开了上下文。真实提交、回滚与重新打开则由前面的 SQLite 测试检查。测试范围依据实际替换的动作判断，不必仅凭目录名决定。

capsys 与 caplog 相似，但它接收 Python 的标准输出和标准错误输出。下一节用 captured = capsys.readouterr() 分别读取 captured.out 和 captured.err，检查入口给使用者显示的结果。输出捕获