unittest 与 unittest.mock：标准库测试与替身

# test_billing.py —— python -m unittest test_billing 直接运行，零配置
import unittest

from billing import split_bill

class TestSplitBill(unittest.TestCase):
    def test_even_split(self):
        self.assertEqual(split_bill(100, 4), (25, 0))

    def test_rejects_zero_people(self):
        with self.assertRaises(ValueError):
            split_bill(100, 0)

    def test_remainder_goes_to_first(self):
        self.assertEqual(split_bill(101, 4), (26, 25, 25, 25))

if __name__ == "__main__":
    unittest.main()

class AssertFamilyDemo(unittest.TestCase):
    def test_family(self):
        # 相等与近似
        self.assertEqual(1 + 1, 2)                     # == 与失败时 diff
        self.assertAlmostEqual(0.1 + 0.2, 0.3, places=7)   # 浮点必用
        # 布尔与成员
        self.assertTrue("a" in "abc")
        self.assertIn("error", "an error occurred")
        self.assertIsNone(None)                        # is None，不是 == None
        # 异常与警告
        with self.assertRaises(ValueError) as cm:
            int("not-a-number")
        self.assertIn("invalid literal", str(cm.exception))  # 断言异常消息
        # 类型与日志
        self.assertIsInstance([], list)

class TestUserRepository(unittest.TestCase):
    def setUp(self):                    # 每个测试方法前各跑一次 ≈ pytest fixture（函数级）
        self.repo = UserRepository(":memory:")   # SQLite 内存库（见 825 篇）

    def tearDown(self):                 # 每个测试方法后各跑一次 ≈ fixture 的 yield 之后
        self.repo.close()

    @classmethod
    def setUpClass(cls):                # 整个类一次 ≈ fixture(scope="class")
        cls.config = load_test_config()

    @classmethod
    def tearDownClass(cls):             # 类结束时一次
        cls.config.cleanup()

    def test_create_user(self):
        uid = self.repo.create("ada")
        self.assertEqual(self.repo.get(uid).name, "ada")

class TestWithCleanup(unittest.TestCase):
    def setUp(self):
        self.tmpdir = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmpdir.cleanup)   # 即使下一行抛异常也会清理
        self.repo = UserRepository(os.path.join(self.tmpdir.name, "db.sqlite"))

from unittest.mock import MagicMock

def test_mock_intuition():
    db = MagicMock()                        # 万能假对象
    db.get_user(42).name = "Ada"            # 链式配置返回值

    result = process(db, 42)                # 被测代码拿着假 db 正常工作
    db.get_user.assert_called_once_with(42) # 断言「以 42 为参恰好调了一次」

from unittest.mock import MagicMock, call

def test_configuration_and_assertions():
    svc = MagicMock()
    # 行为配置
    svc.fetch.return_value = [1, 2, 3]              # 固定返回值
    svc.fetch.side_effect = TimeoutError("slow")    # 抛异常；或给函数/可迭代实现动态返回
    svc.fetch_all.side_effect = lambda q: [q] * 2   # 用函数模拟真实逻辑
    # 调用断言
    svc.fetch("q1")
    svc.fetch.assert_called()                       # 调过（至少一次）
    svc.fetch.assert_called_once_with("q1")         # 恰好一次且参数正确
    svc.fetch.assert_called_with("q1")              # 最近一次参数正确
    assert svc.fetch.call_count == 1
    assert svc.fetch.call_args == call("q1")        # call 对象可比较

# app/client.py —— 被测模块
import http.client

def fetch_status(host: str, path: str) -> int:
    conn = http.client.HTTPConnection(host, timeout=2)
    conn.request("GET", path)
    resp = conn.getresponse()
    return resp.status

# test_client.py
from unittest import mock
import unittest

class TestFetchStatus(unittest.TestCase):
    @mock.patch("app.client.http.client.HTTPConnection")   # 正确：patch 使用处
    def test_timeout_branch(self, mock_conn_cls):
        mock_conn_cls.side_effect = TimeoutError

        with self.assertRaises(TimeoutError):
            fetch_status("api.example.com", "/health")

    @mock.patch("http.client.HTTPConnection")               # 错误示范
    def test_wrong_target(self, mock_conn_cls):
        mock_conn_cls.side_effect = TimeoutError
        # fetch_status 里查的是 app.client.http 这个模块属性，
        # 拿到的仍是真实的 HTTPConnection —— 测试在打真网络，还可能「通过」
        fetch_status("api.example.com", "/health")

from unittest import mock

# 用法一：装饰器（最常见，参数逆序注入）
@mock.patch("app.client.http.client.HTTPConnection")
def test_a(mock_conn): ...

# 用法二：上下文管理器（只在这几行生效，不污染整个方法）
def test_b(self):
    with mock.patch("app.client.get_token", return_value="TEST"):
        ...

# 用法三：setUpClass/start-stop（整个类共享，手动控制）
# 用法四：patch.object(SomeClass, "method")（按对象而非字符串定位，重构友好）
def test_d(self):
    with mock.patch.object(UserRepository, "create", return_value=1):
        ...

# app/service.py
def transfer_points(repo, sender_id: int, receiver_id: int, amount: int) -> None:
    sender = repo.get_user(sender_id)
    if sender.points < amount:
        raise ValueError("余额不足")
    repo.update_points(sender_id, sender.points - amount)
    repo.update_points(receiver_id, sender.points + amount)

# tests/test_service.py
import unittest
from unittest import mock

from app.service import transfer_points

class TestTransferPoints(unittest.TestCase):
    def setUp(self):
        self.repo = mock.MagicMock()
        self.repo.get_user.return_value = mock.Mock(points=100)

    def test_happy_path_updates_both_sides(self):
        transfer_points(self.repo, sender_id=1, receiver_id=2, amount=30)

        self.repo.update_points.assert_has_calls(
            [mock.call(1, 70), mock.call(2, 130)]
        )
        self.assertEqual(self.repo.update_points.call_count, 2)

    def test_insufficient_balance_rejects(self):
        self.repo.get_user.return_value = mock.Mock(points=10)

        with self.assertRaises(ValueError):
            transfer_points(self.repo, 1, 2, 30)

        self.repo.update_points.assert_not_called()   # 拒绝路径必须零写入

import unittest
from unittest import mock

from app.client import fetch_status

class TestFetchStatusBranches(unittest.TestCase):
    def _mock_response(self, status: int) -> mock.Mock:
        resp = mock.Mock()
        resp.status = status
        return resp

    @mock.patch("app.client.http.client.HTTPConnection")
    def test_returns_status(self, cls):
        cls.return_value.getresponse.return_value = self._mock_response(200)
        self.assertEqual(fetch_status("h", "/x"), 200)

    @mock.patch("app.client.http.client.HTTPConnection")
    def test_timeout_maps_to_custom_error(self, cls):
        cls.side_effect = TimeoutError
        with self.assertRaisesRegex(TimeoutError, "健康检查超时"):
            fetch_status("h", "/x")

    @mock.patch("app.client.http.client.HTTPConnection")
    def test_503_still_returns_int(self, cls):
        cls.return_value.getresponse.return_value = self._mock_response(503)
        self.assertEqual(fetch_status("h", "/x"), 503)

# app/client.py
import http.client

def fetch_status(host: str, path: str) -> int:
    try:
        conn = http.client.HTTPConnection(host, timeout=2)
        conn.request("GET", path)
        resp = conn.getresponse()
        return resp.status
    except (TimeoutError, ConnectionError) as exc:
        raise TimeoutError(f"健康检查超时: {host}") from exc   # 异常链见 130 篇

# app/cli.py
import argparse
import sys

from billing import split_bill

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="bill")
    parser.add_argument("total", type=int)
    parser.add_argument("people", type=int)
    args = parser.parse_args(argv)
    try:
        shares = split_bill(args.total, args.people)
    except ValueError as exc:
        print(f"错误: {exc}", file=sys.stderr)
        return 2
    print(" ".join(map(str, shares)))
    return 0

# tests/test_cli.py —— unittest 风格的 CLI 集成测试
import contextlib
import io
import unittest

from app.cli import main

class TestCli(unittest.TestCase):
    def test_prints_shares_and_exit_zero(self):
        stdout = io.StringIO()
        with contextlib.redirect_stdout(stdout):
            code = main(["101", "4"])
        self.assertEqual(code, 0)
        self.assertEqual(stdout.getvalue().strip(), "26 25 25 25")

    def test_bad_input_gives_exit_two_and_stderr(self):
        stderr = io.StringIO()
        with contextlib.redirect_stderr(stderr):
            code = main(["100", "0"])
        self.assertEqual(code, 2)
        self.assertIn("错误", stderr.getvalue())

pytest tests/                      # unittest 的 TestCase 一样被收集运行
pytest tests/test_legacy.py -v     # 旧类式测试也吃 -v / -x / -k 全套选项

import unittest
from unittest import mock

from app.client import fetch_status

class TestClient(unittest.TestCase):
    @mock.patch("http.client.HTTPConnection")
    def test_status(self, mock_conn):
        mock_conn.return_value.getresponse.return_value.status = 200
        self.assertEqual(fetch_status("h", "/x"), 200)

def test_alerts_both_channels(self):
    pager = mock.Mock()
    slack = mock.Mock()

    alert_all([pager, slack], "磁盘 91%")

    pager.send.assert_called_with("磁盘 91%")
    slack.send.assert_called_with("磁盘 91%")
    pager.send.assert_called_once_with("磁盘 91%")