87 lines
2.6 KiB
Python
87 lines
2.6 KiB
Python
import pytest
|
|
from unittest.mock import patch
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
from pydantic import ValidationError
|
|
from kilostar.utils.error import UserNotExistError
|
|
from kilostar.core.postgres_database import database_exception
|
|
|
|
|
|
@database_exception
|
|
async def success_func():
|
|
return "success"
|
|
|
|
|
|
@database_exception
|
|
async def validation_error_func():
|
|
raise ValidationError.from_exception_data(title="Mock", line_errors=[])
|
|
|
|
|
|
@database_exception
|
|
async def integrity_error_func():
|
|
raise IntegrityError("mock_statement", "mock_params", "mock_orig")
|
|
|
|
|
|
@database_exception
|
|
async def operational_error_func():
|
|
raise OperationalError("mock_statement", "mock_params", "mock_orig")
|
|
|
|
|
|
@database_exception
|
|
async def user_not_exist_error_func():
|
|
raise UserNotExistError("mock user")
|
|
|
|
|
|
@database_exception
|
|
async def exception_func():
|
|
raise Exception("mock generic exception")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_func():
|
|
assert await success_func() == "success"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("kilostar.core.database.database_exception.logger")
|
|
async def test_validation_error(mock_logger):
|
|
with pytest.raises(ValidationError):
|
|
await validation_error_func()
|
|
mock_logger.error.assert_called_once()
|
|
assert "对象校验失败" in mock_logger.error.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("kilostar.core.database.database_exception.logger")
|
|
async def test_integrity_error(mock_logger):
|
|
with pytest.raises(IntegrityError):
|
|
await integrity_error_func()
|
|
mock_logger.error.assert_called_once()
|
|
assert "数据库完整性错误" in mock_logger.error.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("kilostar.core.database.database_exception.logger")
|
|
async def test_operational_error(mock_logger):
|
|
with pytest.raises(OperationalError):
|
|
await operational_error_func()
|
|
mock_logger.error.assert_called_once()
|
|
assert "数据库连接异常" in mock_logger.error.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("kilostar.core.database.database_exception.logger")
|
|
async def test_user_not_exist_error(mock_logger):
|
|
result = await user_not_exist_error_func()
|
|
assert result is None
|
|
mock_logger.error.assert_called_once()
|
|
assert "更改密码失败，用户不存在" in mock_logger.error.call_args[0][0]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@patch("kilostar.core.database.database_exception.logger")
|
|
async def test_generic_exception(mock_logger):
|
|
with pytest.raises(Exception, match="mock generic exception"):
|
|
await exception_func()
|
|
mock_logger.exception.assert_called_once()
|
|
assert "未预期的数据库错误" in mock_logger.exception.call_args[0][0]
|