test.py

import sqlite3
import os
from datetime import datetime, timedelta

from database import DatabaseConnection
from models import Book, Purchase, Elimination, Borrowing, InventoryRecord
from dao import BookDAO, PurchaseDAO, EliminationDAO, BorrowingDAO, InventoryDAO
from services import BookService, QueryService, StatisticsService


def test_database_initialization():
    """测试数据库初始化"""
    print("=== 测试数据库初始化 ===")
    try:
        db_path = "testdb.db"
        if os.path.exists(db_path):
            os.remove(db_path)
        db = DatabaseConnection('testdb.db')
        db.initialize_database()
        print("✓ 数据库初始化成功")
        return db
    except Exception as e:
        print(f"✗ 数据库初始化失败: {e}")
        import traceback
        traceback.print_exc()
        return None


def test_book_purchase(db):
    """测试图书采购功能"""
    print("\n=== 测试图书采购功能 ===")
    
    # 初始化DAO
    book_dao = BookDAO(db)
    purchase_dao = PurchaseDAO(db)
    elimination_dao = EliminationDAO(db)
    borrowing_dao = BorrowingDAO(db)
    inventory_dao = InventoryDAO(db)
    
    # 初始化服务
    book_service = BookService(db, book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao)
    
    # 测试采购新书
    print("1. 采购新书:")
    book_data = {
        'isbn': '978-7-302-12345-6',
        'title': 'Python编程从入门到实践',
        'author': 'Eric Matthes',
        'publisher': '清华大学出版社',
        'publish_year': 2020,
        'category': '编程',
        'price': 89.90,
        'total_copies': 0,
        'available_copies': 0,
        'status': 'active'
    }
    
    purchase_data = {
        'purchase_date': datetime.now().strftime('%Y-%m-%d'),
        'quantity': 10,
        'unit_price': 79.90,
        'supplier': '清华大学出版社'
    }
    
    try:
        purchase_id = book_service.purchase_book(book_data, purchase_data)
        print(f"✓ 采购成功，采购记录ID: {purchase_id}")
        
        # 验证图书信息
        book = book_dao.get_by_isbn('978-7-302-12345-6')
        if book:
            print(f"✓ 图书信息: {book.title}, 库存: {book.total_copies}本, 可借: {book.available_copies}本")
    except Exception as e:
        print(f"✗ 采购失败: {e}")
        import traceback
        traceback.print_exc()
    
    # 测试采购已存在图书
    print("\n2. 增加已存在图书库存:")
    purchase_data2 = {
        'purchase_date': datetime.now().strftime('%Y-%m-%d'),
        'quantity': 5,
        'unit_price': 75.00,
        'supplier': '当当网'
    }
    
    try:
        purchase_id2 = book_service.purchase_book({'isbn': '978-7-302-12345-6'}, purchase_data2)
        print(f"✓ 补货成功，采购记录ID: {purchase_id2}")
        
        book = book_dao.get_by_isbn('978-7-302-12345-6')
        if book:
            print(f"✓ 更新后库存: 总{book.total_copies}本, 可借{book.available_copies}本")
    except Exception as e:
        print(f"✗ 补货失败: {e}")
        import traceback
        traceback.print_exc()
    
    # 采购另一本书
    print("\n3. 采购另一本图书:")
    book_data2 = {
        'isbn': '978-7-115-54321-0',
        'title': '数据结构与算法',
        'author': '严蔚敏',
        'publisher': '人民邮电出版社',
        'publish_year': 2019,
        'category': '计算机科学',
        'price': 69.80,
        'total_copies': 0,
        'available_copies': 0,
        'status': 'active'
    }
    
    purchase_data3 = {
        'purchase_date': datetime.now().strftime('%Y-%m-%d'),
        'quantity': 8,
        'unit_price': 59.80,
        'supplier': '京东'
    }
    
    try:
        purchase_id3 = book_service.purchase_book(book_data2, purchase_data3)
        print(f"✓ 采购成功，采购记录ID: {purchase_id3}")
    except Exception as e:
        print(f"✗ 采购失败: {e}")
        import traceback
        traceback.print_exc()
    
    return book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao, book_service


def test_book_elimination(db, book_dao, book_service):
    """测试图书淘汰功能"""
    print("\n=== 测试图书淘汰功能 ===")
    
    # 获取图书列表
    books = book_dao.get_all()
    if not books:
        print("✗ 没有找到图书")
        return
    
    book_id = books[0].book_id
    book_title = books[0].title
    
    print(f"1. 淘汰图书 '{book_title}':")
    
    try:
        elimination_id = book_service.eliminate_book(book_id, quantity=2, reason='损坏')
        print(f"✓ 淘汰成功，淘汰记录ID: {elimination_id}")
        
        # 检查库存
        updated_book = book_dao.get_by_id(book_id)
        if updated_book:
            print(f"✓ 淘汰后库存: 总{updated_book.total_copies}本, 可借{updated_book.available_copies}本")
    except Exception as e:
        print(f"✗ 淘汰失败: {e}")
        import traceback
        traceback.print_exc()


def test_book_borrowing_and_return(db, book_dao, book_service):
    """测试图书借阅和归还功能"""
    print("\n=== 测试图书借阅和归还功能 ===")
    
    # 获取可借阅的图书
    books = book_dao.get_all()
    if not books:
        print("✗ 没有找到可借阅的图书")
        return
    
    # 选择第一本有库存的书
    available_book = None
    for book in books:
        if book.available_copies > 0:
            available_book = book
            break
    
    if not available_book:
        print("✗ 没有找到有库存的图书")
        return
    
    book_id = available_book.book_id
    book_title = available_book.title
    
    print(f"1. 借阅图书 '{book_title}':")
    
    try:
        # 借阅图书
        borrower_info = {
            'borrower_id': '20230001',
            'borrower_name': '张三'
        }
        borrowing_id = book_service.borrow_book(book_id, borrower_info)
        print(f"✓ 借阅成功，借阅记录ID: {borrowing_id}")
        
        # 检查库存
        updated_book = book_dao.get_by_id(book_id)
        if updated_book:
            print(f"✓ 借阅后库存: 可借{updated_book.available_copies}本")
        
        # 测试归还
        print(f"\n2. 归还图书:")
        late_fee = book_service.return_book(borrowing_id)
        print(f"✓ 归还成功")
        
        if late_fee > 0:
            print(f"✓ 逾期费用: {late_fee}元")
        
        # 再次检查库存
        updated_book2 = book_dao.get_by_id(book_id)
        if updated_book2:
            print(f"✓ 归还后库存: 可借{updated_book2.available_copies}本")
        
    except Exception as e:
        print(f"✗ 借阅/归还失败: {e}")
        import traceback
        traceback.print_exc()


def test_query_functions(db, book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao):
    """测试查询功能"""
    print("\n=== 测试查询功能 ===")
    
    # 初始化查询服务
    query_service = QueryService(book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao)
    
    # 1. 查询所有图书
    print("1. 查询所有图书:")
    books = book_dao.get_all()
    print(f"✓ 找到{len(books)}本图书:")
    for i, book in enumerate(books[:3], 1):  # 只显示前3本
        print(f"  {i}. {book.title} - {book.author} (库存: {book.available_copies}/{book.total_copies})")
    if len(books) > 3:
        print(f"  ... 还有{len(books)-3}本图书")
    
    # 2. 搜索图书
    print("\n2. 搜索图书(关键词: 'Python'):")
    results = query_service.search_books(keyword='Python')
    print(f"✓ 找到{len(results)}本相关图书:")
    for book in results:
        print(f"  - {book.title} ({book.author})")
    
    # 3. 查询图书详情
    print("\n3. 查询图书详情:")
    if books:
        # 使用第一个书
        book = books[0]
        # 获取相关记录
        purchases = purchase_dao.get_by_book(book.book_id)
        eliminations = elimination_dao.get_by_book(book.book_id)
        borrowings = borrowing_dao.get_active_by_book(book.book_id)
        inventory_history = inventory_dao.get_by_book(book.book_id, limit=5)
        
        print(f"✓ 图书: {book.title}")
        print(f"  采购记录: {len(purchases)}条")
        print(f"  淘汰记录: {len(eliminations)}条")
        print(f"  活跃借阅: {len(borrowings)}条")
        print(f"  最近库存变更: {len(inventory_history)}条")
    
    # 4. 查询库存状态
    print("\n4. 查询库存状态:")
    inventory_status = book_dao.get_inventory_status()
    if inventory_status:
        print("✓ 按类别库存统计:")
        for stat in inventory_status:
            print(f"  {stat['category'] or '未分类'}: {stat['total_books']}种, 共{stat['total_copies']}本, 可借{stat['available_copies']}本")
    
    # 5. 查询低库存图书
    print("\n5. 查询低库存图书:")
    low_stock = book_dao.get_low_stock_books(threshold=15)  # 设置较高的阈值以确保有结果显示
    print(f"✓ 低库存图书: {len(low_stock)}本")
    for book in low_stock:
        print(f"  - {book.title}: 仅剩{book.available_copies}本")
    
    # 6. 查询借阅记录
    print("\n6. 查询借阅记录:")
    borrowings = borrowing_dao.get_by_borrower('20230001')
    print(f"✓ 借阅记录: {len(borrowings)}条")
    
    # 7. 查询逾期图书
    print("\n7. 查询逾期图书:")
    overdue_books = borrowing_dao.get_overdue()
    print(f"✓ 逾期未还图书: {len(overdue_books)}本")
    
    # 8. 查询库存历史
    print("\n8. 查询库存历史:")
    if books:
        inventory_history = inventory_dao.get_by_book(books[0].book_id, limit=5)
        print(f"✓ 图书库存历史记录: {len(inventory_history)}条")


def test_statistics_functions(db, book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao):
    """测试统计功能"""
    print("\n=== 测试统计功能 ===")
    
    # 初始化统计服务
    stats_service = StatisticsService(book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao)
    
    # 1. 库存统计
    print("1. 库存统计:")
    inventory_stats = stats_service.get_inventory_statistics()
    print(f"✓ 总图书种类: {inventory_stats['total_books']}种")
    print(f"✓ 总藏书量: {inventory_stats['total_copies']}本")
    print(f"✓ 可借数量: {inventory_stats['available_copies']}本")
    print(f"✓ 借出数量: {inventory_stats['borrowed_copies']}本")
    print(f"✓ 利用率: {inventory_stats['utilization_rate']}%")
    
    if inventory_stats['category_stats']:
        print("✓ 按类别统计:")
        for category, stats in inventory_stats['category_stats'].items():
            print(f"  {category}: {stats['book_count']}种, {stats['total_copies']}本")
    
    # 2. 图书馆总览
    print("\n2. 图书馆总览统计:")
    library_summary = stats_service.get_library_summary()
    inventory_summary = library_summary['inventory_summary']
    print(f"✓ 藏书情况: {inventory_summary['total_books']}种图书，"
          f"{inventory_summary['total_copies']}本总量，"
          f"{inventory_summary['available_copies']}本可借")
    
    circulation_summary = library_summary['circulation_summary']
    print(f"✓ 流通情况: {circulation_summary['active_borrowings']}本借出，"
          f"{circulation_summary['overdue_count']}本逾期，"
          f"利用率{circulation_summary['utilization_rate']}%")
    
    # 3. 采购统计
    print("\n3. 采购统计:")
    start_date = (datetime.now() - timedelta(days=365)).strftime('%Y-%m-%d')
    end_date = datetime.now().strftime('%Y-%m-%d')
    
    try:
        purchase_stats = stats_service.get_purchase_statistics(start_date, end_date)
        print(f"✓ 采购统计: {purchase_stats['total_purchases']}次采购, "
              f"{purchase_stats['total_quantity']}本, "
              f"总金额{purchase_stats['total_amount']}元")
    except Exception as e:
        print(f"采购统计出错: {e}")
        # 简化显示
        purchases = purchase_dao.get_purchase_statistics()
        if purchases:
            print(f"✓ 采购统计: 总{purchases[0]['purchase_count']}次, "
                  f"{purchases[0]['total_quantity']}本, "
                  f"{purchases[0]['total_amount']}元")
    
    # 4. 淘汰统计
    print("\n4. 淘汰统计:")
    try:
        elimination_stats = stats_service.get_elimination_statistics(start_date, end_date)
        print(f"✓ 淘汰统计: {elimination_stats['total_eliminations']}次淘汰, "
              f"{elimination_stats['total_quantity']}本")
    except Exception as e:
        print(f"淘汰统计出错: {e}")
        eliminations = elimination_dao.get_elimination_statistics()
        if eliminations:
            print(f"✓ 淘汰统计: 总{len(eliminations)}种原因")


def run_tests():
    """运行所有测试"""
    print("=" * 60)
    print("图书管理系统测试程序")
    print("=" * 60)
    
    # 1. 测试数据库初始化
    db = test_database_initialization()
    if not db:
        print("数据库初始化失败，退出测试")
        return
    
    try:
        # 2. 测试图书采购
        book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao, book_service = test_book_purchase(db)
        
        # 3. 测试图书淘汰
        test_book_elimination(db, book_dao, book_service)
        
        # 4. 测试图书借阅和归还
        test_book_borrowing_and_return(db, book_dao, book_service)
        
        # 5. 测试查询功能
        test_query_functions(db, book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao)
        
        # 6. 测试统计功能
        test_statistics_functions(db, book_dao, purchase_dao, elimination_dao, borrowing_dao, inventory_dao)
        
        print("\n" + "=" * 60)
        print("所有测试完成!")
        print("=" * 60)
        
    except Exception as e:
        print(f"\n测试过程中出现错误: {e}")
        import traceback
        traceback.print_exc()
    
    finally:
        # 关闭数据库连接
        db.close_connection()
        print("数据库连接已关闭")


def create_sample_data():
    """创建示例数据用于演示"""
    print("\n=== 创建示例数据 ===")
    
    # 使用文件数据库
    db = DatabaseConnection('library.db')
    db.initialize_database()
    
    # 初始化DAO
    book_dao = BookDAO(db)
    purchase_dao = PurchaseDAO(db)
    elimination_dao = EliminationDAO(db)
    borrowing_dao = BorrowingDAO(db)
    inventory_dao = InventoryDAO(db)
    
    # 创建一些示例图书
    sample_books = [
        {
            'isbn': '978-7-111-54321-9',
            'title': '深入理解计算机系统',
            'author': 'Randal E. Bryant',
            'publisher': '机械工业出版社',
            'publish_year': 2016,
            'category': '计算机科学',
            'price': 139.00,
            'total_copies': 15,
            'available_copies': 12
        },
        {
            'isbn': '978-7-121-33456-7',
            'title': '算法导论',
            'author': 'Thomas H. Cormen',
            'publisher': '电子工业出版社',
            'publish_year': 2013,
            'category': '算法',
            'price': 128.00,
            'total_copies': 20,
            'available_copies': 18
        },
        {
            'isbn': '978-7-532-71234-5',
            'title': '百年孤独',
            'author': '加西亚·马尔克斯',
            'publisher': '南海出版公司',
            'publish_year': 2011,
            'category': '文学',
            'price': 39.50,
            'total_copies': 25,
            'available_copies': 22
        }
    ]
    
    for book_data in sample_books:
        # 检查图书是否已存在
        existing_book = book_dao.get_by_isbn(book_data['isbn'])
        if not existing_book:
            book = Book(**book_data)
            book_id = book_dao.add(book)
            print(f"✓ 创建示例图书: {book.title}")
    
    db.close_connection()
    print("示例数据创建完成")


def main():
    """主程序"""
    print("请选择操作:")
    print("1. 运行完整测试 (内存数据库)")
    print("2. 创建示例数据 (文件数据库 library.db)")
    print("3. 退出")
    
    choice = input("请输入选择 (1-3): ").strip()
    
    if choice == '1':
        run_tests()
    elif choice == '2':
        create_sample_data()
        print("\n示例数据已创建到 library.db 数据库")
    elif choice == '3':
        print("退出程序")
    else:
        print("无效选择")


if __name__ == "__main__":
    main()