JEPA4Japan · 教程

测试、调试与日志记录

1,993字 7分钟阅读 #Python

验证行为、调查故障,并留下有用的运行证据。

课程进度 课程大纲 已发布 24/24 课

三种工具回答三个问题

程序成功运行一次固然很好,但这并不能说明太多问题。测试、调试器和日志各自以不同的方式提供帮助。

  1. 测试 约定是否仍然成立?
  2. 调试器 现在正在发生什么?
  3. 日志 一次运行中发生了什么?
每种工具收集的证据类型都不同。

一次调试会话不会留下可重复执行的测试。一堆日志也无法证明程序正确无误。请选择能够回答当前问题的最小工具。

让测试简单、稳定且可重复

确定性测试会控制自己的输入,并检查公开行为。它不应该暗中依赖时钟、网络、随机状态、当前文件夹、测试顺序或其他测试留下的内容。

  1. 准备 创建受控输入
  2. 执行 运行一种行为
  3. 断言 与约定进行比较
一个小型测试应当只有一套清晰的准备步骤、一个操作和一个断言。

unittest 会发现以 test 开头的方法。结构相同的边界情况很适合整齐地放进 subTest():

import unittest


def pace(minutes):
    if minutes < 0:
        raise ValueError("minutes must not be negative")
    return "short" if minutes <= 20 else "long"


class PaceTests(unittest.TestCase):
    def test_boundaries(self):
        for minutes, expected in [(0, "short"), (20, "short"), (21, "long")]:
            with self.subTest(minutes=minutes):
                self.assertEqual(pace(minutes), expected)

    def test_negative_value_fails(self):
        with self.assertRaisesRegex(ValueError, "negative"):
            pace(-1)


suite = unittest.defaultTestLoader.loadTestsFromTestCase(PaceTests)
result = unittest.TestResult()
suite.run(result)
print(result.testsRun, len(result.failures), len(result.errors))

输出:

2 0 0

请测试空输入、最后一个可接受的值、第一个被拒绝的值以及预期会出现的失败。只有当浮点数舍入是约定的一部分时,才使用 assertAlmostEqual();对于字符串、整数和集合,应优先使用精确断言。失败的测试应当指向一个出错的行为,而不是重复实现中的每个细节。

只模拟狭窄的入口

模拟对象是一位虚构的协作者,它会记住调用情况。请在发布器、时钟或客户端之类的外部入口处使用模拟对象,而不要为每个纯辅助函数都创建模拟对象。

from unittest.mock import Mock


def publish(summary, sender):
    sender(summary)
    return "sent"


sender = Mock()
print(publish({"minutes": 45}, sender))
print(sender.call_args)

输出:

sent
call({'minutes': 45})

依赖注入能让这个入口清晰可见;测试可以使用 sender.assert_called_once_with(...)。如果需要使用 patch(),请在被测代码查找该名称的位置进行修补。如果 report.py 直接导入了 send,就修补 report.send,而不是最初定义 send 的模块。对于真实接口,请优先使用 spec= 或 autospec=True,并让大部分真实代码保持连接。

修改代码之前先阅读线索

遇到异常时,先阅读回溯信息的最后一行,也就是异常类型和消息。然后从最近的栈帧开始向上查看,找出实际状态最先偏离预期的位置。在添加领域相关的描述时,请使用 raise NewError(...) from cause 保留底层证据。

  1. 阅读 先看最后一行,再看各个栈帧
  2. 缩小 找出最小的错误输入
  3. 暂停 使用 breakpoint() 检查状态
  4. 修复 修改一个错误的假设
先收集证据,再修改代码。
def safe_ratio(part, whole, *, debug=False):
    if debug:
        breakpoint()
    if whole == 0:
        raise ValueError("whole must not be zero")
    return part / whole


print(safe_ratio(6, 3))

正常输出是 2.0。当 debug=True 时,在常见的 pdb 提示符中,你可以使用 p expression,用 n 执行下一行,用 s 进入函数内部,用 where 查看调用栈,用 c 继续执行,以及用 q 退出。不要使用会改变状态的表达式进行检查。在运行自动化流程之前,请移除无条件执行的断点;否则程序可能会永远等待一个并不存在的人。

构建一份可观察的学习报告

日志记录是一种数据流:logger 接收记录,handler 选择输出目标和级别,formatter 决定稳定的文本格式。请在入口点配置由应用程序管理的 logger,不要在可复用的库中配置 root logger。

将下面这个完整的标准库项目保存为 observable_report.py:

import logging
import sys
import unittest
from dataclasses import dataclass
from io import StringIO
from unittest.mock import Mock


@dataclass(frozen=True)
class Record:
    topic: str
    minutes: int


def parse_record(line):
    topic, separator, minutes_text = line.partition("|")
    if not separator or not topic.strip():
        raise ValueError("expected topic|minutes")
    try:
        minutes = int(minutes_text)
    except ValueError as cause:
        raise ValueError("minutes must be an integer") from cause
    if minutes <= 0:
        raise ValueError("minutes must be positive")
    return Record(topic.strip(), minutes)


def load_records(lines, logger):
    records = []
    for number, line in enumerate(lines, start=1):
        try:
            record = parse_record(line)
        except ValueError as error:
            logger.warning(
                "skipped line %d: %s",
                number,
                error,
                extra={"event": "invalid_record"},
            )
        else:
            records.append(record)
    logger.info(
        "loaded %d records",
        len(records),
        extra={"event": "import_complete"},
    )
    return records


def summarize(records):
    return {"count": len(records), "minutes": sum(r.minutes for r in records)}


def publish_summary(records, publisher):
    summary = summarize(records)
    publisher(summary)
    return summary


def configure_logger(stream):
    logger = logging.getLogger("observable_report")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler(stream)
    handler.setLevel(logging.INFO)
    handler.setFormatter(logging.Formatter("%(levelname)s %(event)s %(message)s"))
    logger.addHandler(handler)
    return logger


class ReportTests(unittest.TestCase):
    def test_boundaries(self):
        cases = [("Testing | 1", Record("Testing", 1)), (" Logging | 45 ", Record("Logging", 45))]
        for text, expected in cases:
            with self.subTest(text=text):
                self.assertEqual(parse_record(text), expected)

    def test_invalid_minutes_fail(self):
        with self.assertRaisesRegex(ValueError, "integer"):
            parse_record("Testing | many")

    def test_logs_are_stable(self):
        stream = StringIO()
        records = load_records(["Testing | 20", "Bad | zero"], configure_logger(stream))
        self.assertEqual(records, [Record("Testing", 20)])
        self.assertEqual(
            stream.getvalue().splitlines(),
            [
                "WARNING invalid_record skipped line 2: minutes must be an integer",
                "INFO import_complete loaded 1 records",
            ],
        )

    def test_publisher_doorway(self):
        publisher = Mock()
        summary = publish_summary([Record("Testing", 20)], publisher)
        self.assertEqual(summary, {"count": 1, "minutes": 20})
        publisher.assert_called_once_with(summary)


suite = unittest.defaultTestLoader.loadTestsFromTestCase(ReportTests)
test_result = unittest.TestResult()
suite.run(test_result)
if not test_result.wasSuccessful():
    raise AssertionError(test_result.failures + test_result.errors)
print(f"Tests: {test_result.testsRun}, failures: 0, errors: 0")

logger = configure_logger(sys.stdout)
records = load_records(["Testing | 30", "Broken | many", "Logging | 25"], logger)
summary = summarize(records)
print(f"Total: {summary['count']} records, {summary['minutes']} min")

运行 python3 observable_report.py:

Tests: 4, failures: 0, errors: 0
WARNING invalid_record skipped line 2: minutes must be an integer
INFO import_complete loaded 2 records
Total: 2 records, 55 min

测试拥有自己的 StringIO,而 propagate = False 可以防止祖先 logger 重复输出。时间戳、机器路径、全局配置或网络都不会让检查结果变得不稳定。稳定的 event 字段可以添加便于搜索的上下文;绝不要在 extra 中放入机密信息或 LogRecord 的保留名称。

三个小任务

  1. 为零分钟、空白主题和最小的有效分钟数添加子测试。
  2. 让注入的模拟 publisher 抛出 RuntimeError,并测试该失败仍然清晰可见。
  3. 为每条日志记录添加一个稳定的 source 字段,并更新 formatter 和精确日志测试。

准备好学习第 20 章了吗?

  • 我能针对成功情况、边界情况和失败情况编写确定性的 unittest 测试用例。
  • 我能使用 subTest() 标记重复的测试情况。
  • 我只模拟一个协作者入口,而不是每个内部辅助函数。
  • 我会在修改代码之前阅读回溯信息,并有意识地使用 breakpoint()。
  • 我能配置具名 logger、handler、level、formatter 和稳定的上下文。
  • 我运行了这份报告,并测试了程序行为、日志和 publisher。

接下来,你将运用这些技能完善、打包并发布一个完整的命令行应用程序。