05-重构与补测试

识别风险 -> 建立行为基线 -> 先补测试 -> 小步重构 -> 每步验证
-> 检查兼容性和性能 -> 审查 diff -> 预备回滚

git status --short --branch
git branch --show-current
git log -5 --oneline
find . -maxdepth 2 -type f \( -name 'AGENTS.md' -o -name 'pyproject.toml' -o -name 'Makefile' \) -print
python --version
pytest --version

请只读检查 @src/order_service.py、所有调用方和现有测试。
先不要修改。输出入口、返回值、异常、副作用和边界缺口，
区分事实与推测，最后给出分步计划和每步回滚方式。

rg "get_order_summary|OrderService|order_service" src tests scripts
rg "status|total|items|None|TimeoutError" src tests

python -m pytest tests/test_orders.py -q
python -m pytest tests/test_orders.py -q -x

@pytest.mark.parametrize(
    ("raw", "expected"),
    [([], []), ([{"id": "a"}], [{"id": "a"}])],
)
def test_sort_orders_keeps_contract(raw, expected):
    assert sort_orders(raw) == expected


def test_timeout_is_not_silently_converted():
    with pytest.raises(TimeoutError, match="orders"):
        load_orders("u-1", repository=TimeoutRepository())

@pytest.fixture(autouse=True)
def reset_state():
    clear_order_cache()
    yield
    clear_order_cache()

# order_service.py
_orders_cache = {}

def get_order_summary(user_id, repository, currency="CNY"):
    if user_id is None:
        return None
    if user_id in _orders_cache:
        orders = _orders_cache[user_id]
    else:
        try:
            orders = repository.find_by_user(user_id)
        except Exception:
            return []
        _orders_cache[user_id] = orders
    result = []
    for order in orders:
        if order.get("cancelled"):
            continue
        items = repository.find_items(order["id"])
        total = 0
        for item in items:
            if item.get("deleted"):
                continue
            total += item.get("price", 0) * item.get("quantity", 1)
        if currency == "CNY":
            total = round(total, 2)
        result.append({"id": order["id"], "total": total,
                       "item_count": len(items)})
    result.sort(key=lambda row: row["id"])
    return result

class Repository:
    def __init__(self, orders=None, items=None, error=None):
        self.orders = orders or []
        self.items = items or {}
        self.error = error
        self.calls = []

    def find_by_user(self, user_id):
        self.calls.append(("find_by_user", user_id))
        if self.error:
            raise self.error
        return self.orders

    def find_items(self, order_id):
        self.calls.append(("find_items", order_id))
        return self.items.get(order_id, [])

def test_summary_filters_and_sorts():
    repo = Repository(
        orders=[{"id": "o-2"}, {"id": "o-1", "cancelled": True}],
        items={"o-2": [{"price": 3, "quantity": 2}]},
    )
    assert get_order_summary("u-1", repo) == [
        {"id": "o-2", "total": 6, "item_count": 1}
    ]


def test_none_user_is_legacy_none():
    repo = Repository()
    assert get_order_summary(None, repo) is None
    assert repo.calls == []


def test_deleted_item_is_not_in_total_but_is_in_legacy_count():
    repo = Repository(
        orders=[{"id": "o-1"}],
        items={"o-1": [
            {"price": 10, "quantity": 1, "deleted": True},
            {"price": 2, "quantity": 3},
        ]},
    )
    assert get_order_summary("u-1", repo) == [
        {"id": "o-1", "total": 6, "item_count": 2}
    ]


def test_repository_timeout_is_currently_hidden():
    assert get_order_summary("u-1", Repository(error=TimeoutError())) == []

python -m pytest -q

git add order_service.py test_order_service.py
git commit -m "test: record order summary behavior"

def _calculate_total(items, currency):
    total = 0
    for item in items:
        if item.get("deleted"):
            continue
        total += item.get("price", 0) * item.get("quantity", 1)
    return round(total, 2) if currency == "CNY" else total

python -m pytest tests/test_order_service.py -q
python -m pytest -q

def _load_orders(user_id, repository):
    try:
        return repository.find_by_user(user_id)
    except Exception:
        return []


def _build_summary(orders, repository, currency):
    result = []
    for order in orders:
        if order.get("cancelled"):
            continue
        items = repository.find_items(order["id"])
        result.append({
            "id": order["id"],
            "total": _calculate_total(items, currency),
            "item_count": len(items),
        })
    return sorted(result, key=lambda row: row["id"])

def get_order_summary(user_id, repository, currency="CNY"):
    if user_id is None:
        return None
    if user_id not in _orders_cache:
        _orders_cache[user_id] = _load_orders(user_id, repository)
    return _build_summary(_orders_cache[user_id], repository, currency)

class OrderService:
    def __init__(self, repository, cache=None):
        self.repository = repository
        self.cache = cache if cache is not None else {}

    def get_summary(self, user_id, currency="CNY"):
        if user_id is None:
            return None
        if user_id not in self.cache:
            self.cache[user_id] = _load_orders(user_id, self.repository)
        return _build_summary(self.cache[user_id], self.repository, currency)


def get_order_summary(user_id, repository, currency="CNY"):
    return OrderService(repository, cache=_orders_cache).get_summary(
        user_id, currency
    )

def _build_summary_with_batch(orders, repository, currency):
    visible = [o for o in orders if not o.get("cancelled")]
    ids = [o["id"] for o in visible]
    grouped = repository.find_items_by_order_ids(ids) if ids else {}
    result = []
    for order in visible:
        items = grouped.get(order["id"], [])
        result.append({
            "id": order["id"],
            "total": _calculate_total(items, currency),
            "item_count": len(items),
        })
    return sorted(result, key=lambda row: row["id"])

python -m timeit -s "from benchmark import run" "run(1000)"
python -m cProfile -s cumulative benchmark.py
python -m pytest tests/benchmark -q

1. test: record current behavior
2. refactor: extract total calculation
3. refactor: introduce service adapter
4. perf: add batch loading behind flag
5. chore: remove deprecated entry point

git diff -- order_service.py test_order_service.py
git restore --source=HEAD -- order_service.py

命令：python -m pytest tests/test_order_service.py -q
分类：实现回归
首个失败：重复 ID 的结果数量改变
事实：批量路径按 ID 去重，旧路径返回两条
影响：公共返回列表长度变化
下一步：回退批量替换，确认重复 ID 契约

任务：重构 @src/order_service.py，保持 get_order_summary 兼容。
先不要修改。读取调用方、测试和项目规则，输出当前行为基线、坏味道、
先补哪些测试、每步修改文件、测试命令、性能检查和回滚方式。
把事实、推测和需要我决定的行为变化分开。

只执行里程碑 1：为当前行为补测试。
不要重构实现、不要添加依赖。先让回归测试在旧实现上运行，
报告失败是否对应目标问题，完成后展示 diff 和测试命令。

只执行里程碑 2：抽出纯金额计算。
保持签名、默认值、异常、输出字段和排序不变；不要处理缓存、批量查询
或接口改名。运行目标测试和完整测试，逐条报告结果。

只执行里程碑 3：新增 OrderService，让旧入口作为适配器调用它。
保持参数、返回字段、None 语义、异常语义和导入路径。
先列出调用方，增加兼容测试，不要删除旧入口。

只执行里程碑 4：评估批量读取明细。
先记录单条路径耗时、查询数和结果。批量路径必须通过结果等价、空输入、
重复 ID、缺失明细和部分失败测试；契约不清楚就停止并列问题。

git diff --check
python -m pytest tests/test_order_service.py -q
python -m pytest -q
python -m compileall src tests
ruff check .
mypy src
python -m pytest tests/benchmark -q
git status --short
git diff --stat
git diff --name-only

[ ] 确认仓库、分支、工作区和项目规则
[ ] 找到入口、调用方、现有测试和运行命令
[ ] 记录返回值、异常、副作用、性能和状态基线
[ ] 先补正常、边界和失败场景测试，并验证回归测试会失败
[ ] 按单一目的拆分里程碑，每步限制文件范围并运行测试
[ ] 检查签名、字段、排序、异常和导入路径兼容
[ ] 单独验证缓存、批量查询、性能和特性开关
[ ] 记录失败分类、未验证假设和回滚方式
[ ] 运行测试、lint、类型检查、构建和 diff 检查
[ ] 确认没有无关文件、敏感数据和临时产物