ShowDocs
搜索文章Ctrl K

外观

← 返回Python列表
开发实践

pandas 完整实践：销售质量校验与客户月报 · 错误样本与测试规格至综合练习 ​

返回分篇目录

前置：连接与分组。目标是把销售事实表与客户地区维度合并，在上海时区按月统计订单数与总分。所有金额和日期规则都在进入聚合前验证；发现坏数据就不发布整份报告。示例未执行，以下输出是预期结果，静态检查不等于运行正确。

错误样本与测试规格 ​

pandas月报应测试哪些失败：重复销售、重复客户、未知客户、金额非法、空ID、无偏移时间及超出当地年份都必须拒绝，并断言既有输出内容不变。测试还用mock令替换失败，核对临时文件清理；正常场景按字节对比固定CSV。这些是待读者执行的行为规格，当前仅验证测试源码AST能解析。

错误输入：逐一替换对应销售或客户文件
text
customer_id,region
001,east
001,west
text
sale_id,customer_id,sold_at,amount_cents
s01,999,2025-01-31T15:59:59Z,1000
text
sale_id,customer_id,sold_at,amount_cents
s01,001,2025-01-31T15:59:59Z,1.25
text
sale_id,customer_id,sold_at,amount_cents
s01,001,2025-01-31T15:59:59,1000
text
sale_id,customer_id,sold_at,amount_cents
s01,,2025-01-31T15:59:59Z,1000
text
sale_id,customer_id,sold_at,amount_cents
s01,001,2025-12-31T16:00:00Z,1000
text
sale_id,customer_id,sold_at,amount_cents
s01,001,2025-01-31T15:59:59Z,1000
s01,001,2025-01-31T15:59:59Z,1000
test_report.py：完整测试规格，未执行
python
"""Behavior specifications; 示例未执行. Run only in a prepared environment."""
from pathlib import Path
from tempfile import TemporaryDirectory
import unittest
from unittest.mock import patch

from report import build_report, main, publish

HERE = Path(__file__).resolve().parent


class SalesReportTests(unittest.TestCase):
    def test_golden_cli(self):
        with TemporaryDirectory() as folder:
            target = Path(folder) / 'monthly.csv'
            self.assertEqual(main([str(HERE / 'input/sales.csv'),
                                   str(HERE / 'input/customers.csv'), str(target)]), 0)
            self.assertEqual(target.read_bytes(), (HERE / 'expected/monthly.csv').read_bytes())

    def test_invalid_input_preserves_report(self):
        for name in ['unknown-customer', 'invalid-money', 'naive-time', 'missing-id',
                     'outside-year', 'duplicate-sale', 'duplicate-customers']:
            with self.subTest(name=name), TemporaryDirectory() as folder:
                target = Path(folder) / 'report.csv'
                target.write_bytes(b'previous report\n')
                sales = HERE / ('input/sales.csv' if name == 'duplicate-customers'
                                else f'invalid/{name}.csv')
                customers = HERE / ('invalid/duplicate-customers.csv' if name == 'duplicate-customers'
                                    else 'input/customers.csv')
                self.assertEqual(main([str(sales), str(customers), str(target)]), 1)
                self.assertEqual(target.read_bytes(), b'previous report\n')
                self.assertEqual(list(Path(folder).glob('.sales-*.tmp')), [])

    def test_replace_failure_removes_temporary(self):
        report = build_report(HERE / 'input/sales.csv', HERE / 'input/customers.csv')
        with TemporaryDirectory() as folder:
            target = Path(folder) / 'report.csv'
            target.write_bytes(b'old\n')
            with patch('report.os.replace', side_effect=OSError('locked')):
                with self.assertRaises(OSError):
                    publish(report, target)
            self.assertEqual(target.read_bytes(), b'old\n')
            self.assertEqual(list(Path(folder).glob('.sales-*.tmp')), [])

    def test_schema_and_bounds(self):
        header = 'sale_id,customer_id,sold_at,amount_cents\n'
        cases = [header, 'wrong,header\n', header + 's,001,2025-01-01T00:00:00Z,1,extra\n',
                 header + 's,001,2025-02-30T00:00:00Z,1\n',
                 header + 's,001,2025-01-01T00:00:00Z,1000000000001\n',
                 header + 's,001,2025-01-01T00:00:00Z,-1\n']
        for source in cases:
            with self.subTest(source=source), TemporaryDirectory() as folder:
                path = Path(folder) / 'sales.csv'
                path.write_text(source, encoding='utf-8')
                with self.assertRaises(ValueError):
                    build_report(path, HERE / 'input/customers.csv')

    def test_destination_cannot_be_input(self):
        with TemporaryDirectory() as folder:
            sales = Path(folder) / 'sales.csv'
            original = (HERE / 'input/sales.csv').read_bytes()
            sales.write_bytes(original)
            self.assertEqual(main([str(sales), str(HERE / 'input/customers.csv'), str(sales)]), 1)
            self.assertEqual(sales.read_bytes(), original)


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

综合练习 ​

先不改代码，手算五笔销售在上海月份下的分组。再设计一条能通过时间正则、但必须在日期解析失败的输入。最后若要允许退款，列出需要改的校验和测试，不只删除一个负号检查。

参考解法

一月east1000；二月east2000、unknown0、west3000；三月west1500，每组一笔。2025-02-30T00:00:00Z格式完整但日期不存在。退款需要带符号语法、业务身份与冲销关联、分组正负累计上界和净额口径；当前非负全局和保护分组的证明会失效，应先形成新契约再实施。

继续阅读