renjianbo
f2e65a8fbb
- Add company module (3-tier org, CEO planning, parallel departments) - Add company orchestrator, knowledge extractor, presets, scheduler - Add company API endpoints, models, and frontend views - Add 今天吃啥 PWA app (69 dishes, real images, offline support) - Add team_projects output directory structure - Add unified manage.ps1 for service lifecycle - Add Windows startup guide v1.0 - Add TTS troubleshooting doc - Update frontend (AgentChat UX overhaul, new views) - Update backend (voice engine fix, multi-tenant, RBAC) - Remove deprecated startup scripts and old docs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
452 lines
17 KiB
Python
452 lines
17 KiB
Python
# ============================================================================
|
|
# 家庭阅读激励工具 - DAL 单元测试
|
|
# ============================================================================
|
|
"""
|
|
验证所有 CRUD 操作、约束、事务。
|
|
|
|
用法:
|
|
python 07-test_dal.py
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import tempfile
|
|
|
|
# 使用临时数据库，不影响开发数据
|
|
os.environ["READING_INCENTIVE_DB_NAME"] = "test_reading_incentive.db"
|
|
os.environ["READING_INCENTIVE_DB_DIR"] = tempfile.mkdtemp()
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
# 重新导入以应用环境变量
|
|
import importlib
|
|
import db_config
|
|
importlib.reload(db_config)
|
|
|
|
from db_config import get_connection, transaction, DB_PATH
|
|
from dal import (
|
|
ChildrenDAO, BooksDAO, ReadingRecordsDAO,
|
|
PointsDAO, RewardsDAO, RedemptionRecordsDAO, PhotosDAO,
|
|
)
|
|
from models import Child, Book, ReadingRecord, PointsRecord, Reward, Photo
|
|
|
|
|
|
class BaseTest(unittest.TestCase):
|
|
"""基类：初始化数据库 Schema。"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
# 加载 schema.sql 并执行
|
|
schema_path = os.path.join(os.path.dirname(__file__), "01-schema.sql")
|
|
with open(schema_path, "r", encoding="utf-8") as f:
|
|
schema_sql = f.read()
|
|
|
|
conn = db_config.get_connection()
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS _schema_version (
|
|
version INTEGER PRIMARY KEY,
|
|
description TEXT NOT NULL,
|
|
applied_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime'))
|
|
)
|
|
""")
|
|
conn.executescript(schema_sql)
|
|
conn.commit()
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
# 清理测试数据库
|
|
if os.path.exists(DB_PATH):
|
|
os.remove(DB_PATH)
|
|
|
|
def setUp(self):
|
|
"""每个测试前清理数据。"""
|
|
with transaction() as conn:
|
|
tables = [
|
|
"photos", "redemption_records", "rewards",
|
|
"points", "reading_records", "books", "children",
|
|
]
|
|
for t in tables:
|
|
conn.execute(f"DELETE FROM {t}")
|
|
# 重置自增 ID
|
|
conn.execute("DELETE FROM sqlite_sequence")
|
|
|
|
|
|
# ============================================================================
|
|
# Children Tests
|
|
# ============================================================================
|
|
class TestChildrenCRUD(BaseTest):
|
|
"""孩子账户 CRUD 测试。"""
|
|
|
|
def test_create_child(self):
|
|
child = ChildrenDAO.create("小明", age=7)
|
|
self.assertIsNotNone(child.id)
|
|
self.assertEqual(child.name, "小明")
|
|
self.assertEqual(child.age, 7)
|
|
self.assertEqual(child.total_points, 0)
|
|
self.assertTrue(child.is_active)
|
|
|
|
def test_create_empty_name_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
ChildrenDAO.create("")
|
|
|
|
def test_get_by_id(self):
|
|
created = ChildrenDAO.create("小红")
|
|
fetched = ChildrenDAO.get_by_id(created.id)
|
|
self.assertEqual(fetched.name, "小红")
|
|
|
|
def test_get_nonexistent(self):
|
|
self.assertIsNone(ChildrenDAO.get_by_id(99999))
|
|
|
|
def test_list_all_active(self):
|
|
ChildrenDAO.create("孩子A")
|
|
ChildrenDAO.create("孩子B")
|
|
archived = ChildrenDAO.create("孩子C")
|
|
ChildrenDAO.archive(archived.id)
|
|
|
|
active = ChildrenDAO.list_all()
|
|
self.assertEqual(len(active), 2)
|
|
|
|
all_kids = ChildrenDAO.list_all(include_inactive=True)
|
|
self.assertEqual(len(all_kids), 3)
|
|
|
|
def test_update(self):
|
|
child = ChildrenDAO.create("测试")
|
|
updated = ChildrenDAO.update(child.id, name="改名", age=10)
|
|
self.assertEqual(updated.name, "改名")
|
|
self.assertEqual(updated.age, 10)
|
|
|
|
def test_delete_cascade(self):
|
|
"""删除孩子时应级联删除关联数据。"""
|
|
child = ChildrenDAO.create("待删除")
|
|
book = BooksDAO.create("测试书")
|
|
record = ReadingRecordsDAO.create(child.id, book.id, "2025-01-01", 30)
|
|
PointsDAO.create(child.id, "reading", 10, 10, record.id)
|
|
PhotosDAO.create(child.id, "/tmp/test.jpg", record.id)
|
|
|
|
# 确认关联数据存在
|
|
self.assertIsNotNone(ReadingRecordsDAO.get_by_id(record.id))
|
|
|
|
# 删除孩子
|
|
result = ChildrenDAO.delete(child.id)
|
|
self.assertTrue(result)
|
|
self.assertIsNone(ChildrenDAO.get_by_id(child.id))
|
|
# 级联检查
|
|
self.assertIsNone(ReadingRecordsDAO.get_by_id(record.id))
|
|
|
|
def test_get_points(self):
|
|
child = ChildrenDAO.create("积分测试")
|
|
self.assertEqual(ChildrenDAO.get_points(child.id), 0)
|
|
ChildrenDAO.update(child.id, total_points=100)
|
|
self.assertEqual(ChildrenDAO.get_points(child.id), 100)
|
|
|
|
|
|
# ============================================================================
|
|
# Books Tests
|
|
# ============================================================================
|
|
class TestBooksCRUD(BaseTest):
|
|
"""书籍 CRUD 测试。"""
|
|
|
|
def test_create_book(self):
|
|
book = BooksDAO.create("好饿的毛毛虫", author="艾瑞·卡尔", total_pages=32)
|
|
self.assertIsNotNone(book.id)
|
|
self.assertEqual(book.title, "好饿的毛毛虫")
|
|
self.assertEqual(book.author, "艾瑞·卡尔")
|
|
|
|
def test_create_empty_title_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
BooksDAO.create("")
|
|
|
|
def test_get_by_isbn(self):
|
|
BooksDAO.create("书A", isbn="978-7-123-45678-9")
|
|
book = BooksDAO.get_by_isbn("978-7-123-45678-9")
|
|
self.assertIsNotNone(book)
|
|
self.assertEqual(book.title, "书A")
|
|
|
|
def test_search(self):
|
|
BooksDAO.create("哈利·波特与魔法石")
|
|
BooksDAO.create("哈利·波特与密室")
|
|
BooksDAO.create("小王子")
|
|
results = BooksDAO.search("哈利")
|
|
self.assertEqual(len(results), 2)
|
|
|
|
def test_delete_book_sets_null(self):
|
|
"""删除书籍时关联阅读记录的 book_id 应被 SET NULL。"""
|
|
child = ChildrenDAO.create("测试孩子")
|
|
book = BooksDAO.create("待删除书")
|
|
record = ReadingRecordsDAO.create(child.id, book.id, "2025-01-01", 30)
|
|
|
|
BooksDAO.delete(book.id)
|
|
updated_record = ReadingRecordsDAO.get_by_id(record.id)
|
|
self.assertIsNone(updated_record.book_id)
|
|
|
|
|
|
# ============================================================================
|
|
# ReadingRecords Tests
|
|
# ============================================================================
|
|
class TestReadingRecordsCRUD(BaseTest):
|
|
"""阅读打卡记录 CRUD 测试。"""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.child = ChildrenDAO.create("测试孩子")
|
|
self.book = BooksDAO.create("测试书")
|
|
|
|
def test_create_record(self):
|
|
record = ReadingRecordsDAO.create(
|
|
self.child.id, self.book.id, "2025-06-01", duration_minutes=30, pages_read=10
|
|
)
|
|
self.assertIsNotNone(record.id)
|
|
self.assertEqual(record.duration_minutes, 30)
|
|
self.assertEqual(record.child_name, "测试孩子")
|
|
self.assertEqual(record.book_title, "测试书")
|
|
|
|
def test_duplicate_daily_raises(self):
|
|
"""同一天同一本书不能重复打卡。"""
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", 30)
|
|
with self.assertRaises(Exception): # sqlite3.IntegrityError
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", 20)
|
|
|
|
def test_list_by_child_date_range(self):
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", 30)
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-05", 20)
|
|
|
|
in_range = ReadingRecordsDAO.list_by_child(
|
|
self.child.id, date_from="2025-06-01", date_to="2025-06-03"
|
|
)
|
|
self.assertEqual(len(in_range), 1)
|
|
|
|
all_records = ReadingRecordsDAO.list_by_child(self.child.id)
|
|
self.assertEqual(len(all_records), 2)
|
|
|
|
def test_negative_duration_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", -5)
|
|
|
|
def test_count(self):
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", 30)
|
|
ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-02", 20)
|
|
self.assertEqual(ReadingRecordsDAO.count_by_child(self.child.id), 2)
|
|
|
|
|
|
# ============================================================================
|
|
# Points Tests
|
|
# ============================================================================
|
|
class TestPointsCRUD(BaseTest):
|
|
"""积分记录 CRUD 测试。"""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.child = ChildrenDAO.create("积分测试孩子")
|
|
|
|
def test_create_point_record(self):
|
|
record = PointsDAO.create(
|
|
self.child.id, "reading", 10, 10, reason="测试积分"
|
|
)
|
|
self.assertIsNotNone(record.id)
|
|
self.assertEqual(record.points_change, 10)
|
|
self.assertEqual(record.balance_after, 10)
|
|
|
|
def test_invalid_source_type_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
PointsDAO.create(self.child.id, "invalid_type", 10, 10)
|
|
|
|
def test_get_balance(self):
|
|
PointsDAO.create(self.child.id, "reading", 10, 10)
|
|
PointsDAO.create(self.child.id, "reading", 5, 15)
|
|
PointsDAO.create(self.child.id, "redemption", -8, 7)
|
|
self.assertEqual(PointsDAO.get_balance(self.child.id), 7)
|
|
|
|
def test_get_balance_no_records(self):
|
|
"""无积分记录时应从 children.total_points 兜底。"""
|
|
self.assertEqual(PointsDAO.get_balance(self.child.id), 0)
|
|
ChildrenDAO.update(self.child.id, total_points=50)
|
|
self.assertEqual(PointsDAO.get_balance(self.child.id), 50)
|
|
|
|
def test_list_by_child(self):
|
|
PointsDAO.create(self.child.id, "reading", 10, 10)
|
|
PointsDAO.create(self.child.id, "reading", 5, 15)
|
|
records = PointsDAO.list_by_child(self.child.id)
|
|
self.assertEqual(len(records), 2)
|
|
# 按时间倒序
|
|
self.assertEqual(records[0].balance_after, 15)
|
|
|
|
|
|
# ============================================================================
|
|
# Rewards Tests
|
|
# ============================================================================
|
|
class TestRewardsCRUD(BaseTest):
|
|
"""奖品 CRUD 测试。"""
|
|
|
|
def test_create_reward(self):
|
|
reward = RewardsDAO.create("多看动画片", 20, description="15分钟")
|
|
self.assertIsNotNone(reward.id)
|
|
self.assertEqual(reward.points_required, 20)
|
|
self.assertEqual(reward.stock, -1)
|
|
|
|
def test_create_invalid_points_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
RewardsDAO.create("免费奖品", 0)
|
|
|
|
def test_list_active(self):
|
|
RewardsDAO.create("活动A", 10)
|
|
RewardsDAO.create("活动B", 30)
|
|
inactive = RewardsDAO.create("下线C", 50)
|
|
RewardsDAO.update(inactive.id, is_active=0)
|
|
|
|
active = RewardsDAO.list_active()
|
|
self.assertEqual(len(active), 2)
|
|
|
|
def test_stock_tracking(self):
|
|
child = ChildrenDAO.create("测试")
|
|
reward = RewardsDAO.create("限量奖品", 10, stock=2)
|
|
|
|
self.assertEqual(RewardsDAO.get_available_stock(reward.id), 2)
|
|
|
|
RedemptionRecordsDAO.create(child.id, reward.id, 10)
|
|
self.assertEqual(RewardsDAO.get_available_stock(reward.id), 1)
|
|
|
|
RedemptionRecordsDAO.create(child.id, reward.id, 10)
|
|
self.assertEqual(RewardsDAO.get_available_stock(reward.id), 0)
|
|
|
|
|
|
# ============================================================================
|
|
# RedemptionRecords Tests
|
|
# ============================================================================
|
|
class TestRedemptionRecordsCRUD(BaseTest):
|
|
"""兑换记录 CRUD 测试。"""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.child = ChildrenDAO.create("兑换孩子")
|
|
self.reward = RewardsDAO.create("测试奖品", 30)
|
|
|
|
def test_create_redemption(self):
|
|
record = RedemptionRecordsDAO.create(self.child.id, self.reward.id, 30)
|
|
self.assertIsNotNone(record.id)
|
|
self.assertEqual(record.status, "pending")
|
|
self.assertEqual(record.child_name, "兑换孩子")
|
|
self.assertEqual(record.reward_name, "测试奖品")
|
|
|
|
def test_fulfill(self):
|
|
record = RedemptionRecordsDAO.create(self.child.id, self.reward.id, 30)
|
|
fulfilled = RedemptionRecordsDAO.fulfill(record.id)
|
|
self.assertEqual(fulfilled.status, "fulfilled")
|
|
self.assertIsNotNone(fulfilled.fulfilled_at)
|
|
|
|
def test_cancel(self):
|
|
record = RedemptionRecordsDAO.create(self.child.id, self.reward.id, 30)
|
|
cancelled = RedemptionRecordsDAO.cancel(record.id)
|
|
self.assertEqual(cancelled.status, "cancelled")
|
|
|
|
def test_list_pending(self):
|
|
RedemptionRecordsDAO.create(self.child.id, self.reward.id, 30)
|
|
record2 = RedemptionRecordsDAO.create(self.child.id, self.reward.id, 30)
|
|
RedemptionRecordsDAO.fulfill(record2.id)
|
|
|
|
pending = RedemptionRecordsDAO.list_pending()
|
|
self.assertEqual(len(pending), 1)
|
|
|
|
|
|
# ============================================================================
|
|
# Photos Tests
|
|
# ============================================================================
|
|
class TestPhotosCRUD(BaseTest):
|
|
"""照片 CRUD 测试。"""
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.child = ChildrenDAO.create("照片孩子")
|
|
self.book = BooksDAO.create("测试书")
|
|
self.record = ReadingRecordsDAO.create(self.child.id, self.book.id, "2025-06-01", 30)
|
|
|
|
def test_create_photo(self):
|
|
photo = PhotosDAO.create(
|
|
self.child.id, "/photos/test.jpg", self.record.id,
|
|
thumbnail_path="/photos/test_thumb.jpg",
|
|
file_size=102400, width=1920, height=1080,
|
|
)
|
|
self.assertIsNotNone(photo.id)
|
|
self.assertEqual(photo.file_path, "/photos/test.jpg")
|
|
self.assertEqual(photo.width, 1920)
|
|
|
|
def test_list_by_child(self):
|
|
PhotosDAO.create(self.child.id, "/photos/1.jpg")
|
|
PhotosDAO.create(self.child.id, "/photos/2.jpg")
|
|
photos = PhotosDAO.list_by_child(self.child.id)
|
|
self.assertEqual(len(photos), 2)
|
|
|
|
def test_list_by_reading_record(self):
|
|
PhotosDAO.create(self.child.id, "/photos/1.jpg", self.record.id)
|
|
PhotosDAO.create(self.child.id, "/photos/2.jpg", self.record.id)
|
|
photos = PhotosDAO.list_by_reading_record(self.record.id)
|
|
self.assertEqual(len(photos), 2)
|
|
|
|
def test_delete_photo(self):
|
|
photo = PhotosDAO.create(self.child.id, "/photos/to_delete.jpg")
|
|
result = PhotosDAO.delete(photo.id)
|
|
self.assertTrue(result)
|
|
self.assertIsNone(PhotosDAO.get_by_id(photo.id))
|
|
|
|
|
|
# ============================================================================
|
|
# Transaction Tests
|
|
# ============================================================================
|
|
class TestTransactions(BaseTest):
|
|
"""事务回滚测试。"""
|
|
|
|
def test_rollback_on_error(self):
|
|
"""事务中抛出异常时应自动回滚。"""
|
|
child = ChildrenDAO.create("事务测试")
|
|
|
|
try:
|
|
with transaction() as conn:
|
|
conn.execute(
|
|
"INSERT INTO reading_records (child_id, book_id, read_date) VALUES (?, ?, ?)",
|
|
(child.id, None, "2025-06-01"),
|
|
)
|
|
# 故意触发错误（违反 CHECK 约束）
|
|
conn.execute(
|
|
"INSERT INTO points (child_id, source_type, points_change, balance_after) VALUES (?, 'invalid', 10, 10)",
|
|
(child.id,),
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# 事务回滚后，阅读记录不应存在
|
|
records = ReadingRecordsDAO.list_by_child(child.id)
|
|
self.assertEqual(len(records), 0)
|
|
|
|
|
|
# ============================================================================
|
|
# Edge Case Tests
|
|
# ============================================================================
|
|
class TestEdgeCases(BaseTest):
|
|
"""边界条件测试。"""
|
|
|
|
def test_delete_nonexistent(self):
|
|
self.assertFalse(ChildrenDAO.delete(99999))
|
|
self.assertFalse(BooksDAO.delete(99999))
|
|
|
|
def test_orphan_reading_record(self):
|
|
"""book_id 为 NULL 的阅读记录。"""
|
|
child = ChildrenDAO.create("孤儿记录测试")
|
|
record = ReadingRecordsDAO.create(child.id, None, "2025-06-01", 30)
|
|
self.assertIsNotNone(record.id)
|
|
self.assertIsNone(record.book_id)
|
|
|
|
def test_pagination(self):
|
|
child = ChildrenDAO.create("分页测试")
|
|
for i in range(5):
|
|
book = BooksDAO.create(f"书{i}")
|
|
ReadingRecordsDAO.create(child.id, book.id, f"2025-01-0{i+1}", 30)
|
|
|
|
page1 = ReadingRecordsDAO.list_by_child(child.id, limit=2, offset=0)
|
|
page2 = ReadingRecordsDAO.list_by_child(child.id, limit=2, offset=2)
|
|
self.assertEqual(len(page1), 2)
|
|
self.assertEqual(len(page2), 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|