350 lines
8.9 KiB
Python
350 lines
8.9 KiB
Python
"""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
|