零基础入门python29：用 pytest 验收完整 Flask 业务流程

零基础入门python29用 pytest 验收完整 Flask 业务流程一、上一篇课后练习讲解报表kind筛选应在基础查询上追加Transaction.kind kind无效月份在解析阶段返回 400。测试不仅断言响应还要断言用户隔离和数据库状态。上一篇课后练习完整答案上一篇练习的要求已落实到下面完整文件先运行项目测试再用 curl 对照状态码和数据库持久化结果答案要点报表用 SUM/CASE/COALESCE 在数据库聚合过滤用户与软删除记录Decimal 输出避免浮点误差。文件app/reports.py完整参考答案文件完整文件app/reports.pyfromdecimalimportDecimalfromflaskimportBlueprint,requestfromflask_loginimportlogin_required,current_userfromsqlalchemyimportfunc,casefrom.extensionsimportdbfrom.modelsimportEntry bpBlueprint(reports,__name__,url_prefix/api/reports)bp.get(/month)login_requireddefmonth():start,endrequest.args[start],request.args[end]incomefunc.coalesce(func.sum(case((Entry.kindincome,Entry.amount),else_0)),0)expensefunc.coalesce(func.sum(case((Entry.kindexpense,Entry.amount),else_0)),0)rowdb.session.query(income,expense).filter(Entry.user_idcurrent_user.id,Entry.happened_on.between(start,end)).one()return{income:str(Decimal(row[0])),expense:str(Decimal(row[1]))}完整参考答案文件本篇对应的交付源码完整文件flask-ledger/app/ledger.pyfromdatetimeimportdatefromdecimalimportDecimal,InvalidOperationfromflaskimportBlueprint,requestfromflask_loginimportcurrent_user,login_requiredfromsqlalchemyimportfuncfrom.extensionsimportdbfrom.modelsimportCategory,Transaction bpBlueprint(ledger,__name__,url_prefix/api)defowned_category(category_id:int):returndb.session.scalar(db.select(Category).where(Category.idcategory_id,Category.user_idcurrent_user.id))bp.get(/categories)login_requireddeflist_categories():rowsdb.session.scalars(db.select(Category).where(Category.user_idcurrent_user.id).order_by(Category.name)).all()return[{id:row.id,name:row.name}forrowinrows]bp.post(/categories)login_requireddefcreate_category():namestr((request.get_json(silentTrue)or{}).get(name,)).strip()ifnot1len(name)40:return{message:分类名称长度应为1到40},400existsdb.session.scalar(db.select(Category).where(Category.user_idcurrent_user.id,Category.namename))ifexists:return{message:分类已存在},409rowCategory(namename,user_idcurrent_user.id)db.session.add(row)db.session.commit()return{id:row.id,name:row.name},201bp.post(/transactions)login_requireddefcreate_transaction():datarequest.get_json(silentTrue)or{}try:amountDecimal(str(data.get(amount,0))).quantize(Decimal(0.01))happened_ondate.fromisoformat(str(data.get(happened_on,date.today())))category_idint(data.get(category_id))except(InvalidOperation,ValueError,TypeError):return{message:金额、日期或分类格式不正确},400kindstr(data.get(kind,))ifkindnotin{income,expense}oramount0:return{message:类型或金额不正确},400ifnotowned_category(category_id):return{message:分类不存在},404rowTransaction(kindkind,amountamount,notestr(data.get(note,))[:200],happened_onhappened_on,category_idcategory_id,user_idcurrent_user.id,)db.session.add(row)db.session.commit()returnrow.to_dict(),201bp.get(/transactions)login_requireddeflist_transactions():pagemax(request.args.get(page,1,typeint),1)sizemin(max(request.args.get(size,10,typeint),1),100)querydb.select(Transaction).where(Transaction.user_idcurrent_user.id)ifkind:request.args.get(kind):queryquery.where(Transaction.kindkind)ifcategory_id:request.args.get(category_id,typeint):queryquery.where(Transaction.category_idcategory_id)rowsdb.session.scalars(query.order_by(Transaction.happened_on.desc(),Transaction.id.desc()).offset((page-1)*size).limit(size)).all()return{page:page,size:size,items:[row.to_dict()forrowinrows]}bp.delete(/transactions/int:transaction_id)login_requireddefdelete_transaction(transaction_id:int):rowdb.session.scalar(db.select(Transaction).where(Transaction.idtransaction_id,Transaction.user_idcurrent_user.id))ifnotrow:return{message:账目不存在},404db.session.delete(row)db.session.commit()return,204bp.get(/statistics/monthly)login_requireddefmonthly_statistics():monthrequest.args.get(month,date.today().strftime(%Y-%m))try:startdate.fromisoformat(month-01)exceptValueError:return{message:月份格式应为YYYY-MM},400enddate(start.year(start.month12),1ifstart.month12elsestart.month1,1)rowsdb.session.execute(db.select(Transaction.kind,func.sum(Transaction.amount)).where(Transaction.user_idcurrent_user.id,Transaction.happened_onstart,Transaction.happened_onend).group_by(Transaction.kind)).all()totals{income:Decimal(0),expense:Decimal(0)}totals.update({kind:Decimal(total)forkind,totalinrows})return{month:month,income:str(totals[income]),expense:str(totals[expense]),balance:str(totals[income]-totals[expense])}验收命令python -m pytest -qDjango 项目使用 python manage.py test。预期测试通过若失败先检查迁移、配置和事务回滚。二、本篇测试目标从注册、登录、创建分类、创建账目、读取报表到删除账目写一条完整业务流程再写错误场景验证事务没有留下脏数据。三、测试夹具pytest.fixture()defapp(tmp_path):appcreate_app({TESTING:True,SQLALCHEMY_DATABASE_URI:fsqlite:///{tmp_path/test.db}})withapp.app_context():db.create_all()yieldapp db.session.remove()db.drop_all()每个测试使用独立临时文件测试结束删除不能使用开发数据库否则测试可能删除真实账目。四、验收重点必须覆盖未登录访问 401、重复注册 409、错误金额 400、第二个用户看不到第一个用户数据、报表金额正确、删除后记录不存在。运行python-m pytest-q课后练习为月度报表和分类排行各增加一个断言并测试无效月份。项目增量pytest 验收一条完整业务链测试从注册开始经过登录、分类、账目新增、筛选、报表和越权请求每一步使用前一步产生的 id才能证明项目真正串联起来。deftest_ledger_flow(client):register(client,aexample.com)login(client,aexample.com)categoryclient.post(/categories,data{name:餐饮})entryclient.post(/entries,data{amount:35.00,kind:expense,category_id:category.json[id]})assertentry.status_code201assertclient.get(/reports/month).json[expense]35.00失败场景同样重要重复邮箱、非法金额、他人 id、数据库约束和 session 失效。课后练习把删除账目加入流程并保存 pytest 输出。五、从单接口测试到用户故事测试单元测试只能证明某个函数返回值正确后端更需要验证“注册—登录—创建分类—记账—报表”的完整用户故事。下面的测试使用 Flask 测试客户端所有请求共享 cookiedeftest_user_story(client):assertclient.post(/api/auth/register,json{email:aexample.com,password:correct-horse}).status_code201assertclient.post(/api/auth/login,json{email:aexample.com,password:correct-horse}).status_code200categoryclient.post(/api/ledger/categories,json{name:餐饮})assertcategory.status_code201entryclient.post(/api/ledger/entries,json{amount:35.50,kind:expense,happened_on:2025-01-02,category_id:category.json[id]})assertentry.status_code201reportclient.get(/api/ledger/report?start2025-01-01end2025-01-31)assertreport.json[expense]35.50这段测试有意不直接访问数据库验证的是 HTTP 契约和会话行为。若失败先看哪个断言失败注册失败检查输入校验分类失败检查login_required报表金额错误则检查 Decimal 和软删除过滤。六、失败场景比成功场景更有价值deftest_cannot_read_another_users_entry(client,app):responseclient.get(f/api/ledger/entries/{app.config[OTHER_ENTRY_ID]})assertresponse.status_code404deftest_duplicate_category_rolls_back(client):client.post(/api/ledger/categories,json{name:餐饮})responseclient.post(/api/ledger/categories,json{name:餐饮})assertresponse.status_code409assertlen(client.get(/api/ledger/categories).json[items])1数据库唯一约束和事务回滚必须通过请求验证不能只在模型测试里“相信它会工作”。建议用pytest.mark.parametrize覆盖空邮箱、短密码、负金额、未来日期等输入减少重复代码。七、上一篇练习讲解与本篇练习上一篇的每日余额曲线应先按日期聚合再在 Python 中按日期累加余额不要为每个日期执行一次 SQL典型 N1。本篇练习增加参数化测试覆盖金额边界、分页最后一页、删除后报表和跨用户访问同时把测试数据 fixture 抽成工厂函数避免复制大量字典。测试命名要描述业务结果例如test_deleted_entry_is_not_counted_in_report而不是test_report_1。每次修复越权或事务 bug都先添加一个能复现它的测试再修改实现这样项目在后续重构、切换数据库或增加缓存时仍有保护网。本篇结束完整模块文件下面是交付项目中真实存在的完整文件 flask-ledger/tests/test_app.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。deftest_health(client):assertclient.get(/api/health).get_json(){status:ok}deftest_register_login_and_me(client):responseclient.post(/api/auth/register,json{email:Aexample.com,password:password123})assertresponse.status_code201assertclient.post(/api/auth/login,json{email:aexample.com,password:password123}).status_code200assertclient.get(/api/auth/me).get_json()[email]aexample.comdeftest_ledger_flow(logged_client):categorylogged_client.post(/api/categories,json{name:餐饮}).get_json()createdlogged_client.post(/api/transactions,json{kind:expense,amount:28.50,category_id:category[id],happened_on:2026-08-06,note:午饭,})assertcreated.status_code201assertcreated.get_json()[amount]28.50itemslogged_client.get(/api/transactions).get_json()[items]assertlen(items)1statslogged_client.get(/api/statistics/monthly?month2026-08).get_json()assertstats[expense]28.50assertlogged_client.delete(f/api/transactions/{items[0][id]}).status_code204deftest_users_cannot_share_categories(client):client.post(/api/auth/register,json{email:firstexample.com,password:password123})client.post(/api/auth/login,json{email:firstexample.com,password:password123})categoryclient.post(/api/categories,json{name:工资}).get_json()client.post(/api/auth/logout)client.post(/api/auth/register,json{email:secondexample.com,password:password123})client.post(/api/auth/login,json{email:secondexample.com,password:password123})responseclient.post(/api/transactions,json{kind:income,amount:100,category_id:category[id]})assertresponse.status_code404