IntelligentRailwayCosting/SourceCode/IntelligentRailwayCosting/app/test/sqlserver_test.py

import unittest
from tools.db_helper.sqlserver_helper import SQLServerHelper
from tools.db_helper.base import Base
from sqlalchemy import Column, Integer, String
from typing import Dict, Any
import time

class TestTable(Base):
    __tablename__ = 'test_table'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

class TestSQLServerHelper(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        """测试类初始化，创建数据库帮助类实例"""
        cls.db_helper = SQLServerHelper()
        cls.database = 'Iwb_RecoData2024'  # 使用配置文件中定义的测试数据库

    def setUp(self):
        """每个测试用例开始前的准备工作"""
        self.start_time = time.time()
        # 创建测试表
        engine = self.db_helper.get_engine(self.database)
        Base.metadata.create_all(engine)

    def tearDown(self):
        """每个测试用例结束后的清理工作"""
        # 删除测试表
        engine = self.db_helper.get_engine(self.database)
        Base.metadata.drop_all(engine)
        test_duration = time.time() - self.start_time
        print(f"\n测试用例耗时: {test_duration:.3f}秒")

    @classmethod
    def tearDownClass(cls):
        """测试类结束时的清理工作"""
        cls.db_helper.dispose_all()

    def test_singleton(self):
        """测试单例模式"""
        helper1 = SQLServerHelper()
        helper2 = SQLServerHelper()
        self.assertIs(helper1, helper2)

    def test_database_connection(self):
        """测试数据库连接"""
        try:
            # 测试获取数据库引擎
            engine = self.db_helper.get_engine(self.database)
            self.assertIsNotNone(engine, "数据库引擎创建失败")

            # 测试会话作用域
            with self.db_helper.session_scope(self.database) as session:
                self.assertIsNotNone(session, "数据库会话创建失败")
                # 测试简单查询
                result = session.execute("SELECT 1").scalar()
                self.assertEqual(result, 1, "数据库连接测试失败")

        except Exception as e:
            self.fail(f"数据库连接测试失败: {str(e)}")

    def test_basic_operations(self):
        """测试基本数据库操作"""
        try:
            # 测试查询操作
            query_result = self.db_helper.execute_query(self.database, 'SELECT @@VERSION')
            self.assertIsNotNone(query_result, "查询操作失败")
            self.assertTrue(len(query_result) > 0, "查询结果为空")

            # 测试标量查询
            scalar_result = self.db_helper.execute_scalar(self.database, 'SELECT DB_NAME()')
            self.assertIsNotNone(scalar_result, "标量查询失败")
            self.assertEqual(scalar_result, self.database, "数据库名称不匹配")

            # 测试非查询操作
            # 创建临时表并插入数据
            self.db_helper.execute_non_query(
                self.database,
                "CREATE TABLE #temp_test (id INT, name NVARCHAR(50))"
            )
            insert_result = self.db_helper.execute_non_query(
                self.database,
                "INSERT INTO #temp_test (id, name) VALUES (:id, :name)",
                {"id": 1, "name": "test"}
            )
            self.assertEqual(insert_result, 1, "插入操作失败")

            # 验证插入结果
            result = self.db_helper.execute_scalar(
                self.database,
                "SELECT name FROM #temp_test WHERE id = 1"
            )
            self.assertEqual(result, "test", "数据验证失败")

        except Exception as e:
            self.fail(f"基本操作测试失败: {str(e)}")

    def test_session_management(self):
        """测试会话管理和事务"""
        try:
            # 测试事务回滚
            with self.assertRaises(Exception):
                with self.db_helper.session_scope(self.database) as session:
                    test_record = TestTable(name="test_rollback")
                    session.add(test_record)
                    raise Exception("触发回滚")

            # 验证回滚成功
            with self.db_helper.session_scope(self.database) as session:
                result = session.query(TestTable).filter_by(name="test_rollback").first()
                self.assertIsNone(result, "事务回滚失败")

            # 测试正常事务提交
            with self.db_helper.session_scope(self.database) as session:
                test_record = TestTable(name="test_commit")
                session.add(test_record)

            # 验证提交成功
            with self.db_helper.session_scope(self.database) as session:
                result = session.query(TestTable).filter_by(name="test_commit").first()
                self.assertIsNotNone(result, "事务提交失败")
                self.assertEqual(result.name, "test_commit")

        except Exception as e:
            self.fail(f"会话管理测试失败: {str(e)}")

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