# ============================================================================ # 家庭阅读激励工具 - 业务层单元测试 # ============================================================================ """ Phase 2 业务逻辑层测试套件。 测试范围: 1. PointsConfig - 配置加载/保存/默认值 2. PointsEngine - 积分计算/打卡积分/奖励/调整/每日上限 3. RedemptionEngine - 兑换资格检查/兑换事务/取消退还/库存扣减 4. ChildContext - 切换孩子/数据隔离/统计概览 运行方式: python 12-test_engine.py 前置条件: 需要先运行种子数据或手动初始化数据库 """ import os import sys import unittest import tempfile import shutil from datetime import date, timedelta # 确保导入路径正确 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) class PointsEngineTest(unittest.TestCase): """积分引擎测试。""" @classmethod def setUpClass(cls): """初始化测试数据库环境。""" # 使用临时数据库，不污染开发数据 cls._tmpdir = tempfile.mkdtemp(prefix="reading_test_") cls._orig_db_dir = os.environ.get("READING_INCENTIVE_DB_DIR", "") os.environ["READING_INCENTIVE_DB_DIR"] = cls._tmpdir # 重新加载模块以使用新路径 import importlib import db_config importlib.reload(db_config) # 初始化 Schema schema_path = os.path.join( os.path.dirname(os.path.abspath(__file__)), "01-schema.sql" ) conn = db_config.get_connection() with open(schema_path, "r", encoding="utf-8") as f: conn.executescript(f.read()) conn.commit() # 导入 DAL 和引擎（在环境变量设置之后） from dal import ChildrenDAO, BooksDAO, ReadingRecordsDAO, RewardsDAO cls.ChildrenDAO = ChildrenDAO cls.BooksDAO = BooksDAO cls.ReadingRecordsDAO = ReadingRecordsDAO cls.RewardsDAO = RewardsDAO from points_config import PointsConfig from points_engine import PointsEngine, PointsBreakdown from redemption_engine import RedemptionEngine, EligibilityResult, RedemptionResult from child_context import ChildContext, ChildStats, FamilyStats cls.PointsConfig = PointsConfig cls.PointsEngine = PointsEngine cls.PointsBreakdown = PointsBreakdown cls.RedemptionEngine = RedemptionEngine cls.EligibilityResult = EligibilityResult cls.RedemptionResult = RedemptionResult cls.ChildContext = ChildContext cls.ChildStats = ChildStats cls.FamilyStats = FamilyStats @classmethod def tearDownClass(cls): """清理临时数据库。""" # 恢复环境变量 if cls._orig_db_dir: os.environ["READING_INCENTIVE_DB_DIR"] = cls._orig_db_dir else: os.environ.pop("READING_INCENTIVE_DB_DIR", None) shutil.rmtree(cls._tmpdir, ignore_errors=True) def setUp(self): """每个测试前准备：创建基础数据。""" self.xiaoming = self.ChildrenDAO.create(name="小明", age=7) self.xiaohong = self.ChildrenDAO.create(name="小红", age=9) self.book1 = self.BooksDAO.create(title="好饿的毛毛虫", total_pages=32) self.book2 = self.BooksDAO.create(title="小王子", total_pages=120) self.book3 = self.BooksDAO.create(title="夏洛的网", total_pages=176) self.config = self.PointsConfig.load() self.engine = self.PointsEngine self.redemption = self.RedemptionEngine # ------------------------------------------------------------------ # PointsConfig 测试 # ------------------------------------------------------------------ def test_config_load_defaults(self): """配置加载：默认值正确。""" config = self.PointsConfig.load() self.assertEqual(config.base_points, 10) self.assertTrue(config.streak_enabled) self.assertEqual(config.duration_unit_minutes, 10) self.assertEqual(config.max_daily_points, 50) def test_config_save_and_reload(self): """配置保存：保存后重载值一致。""" config = self.PointsConfig.load() config.base_points = 15 config.streak_enabled = False config.save() reloaded = self.PointsConfig.load() self.assertEqual(reloaded.base_points, 15) self.assertFalse(reloaded.streak_enabled) def test_config_to_dict(self): """配置导出：to_dict 返回完整字段。""" d = self.config.to_dict() self.assertIn("base_points", d) self.assertIn("streak_enabled", d) self.assertIn("max_daily_points", d) self.assertEqual(d["base_points"], 10) # ------------------------------------------------------------------ # PointsEngine.calculate_expected_points 测试 # ------------------------------------------------------------------ def test_calculate_basic(self): """积分计算：仅基础分。""" result = self.engine.calculate_expected_points( duration_minutes=0, pages_read=0, streak_days=0, config=self.config ) self.assertEqual(result.base, 10) self.assertEqual(result.total, 10) self.assertFalse(result.capped) def test_calculate_with_duration(self): """积分计算：含时长加成。""" result = self.engine.calculate_expected_points( duration_minutes=30, pages_read=0, streak_days=0, config=self.config ) # 30分钟 / 10分钟每单位 * 2分 = 6分时长加成 self.assertEqual(result.duration_bonus, 6) self.assertEqual(result.total, 16) def test_calculate_duration_cap(self): """积分计算：时长加成不超过上限。""" result = self.engine.calculate_expected_points( duration_minutes=120, pages_read=0, streak_days=0, config=self.config ) # 120分钟 → 12单位 * 2 = 24，但上限 20 self.assertEqual(result.duration_bonus, 20) def test_calculate_with_pages(self): """积分计算：含页数加成。""" result = self.engine.calculate_expected_points( duration_minutes=0, pages_read=60, streak_days=0, config=self.config ) # 60页 / 20页每分 = 3分 self.assertEqual(result.pages_bonus, 3) self.assertEqual(result.total, 13) def test_calculate_streak_7(self): """积分计算：连续7天加成。""" # streak_days=6 表示打卡前连续6天，本次是第7天 result = self.engine.calculate_expected_points( duration_minutes=0, pages_read=0, streak_days=6, config=self.config ) self.assertEqual(result.streak_bonus, 5) self.assertEqual(result.streak_days, 7) def test_calculate_streak_30(self): """积分计算：连续30天加成。""" result = self.engine.calculate_expected_points( duration_minutes=0, pages_read=0, streak_days=29, config=self.config ) self.assertEqual(result.streak_bonus, 20) self.assertEqual(result.streak_days, 30) def test_calculate_daily_cap(self): """积分计算：超过每日上限被截断。""" result = self.engine.calculate_expected_points( duration_minutes=120, pages_read=200, streak_days=29, config=self.config ) # base=10 + streak=20 + duration=20(capped) + pages=10(capped) = 60 # 但 max_daily=50，最终 capped=True self.assertTrue(result.capped) self.assertEqual(result.total, 50) def test_calculate_zero_duration(self): """积分计算：0分钟只拿基础分。""" result = self.engine.calculate_expected_points( duration_minutes=0, pages_read=0, streak_days=0, config=self.config ) self.assertEqual(result.duration_bonus, 0) self.assertEqual(result.pages_bonus, 0) # ------------------------------------------------------------------ # PointsEngine.add_reading_points 测试 # ------------------------------------------------------------------ def test_add_reading_points_basic(self): """打卡积分：第一次打卡获得基础分。""" record = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=20, pages_read=10, ) result = self.engine.add_reading_points(self.xiaoming.id, record.id, self.config) self.assertGreater(result.total, 0) self.assertEqual(result.base, 10) # 验证余额已更新 balance = self.engine.get_balance(self.xiaoming.id) self.assertEqual(balance, result.total) def test_add_reading_points_updates_balance(self): """打卡积分：两次打卡后余额累计正确。""" record1 = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=20, pages_read=10, ) result1 = self.engine.add_reading_points(self.xiaoming.id, record1.id, self.config) # 换一本书再次打卡（同一天允许不同书） record2 = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book2.id, read_date=date.today().isoformat(), duration_minutes=15, pages_read=5, ) result2 = self.engine.add_reading_points(self.xiaoming.id, record2.id, self.config) balance = self.engine.get_balance(self.xiaoming.id) self.assertEqual(balance, result1.total + result2.total) def test_add_reading_points_daily_cap(self): """打卡积分：超过每日上限后不再发放。""" # 使用已调整的配置突破上限更可控 self.config.max_daily_points = 10 self.config.base_points = 10 self.config.duration_enabled = False self.config.pages_enabled = False record1 = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=30, ) result1 = self.engine.add_reading_points(self.xiaoming.id, record1.id, self.config) self.assertEqual(result1.total, 10) # 换书再打卡 record2 = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book2.id, read_date=date.today().isoformat(), duration_minutes=30, ) result2 = self.engine.add_reading_points(self.xiaoming.id, record2.id, self.config) self.assertEqual(result2.total, 0) self.assertTrue(result2.capped) def test_add_reading_points_invalid_record(self): """打卡积分：记录不存在抛出异常。""" with self.assertRaises(ValueError): self.engine.add_reading_points(self.xiaoming.id, 99999, self.config) def test_add_reading_points_wrong_child(self): """打卡积分：记录归属不匹配抛出异常。""" record = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=20, ) with self.assertRaises(ValueError): self.engine.add_reading_points(self.xiaohong.id, record.id, self.config) # ------------------------------------------------------------------ # PointsEngine.add_bonus_points 测试 # ------------------------------------------------------------------ def test_bonus_points_positive(self): """奖励积分：正向奖励。""" new_balance = self.engine.add_bonus_points(self.xiaoming.id, 50, "表现好") self.assertEqual(new_balance, 50) def test_bonus_points_invalid_amount(self): """奖励积分：负值/零值抛出异常。""" with self.assertRaises(ValueError): self.engine.add_bonus_points(self.xiaoming.id, -10) # ------------------------------------------------------------------ # PointsEngine.adjust_points 测试 # ------------------------------------------------------------------ def test_adjust_points_positive(self): """调整积分：正向调整。""" self.engine.add_bonus_points(self.xiaoming.id, 30) new_balance = self.engine.adjust_points(self.xiaoming.id, 20, "补发") self.assertEqual(new_balance, 50) def test_adjust_points_negative_below_zero(self): """调整积分：扣减不允许为负。""" new_balance = self.engine.adjust_points(self.xiaoming.id, -100, "扣罚") self.assertEqual(new_balance, 0) # ------------------------------------------------------------------ # RedemptionEngine 测试 # ------------------------------------------------------------------ def _setup_rewards(self): """创建测试奖品。""" self.reward_cheap = self.RewardsDAO.create( name="多看15分钟动画片", points_required=20, stock=-1 ) self.reward_expensive = self.RewardsDAO.create( name="去游乐场", points_required=100, stock=-1 ) self.reward_limited = self.RewardsDAO.create( name="挑选一本新书", points_required=30, stock=2 ) def test_check_eligibility_insufficient_points(self): """兑换资格：积分不足。""" self._setup_rewards() result = self.redemption.check_eligibility( self.xiaoming.id, self.reward_expensive.id ) self.assertFalse(result.eligible) self.assertIn("积分不足", result.reason) def test_check_eligibility_sufficient_points(self): """兑换资格：积分充足。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 50) result = self.redemption.check_eligibility( self.xiaoming.id, self.reward_cheap.id ) self.assertTrue(result.eligible) def test_check_eligibility_inactive_reward(self): """兑换资格：奖品已下架。""" self._setup_rewards() # 先给予足够积分 self.engine.add_bonus_points(self.xiaoming.id, 50) # 下架奖品 from dal import RewardsDAO RewardsDAO.update(self.reward_cheap.id, is_active=0) result = self.redemption.check_eligibility( self.xiaoming.id, self.reward_cheap.id ) self.assertFalse(result.eligible) def test_redeem_success(self): """兑换操作：成功兑换。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 50) result = self.redemption.redeem( self.xiaoming.id, self.reward_cheap.id, notes="想要红色" ) self.assertTrue(result.success) self.assertIsNotNone(result.redemption_record) self.assertEqual(result.new_balance, 30) # 50 - 20 # 验证积分已扣减 balance = self.engine.get_balance(self.xiaoming.id) self.assertEqual(balance, 30) def test_redeem_insufficient_points(self): """兑换操作：积分不足返回失败。""" self._setup_rewards() result = self.redemption.redeem(self.xiaoming.id, self.reward_cheap.id) self.assertFalse(result.success) self.assertIn("积分不足", result.error) def test_redeem_limited_stock(self): """兑换操作：限量库存正常扣减。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 100) # 兑换第1次 result1 = self.redemption.redeem( self.xiaoming.id, self.reward_limited.id ) self.assertTrue(result1.success) # 兑换第2次 result2 = self.redemption.redeem( self.xiaoming.id, self.reward_limited.id ) self.assertTrue(result2.success) # 兑换第3次（库存已耗尽） self.engine.add_bonus_points(self.xiaoming.id, 100) # 用小红来兑换剩余库存（小明已用完） from dal import RewardsDAO available = RewardsDAO.get_available_stock(self.reward_limited.id) # 两个pending占用了库存 self.assertEqual(available, 0) def test_cancel_redemption_refunds_points(self): """取消兑换：退还积分。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 50) redeem_result = self.redemption.redeem( self.xiaoming.id, self.reward_cheap.id ) self.assertTrue(redeem_result.success) # 取消兑换 cancel_result = self.redemption.cancel_redemption( redeem_result.redemption_record.id ) self.assertTrue(cancel_result.success) self.assertEqual(cancel_result.new_balance, 50) # 验证积分已退还 balance = self.engine.get_balance(self.xiaoming.id) self.assertEqual(balance, 50) def test_fulfill_redemption(self): """确认发放：pending → fulfilled。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 50) redeem_result = self.redemption.redeem( self.xiaoming.id, self.reward_cheap.id ) fulfill_result = self.redemption.fulfill(redeem_result.redemption_record.id) self.assertTrue(fulfill_result.success) self.assertEqual(fulfill_result.redemption_record.status, "fulfilled") # ------------------------------------------------------------------ # ChildContext 测试 # ------------------------------------------------------------------ def test_switch_to_valid_child(self): """上下文切换：切换到有效的孩子。""" ctx = self.ChildContext() child = ctx.switch_to(self.xiaoming.id) self.assertEqual(child.id, self.xiaoming.id) self.assertEqual(ctx.current_child_id, self.xiaoming.id) def test_switch_to_invalid_child(self): """上下文切换：切换不存在的孩子抛出异常。""" ctx = self.ChildContext() with self.assertRaises(ValueError): ctx.switch_to(99999) def test_switch_to_archived_child(self): """上下文切换：切换已归档孩子抛出异常。""" self.ChildrenDAO.archive(self.xiaoming.id) ctx = self.ChildContext() with self.assertRaises(ValueError): ctx.switch_to(self.xiaoming.id) def test_switch_to_first_active(self): """上下文切换：自动选择第一个活跃孩子。""" ctx = self.ChildContext() child = ctx.switch_to_first_active() self.assertIsNotNone(child) self.assertTrue(child.is_active) def test_context_without_selection_raises(self): """上下文：未选择孩子时调用查询抛出异常。""" ctx = self.ChildContext() with self.assertRaises(RuntimeError): ctx.get_reading_records() def test_get_stats(self): """统计：ChildStats 返回正确数据。""" # 创建一些阅读记录 record = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=30, pages_read=10, ) self.engine.add_reading_points(self.xiaoming.id, record.id, self.config) stats = self.ChildContext.get_child_stats(self.xiaoming.id) self.assertEqual(stats.child_name, "小明") self.assertEqual(stats.total_readings, 1) self.assertGreater(stats.total_points, 0) self.assertTrue(stats.today_read) self.assertGreaterEqual(stats.this_week_readings, 1) def test_family_stats(self): """统计：FamilyStats 包含所有活跃孩子。""" family = self.ChildContext.get_family_stats() self.assertEqual(family.active_children, 2) self.assertEqual(len(family.children_stats), 2) names = [s.child_name for s in family.children_stats] self.assertIn("小明", names) self.assertIn("小红", names) def test_data_isolation(self): """数据隔离：切换孩子后查询结果隔离。""" # 小明打卡 record_m = self.ReadingRecordsDAO.create( child_id=self.xiaoming.id, book_id=self.book1.id, read_date=date.today().isoformat(), duration_minutes=20, ) self.engine.add_reading_points(self.xiaoming.id, record_m.id, self.config) # 小红打卡 record_h = self.ReadingRecordsDAO.create( child_id=self.xiaohong.id, book_id=self.book2.id, read_date=date.today().isoformat(), duration_minutes=30, ) self.engine.add_reading_points(self.xiaohong.id, record_h.id, self.config) ctx = self.ChildContext() # 切换小明 ctx.switch_to(self.xiaoming.id) xm_records = ctx.get_reading_records() xm_book_ids = [r.book_id for r in xm_records] self.assertIn(self.book1.id, xm_book_ids) self.assertNotIn(self.book2.id, xm_book_ids) # 切换小红 ctx.switch_to(self.xiaohong.id) xh_records = ctx.get_reading_records() xh_book_ids = [r.book_id for r in xh_records] self.assertIn(self.book2.id, xh_book_ids) self.assertNotIn(self.book1.id, xh_book_ids) def test_list_active_children(self): """孩子列表：不包含已归档孩子。""" self.ChildrenDAO.archive(self.xiaoming.id) active = self.ChildContext.list_active_children() names = [c.name for c in active] self.assertNotIn("小明", names) self.assertIn("小红", names) def test_list_all_children_includes_archived(self): """孩子列表：包含已归档孩子。""" self.ChildrenDAO.archive(self.xiaoming.id) all_children = self.ChildContext.list_all_children() names = [c.name for c in all_children] self.assertIn("小明", names) def test_redeem_through_context(self): """集成：通过 ChildContext 获取可兑换奖品。""" self._setup_rewards() self.engine.add_bonus_points(self.xiaoming.id, 50) ctx = self.ChildContext(self.xiaoming.id) rewards = ctx.get_available_rewards() self.assertGreater(len(rewards), 0) # 验证有可兑换的奖品 eligible = [r for r in rewards if r["can_redeem"]] self.assertGreater(len(eligible), 0) if __name__ == "__main__": unittest.main(verbosity=2)