KiloStar/tests/core/database/module/user_test.py

import pytest
from unittest.mock import MagicMock, AsyncMock, patch


@pytest.fixture(autouse=True)
def mock_dependencies():
    with patch("kilostar.core.database.module.user.User") as mock_user_cls:
        mock_user_cls.user_name = MagicMock()
        with patch("kilostar.core.database.module.user.select") as mock_select:
            yield mock_user_cls, mock_select


@pytest.fixture
def mock_session_maker():
    maker = MagicMock()
    session = AsyncMock()
    session.add = MagicMock()
    session.delete = MagicMock()
    maker.return_value.__aenter__.return_value = session
    maker.__aenter__.return_value = session
    maker.__aexit__ = AsyncMock()
    return maker, session


@pytest.mark.asyncio
async def test_add_user(mock_session_maker, mock_dependencies):
    mock_user_cls, _ = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_user = MagicMock()
    mock_user.user_name = "testuser"
    mock_user.hashed_password = "hashedpw"
    mock_user_cls.return_value = mock_user

    mock_exec_result = MagicMock()
    mock_exec_result.first.return_value = None
    session.execute = AsyncMock(return_value=mock_exec_result)

    user = await db.add_user("testuser", "hashedpw")

    assert user.user_name == "testuser"
    assert user.hashed_password == "hashedpw"
    session.add.assert_called_once_with(mock_user)
    session.commit.assert_called_once()
    session.refresh.assert_called_once_with(mock_user)


@pytest.mark.asyncio
async def test_change_password_success(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_statement = MagicMock()
    mock_select.return_value.where.return_value = mock_statement

    from kilostar.utils.access import Accessor

    mock_user = MagicMock()
    mock_user.hashed_password = Accessor.hash_password("old_password")

    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = mock_user
    session.execute = AsyncMock(return_value=mock_exec_result)

    user = await db.change_password("testuser", "old_password", "new_password")

    session.execute.assert_called_once_with(mock_statement)
    assert user.hashed_password == "new_password"
    session.add.assert_called_once_with(mock_user)
    session.commit.assert_called_once()
    session.refresh.assert_called_once_with(mock_user)


@pytest.mark.asyncio
async def test_change_password_user_not_exist(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = None
    session.execute = AsyncMock(return_value=mock_exec_result)

    result = await db.change_password("testuser", "old_password", "new_password")
    assert result is None


@pytest.mark.asyncio
async def test_change_password_wrong_password(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    from kilostar.utils.access import Accessor

    mock_user = MagicMock()
    mock_user.hashed_password = Accessor.hash_password("actual_password")
    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = mock_user
    session.execute = AsyncMock(return_value=mock_exec_result)

    from kilostar.utils.error import UserPasswordError

    with pytest.raises(UserPasswordError):
        await db.change_password("testuser", "old_password", "new_password")


@pytest.mark.asyncio
async def test_delete_user_success(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_statement = MagicMock()
    mock_select.return_value.where.return_value = mock_statement

    mock_user = MagicMock()
    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = mock_user
    session.execute = AsyncMock(return_value=mock_exec_result)

    await db.delete_user("testuser")
    session.execute.assert_called_once_with(mock_statement)
    session.delete.assert_called_once_with(mock_user)
    session.commit.assert_called_once()


@pytest.mark.asyncio
async def test_delete_user_not_exist(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = None
    session.execute = AsyncMock(return_value=mock_exec_result)

    result = await db.delete_user("testuser")
    assert result is None


@pytest.mark.asyncio
async def test_login_user_success(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_statement = MagicMock()
    mock_select.return_value.where.return_value = mock_statement

    mock_user = MagicMock()
    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = mock_user
    session.execute = AsyncMock(return_value=mock_exec_result)

    user = await db.login_user("testuser")
    session.execute.assert_called_once_with(mock_statement)
    assert user == mock_user


@pytest.mark.asyncio
async def test_login_user_not_exist(mock_session_maker, mock_dependencies):
    mock_user_cls, mock_select = mock_dependencies
    from kilostar.core.postgres_database.module import AuthDatabase

    maker, session = mock_session_maker
    db = AuthDatabase(maker)

    mock_exec_result = MagicMock()
    mock_exec_result.scalar_one_or_none.return_value = None
    session.execute = AsyncMock(return_value=mock_exec_result)

    result = await db.login_user("testuser")
    assert result is None