erp-nutzfahrzeuge/backend/tests/test_auth_service.py

"""Direct unit tests for auth_service functions."""

import uuid

import pytest
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.database import Base
from app.models.user import User, UserRole
from app.services.auth_service import (
    authenticate_user,
    create_user,
    deactivate_user,
    generate_token_pair,
    get_user_by_email,
    get_user_by_id,
    hash_password,
    list_users,
    refresh_access_token,
    update_user,
    verify_password,
)
from app.utils.jwt import create_refresh_token

TEST_DATABASE_URL = (
    "postgresql+asyncpg://erp_test_user:testpass@localhost:5432/erp_test"
)


@pytest.fixture
def event_loop():
    import asyncio

    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture
async def engine():
    eng = create_async_engine(TEST_DATABASE_URL, echo=False, pool_pre_ping=True)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await eng.dispose()


@pytest.fixture
async def session_factory(engine):
    return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@pytest.fixture
async def session(session_factory):
    async with session_factory() as s:
        yield s


@pytest.mark.asyncio
async def test_hash_and_verify_password():
    hashed = hash_password("mypassword")
    assert hashed != "mypassword"
    assert verify_password("mypassword", hashed) is True
    assert verify_password("wrongpassword", hashed) is False


@pytest.mark.asyncio
async def test_get_user_by_email_found(session):
    user = User(
        email="findme@test.com",
        password_hash=hash_password("Test12345!"),
        full_name="Find Me",
        role=UserRole.admin,
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)

    found = await get_user_by_email(session, "findme@test.com")
    assert found is not None
    assert found.email == "findme@test.com"


@pytest.mark.asyncio
async def test_get_user_by_email_not_found(session):
    found = await get_user_by_email(session, "nobody@test.com")
    assert found is None


@pytest.mark.asyncio
async def test_get_user_by_id_found(session):
    user = User(
        email="byid@test.com",
        password_hash=hash_password("Test12345!"),
        full_name="By ID",
        role=UserRole.verkaeufer,
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)

    found = await get_user_by_id(session, user.id)
    assert found is not None
    assert found.email == "byid@test.com"


@pytest.mark.asyncio
async def test_get_user_by_id_not_found(session):
    found = await get_user_by_id(session, uuid.uuid4())
    assert found is None


@pytest.mark.asyncio
async def test_authenticate_user_valid(session):
    user = User(
        email="auth@test.com",
        password_hash=hash_password("Auth12345!"),
        full_name="Auth User",
        role=UserRole.admin,
        is_active=True,
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)

    result = await authenticate_user(session, "auth@test.com", "Auth12345!")
    assert result is not None
    assert result.email == "auth@test.com"


@pytest.mark.asyncio
async def test_authenticate_user_wrong_password(session):
    user = User(
        email="wrongpw@test.com",
        password_hash=hash_password("Correct123!"),
        full_name="Wrong PW",
        role=UserRole.admin,
        is_active=True,
    )
    session.add(user)
    await session.commit()

    result = await authenticate_user(session, "wrongpw@test.com", "Wrong123!")
    assert result is None


@pytest.mark.asyncio
async def test_authenticate_user_inactive(session):
    user = User(
        email="inactive2@test.com",
        password_hash=hash_password("Test12345!"),
        full_name="Inactive",
        role=UserRole.admin,
        is_active=False,
    )
    session.add(user)
    await session.commit()

    result = await authenticate_user(session, "inactive2@test.com", "Test12345!")
    assert result is None


@pytest.mark.asyncio
async def test_authenticate_user_nonexistent(session):
    result = await authenticate_user(session, "ghost@test.com", "Test12345!")
    assert result is None


@pytest.mark.asyncio
async def test_generate_token_pair(session):
    user = User(
        email="token@test.com",
        password_hash=hash_password("Test12345!"),
        full_name="Token User",
        role=UserRole.admin,
        language="de",
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)

    tokens = generate_token_pair(user)
    assert "access_token" in tokens
    assert "refresh_token" in tokens
    assert tokens["token_type"] == "bearer"
    assert tokens["expires_in"] > 0


@pytest.mark.asyncio
async def test_refresh_access_token_valid(session):
    user = User(
        email="refresh@test.com",
        password_hash=hash_password("Test12345!"),
        full_name="Refresh User",
        role=UserRole.admin,
        language="de",
        is_active=True,
    )
    session.add(user)
    await session.commit()
    await session.refresh(user)

    rt = create_refresh_token(
        user_id=str(user.id), role="admin", email=user.email, lang="de"
    )
    tokens = await refresh_access_token(session, rt)
    assert tokens is not None
    assert "access_token" in tokens


@pytest.mark.asyncio
async def test_refresh_access_token_invalid(session):
    result = await refresh_access_token(session, "invalid.token.here")
    assert result is None


@pytest.mark.asyncio
async def test_refresh_access_token_nonexistent_user(session):
    fake_id = uuid.uuid4()
    rt = create_refresh_token(
        user_id=str(fake_id), role="admin", email="ghost@test.com", lang="de"
    )
    result = await refresh_access_token(session, rt)
    assert result is None


@pytest.mark.asyncio
async def test_create_user_success(session):
    user = await create_user(
        session,
        email="new@test.com",
        password="NewUser123!",
        full_name="New User",
        role="verkaeufer",
        language="en",
    )
    assert user.email == "new@test.com"
    assert user.full_name == "New User"
    assert user.role == UserRole.verkaeufer
    assert user.language == "en"
    assert user.is_active is True
    assert user.password_hash != "NewUser123!"


@pytest.mark.asyncio
async def test_create_user_duplicate(session):
    await create_user(
        session,
        email="dup@test.com",
        password="First123!",
        full_name="First",
    )
    await session.commit()

    with pytest.raises(ValueError, match="already exists"):
        await create_user(
            session,
            email="dup@test.com",
            password="Second123!",
            full_name="Second",
        )


@pytest.mark.asyncio
async def test_list_users_pagination(session):
    for i in range(5):
        u = User(
            email=f"list{i}@test.com",
            password_hash=hash_password("Test12345!"),
            full_name=f"User {i}",
            role=UserRole.verkaeufer,
        )
        session.add(u)
    await session.commit()

    users, total = await list_users(session, page=1, page_size=3)
    assert total == 5
    assert len(users) == 3

    users2, total2 = await list_users(session, page=2, page_size=3)
    assert total2 == 5
    assert len(users2) == 2


@pytest.mark.asyncio
async def test_update_user_success(session):
    user = await create_user(
        session,
        email="update@test.com",
        password="Update123!",
        full_name="Before Update",
    )
    await session.commit()

    updated = await update_user(
        session, user.id, full_name="After Update", language="en"
    )
    assert updated is not None
    assert updated.full_name == "After Update"
    assert updated.language == "en"


@pytest.mark.asyncio
async def test_update_user_not_found(session):
    result = await update_user(session, uuid.uuid4(), full_name="Nobody")
    assert result is None


@pytest.mark.asyncio
async def test_update_user_role(session):
    user = await create_user(
        session,
        email="role@test.com",
        password="Role12345!",
        full_name="Role User",
        role="verkaeufer",
    )
    await session.commit()

    updated = await update_user(session, user.id, role="admin")
    assert updated is not None
    assert updated.role == UserRole.admin


@pytest.mark.asyncio
async def test_deactivate_user_success(session):
    user = await create_user(
        session,
        email="deactivate@test.com",
        password="Deact123!",
        full_name="Deactivate Me",
    )
    await session.commit()

    result = await deactivate_user(session, user.id)
    assert result is not None
    assert result.is_active is False


@pytest.mark.asyncio
async def test_deactivate_user_not_found(session):
    result = await deactivate_user(session, uuid.uuid4())
    assert result is None