Added frontend. Working on user management
This commit is contained in:
385
tests/test_user_repository.py
Normal file
385
tests/test_user_repository.py
Normal file
@@ -0,0 +1,385 @@
|
||||
"""
|
||||
Unit tests for user repository module.
|
||||
|
||||
Tests all CRUD operations for users with MongoDB mocking
|
||||
to ensure proper database interactions without requiring
|
||||
actual MongoDB instance during tests.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock, MagicMock
|
||||
from datetime import datetime
|
||||
from bson import ObjectId
|
||||
from pymongo.errors import DuplicateKeyError
|
||||
|
||||
from app.database.repositories.user_repository import UserRepository
|
||||
from app.models.user import UserCreate, UserUpdate, UserInDB, UserRole
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_database():
|
||||
"""Create mock database with users collection."""
|
||||
db = Mock()
|
||||
collection = Mock()
|
||||
db.users = collection
|
||||
return db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_repository(mock_database):
|
||||
"""Create UserRepository instance with mocked database."""
|
||||
return UserRepository(mock_database)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_user_create():
|
||||
"""Create sample UserCreate object for testing."""
|
||||
return UserCreate(
|
||||
username="testuser",
|
||||
email="test@example.com",
|
||||
hashed_password="hashed_password_123",
|
||||
role=UserRole.USER,
|
||||
is_active=True
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_user_update():
|
||||
"""Create sample UserUpdate object for testing."""
|
||||
return UserUpdate(
|
||||
email="updated@example.com",
|
||||
role=UserRole.ADMIN,
|
||||
is_active=False
|
||||
)
|
||||
|
||||
|
||||
def test_i_can_create_user(user_repository, mock_database, sample_user_create):
|
||||
"""Test successful user creation."""
|
||||
# Mock successful insertion
|
||||
mock_result = Mock()
|
||||
mock_result.inserted_id = ObjectId()
|
||||
mock_database.users.insert_one.return_value = mock_result
|
||||
|
||||
result = user_repository.create_user(sample_user_create)
|
||||
|
||||
assert isinstance(result, UserInDB)
|
||||
assert result.username == sample_user_create.username
|
||||
assert result.email == sample_user_create.email
|
||||
assert result.hashed_password == sample_user_create.hashed_password
|
||||
assert result.role == sample_user_create.role
|
||||
assert result.is_active == sample_user_create.is_active
|
||||
assert result.id is not None
|
||||
assert isinstance(result.created_at, datetime)
|
||||
assert isinstance(result.updated_at, datetime)
|
||||
|
||||
# Verify insert_one was called with correct data
|
||||
mock_database.users.insert_one.assert_called_once()
|
||||
call_args = mock_database.users.insert_one.call_args[0][0]
|
||||
assert call_args["username"] == sample_user_create.username
|
||||
assert call_args["email"] == sample_user_create.email
|
||||
|
||||
|
||||
def test_i_cannot_create_duplicate_username(user_repository, mock_database, sample_user_create):
|
||||
"""Test that creating user with duplicate username raises DuplicateKeyError."""
|
||||
# Mock DuplicateKeyError from MongoDB
|
||||
mock_database.users.insert_one.side_effect = DuplicateKeyError("duplicate key error")
|
||||
|
||||
with pytest.raises(DuplicateKeyError, match="User with username 'testuser' already exists"):
|
||||
user_repository.create_user(sample_user_create)
|
||||
|
||||
|
||||
def test_i_can_find_user_by_username(user_repository, mock_database):
|
||||
"""Test finding user by username."""
|
||||
# Mock user document from database
|
||||
user_doc = {
|
||||
"_id": ObjectId(),
|
||||
"username": "testuser",
|
||||
"email": "test@example.com",
|
||||
"hashed_password": "hashed_password_123",
|
||||
"role": "user",
|
||||
"is_active": True,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
}
|
||||
mock_database.users.find_one.return_value = user_doc
|
||||
|
||||
result = user_repository.find_user_by_username("testuser")
|
||||
|
||||
assert isinstance(result, UserInDB)
|
||||
assert result.username == "testuser"
|
||||
assert result.email == "test@example.com"
|
||||
|
||||
mock_database.users.find_one.assert_called_once_with({"username": "testuser"})
|
||||
|
||||
|
||||
def test_i_cannot_find_nonexistent_user_by_username(user_repository, mock_database):
|
||||
"""Test finding nonexistent user by username returns None."""
|
||||
mock_database.users.find_one.return_value = None
|
||||
|
||||
result = user_repository.find_user_by_username("nonexistent")
|
||||
|
||||
assert result is None
|
||||
mock_database.users.find_one.assert_called_once_with({"username": "nonexistent"})
|
||||
|
||||
|
||||
def test_i_can_find_user_by_id(user_repository, mock_database):
|
||||
"""Test finding user by ID."""
|
||||
user_id = ObjectId()
|
||||
user_doc = {
|
||||
"_id": user_id,
|
||||
"username": "testuser",
|
||||
"email": "test@example.com",
|
||||
"hashed_password": "hashed_password_123",
|
||||
"role": "user",
|
||||
"is_active": True,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
}
|
||||
mock_database.users.find_one.return_value = user_doc
|
||||
|
||||
result = user_repository.find_user_by_id(str(user_id))
|
||||
|
||||
assert isinstance(result, UserInDB)
|
||||
assert result.id == user_id
|
||||
assert result.username == "testuser"
|
||||
|
||||
mock_database.users.find_one.assert_called_once_with({"_id": user_id})
|
||||
|
||||
|
||||
def test_i_cannot_find_user_with_invalid_id(user_repository, mock_database):
|
||||
"""Test finding user with invalid ObjectId returns None."""
|
||||
result = user_repository.find_user_by_id("invalid_id")
|
||||
|
||||
assert result is None
|
||||
# find_one should not be called with invalid ID
|
||||
mock_database.users.find_one.assert_not_called()
|
||||
|
||||
|
||||
def test_i_cannot_find_nonexistent_user_by_id(user_repository, mock_database):
|
||||
"""Test finding nonexistent user by ID returns None."""
|
||||
user_id = ObjectId()
|
||||
mock_database.users.find_one.return_value = None
|
||||
|
||||
result = user_repository.find_user_by_id(str(user_id))
|
||||
|
||||
assert result is None
|
||||
mock_database.users.find_one.assert_called_once_with({"_id": user_id})
|
||||
|
||||
|
||||
def test_i_can_find_user_by_email(user_repository, mock_database):
|
||||
"""Test finding user by email address."""
|
||||
user_doc = {
|
||||
"_id": ObjectId(),
|
||||
"username": "testuser",
|
||||
"email": "test@example.com",
|
||||
"hashed_password": "hashed_password_123",
|
||||
"role": "user",
|
||||
"is_active": True,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
}
|
||||
mock_database.users.find_one.return_value = user_doc
|
||||
|
||||
result = user_repository.find_user_by_email("test@example.com")
|
||||
|
||||
assert isinstance(result, UserInDB)
|
||||
assert result.email == "test@example.com"
|
||||
|
||||
mock_database.users.find_one.assert_called_once_with({"email": "test@example.com"})
|
||||
|
||||
|
||||
def test_i_can_update_user(user_repository, mock_database, sample_user_update):
|
||||
"""Test updating user information."""
|
||||
user_id = ObjectId()
|
||||
|
||||
# Mock successful update
|
||||
mock_update_result = Mock()
|
||||
mock_update_result.matched_count = 1
|
||||
mock_database.users.update_one.return_value = mock_update_result
|
||||
|
||||
# Mock find_one for returning updated user
|
||||
updated_user_doc = {
|
||||
"_id": user_id,
|
||||
"username": "testuser",
|
||||
"email": "updated@example.com",
|
||||
"hashed_password": "hashed_password_123",
|
||||
"role": "admin",
|
||||
"is_active": False,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
}
|
||||
mock_database.users.find_one.return_value = updated_user_doc
|
||||
|
||||
result = user_repository.update_user(str(user_id), sample_user_update)
|
||||
|
||||
assert isinstance(result, UserInDB)
|
||||
assert result.email == "updated@example.com"
|
||||
assert result.role == UserRole.ADMIN
|
||||
assert result.is_active is False
|
||||
|
||||
# Verify update_one was called with correct data
|
||||
mock_database.users.update_one.assert_called_once()
|
||||
call_args = mock_database.users.update_one.call_args
|
||||
assert call_args[0][0] == {"_id": user_id} # Filter
|
||||
update_data = call_args[0][1]["$set"] # Update data
|
||||
assert update_data["email"] == "updated@example.com"
|
||||
assert update_data["role"] == UserRole.ADMIN
|
||||
assert update_data["is_active"] is False
|
||||
assert "updated_at" in update_data
|
||||
|
||||
|
||||
def test_i_cannot_update_nonexistent_user(user_repository, mock_database, sample_user_update):
|
||||
"""Test updating nonexistent user returns None."""
|
||||
user_id = ObjectId()
|
||||
|
||||
# Mock no match found
|
||||
mock_update_result = Mock()
|
||||
mock_update_result.matched_count = 0
|
||||
mock_database.users.update_one.return_value = mock_update_result
|
||||
|
||||
result = user_repository.update_user(str(user_id), sample_user_update)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_i_cannot_update_user_with_invalid_id(user_repository, mock_database, sample_user_update):
|
||||
"""Test updating user with invalid ID returns None."""
|
||||
result = user_repository.update_user("invalid_id", sample_user_update)
|
||||
|
||||
assert result is None
|
||||
# update_one should not be called with invalid ID
|
||||
mock_database.users.update_one.assert_not_called()
|
||||
|
||||
|
||||
def test_i_can_delete_user(user_repository, mock_database):
|
||||
"""Test successful user deletion."""
|
||||
user_id = ObjectId()
|
||||
|
||||
# Mock successful deletion
|
||||
mock_delete_result = Mock()
|
||||
mock_delete_result.deleted_count = 1
|
||||
mock_database.users.delete_one.return_value = mock_delete_result
|
||||
|
||||
result = user_repository.delete_user(str(user_id))
|
||||
|
||||
assert result is True
|
||||
mock_database.users.delete_one.assert_called_once_with({"_id": user_id})
|
||||
|
||||
|
||||
def test_i_cannot_delete_nonexistent_user(user_repository, mock_database):
|
||||
"""Test deleting nonexistent user returns False."""
|
||||
user_id = ObjectId()
|
||||
|
||||
# Mock no deletion occurred
|
||||
mock_delete_result = Mock()
|
||||
mock_delete_result.deleted_count = 0
|
||||
mock_database.users.delete_one.return_value = mock_delete_result
|
||||
|
||||
result = user_repository.delete_user(str(user_id))
|
||||
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_i_cannot_delete_user_with_invalid_id(user_repository, mock_database):
|
||||
"""Test deleting user with invalid ID returns False."""
|
||||
result = user_repository.delete_user("invalid_id")
|
||||
|
||||
assert result is False
|
||||
# delete_one should not be called with invalid ID
|
||||
mock_database.users.delete_one.assert_not_called()
|
||||
|
||||
|
||||
def test_i_can_list_users(user_repository, mock_database):
|
||||
"""Test listing users with pagination."""
|
||||
# Mock cursor with user documents
|
||||
user_docs = [
|
||||
{
|
||||
"_id": ObjectId(),
|
||||
"username": "user1",
|
||||
"email": "user1@example.com",
|
||||
"hashed_password": "hash1",
|
||||
"role": "user",
|
||||
"is_active": True,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
},
|
||||
{
|
||||
"_id": ObjectId(),
|
||||
"username": "user2",
|
||||
"email": "user2@example.com",
|
||||
"hashed_password": "hash2",
|
||||
"role": "admin",
|
||||
"is_active": False,
|
||||
"created_at": datetime.utcnow(),
|
||||
"updated_at": datetime.utcnow()
|
||||
}
|
||||
]
|
||||
|
||||
mock_cursor = Mock()
|
||||
mock_cursor.__iter__.return_value = iter(user_docs)
|
||||
mock_cursor.skip.return_value = mock_cursor
|
||||
mock_cursor.limit.return_value = mock_cursor
|
||||
mock_database.users.find.return_value = mock_cursor
|
||||
|
||||
result = user_repository.list_users(skip=10, limit=50)
|
||||
|
||||
assert len(result) == 2
|
||||
assert all(isinstance(user, UserInDB) for user in result)
|
||||
assert result[0].username == "user1"
|
||||
assert result[1].username == "user2"
|
||||
|
||||
mock_database.users.find.assert_called_once()
|
||||
mock_cursor.skip.assert_called_once_with(10)
|
||||
mock_cursor.limit.assert_called_once_with(50)
|
||||
|
||||
|
||||
def test_i_can_count_users(user_repository, mock_database):
|
||||
"""Test counting total users."""
|
||||
mock_database.users.count_documents.return_value = 42
|
||||
|
||||
result = user_repository.count_users()
|
||||
|
||||
assert result == 42
|
||||
mock_database.users.count_documents.assert_called_once_with({})
|
||||
|
||||
|
||||
def test_i_can_check_user_exists(user_repository, mock_database):
|
||||
"""Test checking if user exists by username."""
|
||||
mock_database.users.count_documents.return_value = 1
|
||||
|
||||
result = user_repository.user_exists("testuser")
|
||||
|
||||
assert result is True
|
||||
mock_database.users.count_documents.assert_called_once_with({"username": "testuser"})
|
||||
|
||||
|
||||
def test_i_can_check_user_does_not_exist(user_repository, mock_database):
|
||||
"""Test checking if user does not exist by username."""
|
||||
mock_database.users.count_documents.return_value = 0
|
||||
|
||||
result = user_repository.user_exists("nonexistent")
|
||||
|
||||
assert result is False
|
||||
mock_database.users.count_documents.assert_called_once_with({"username": "nonexistent"})
|
||||
|
||||
|
||||
def test_i_can_create_indexes_on_initialization(mock_database):
|
||||
"""Test that indexes are created when repository is initialized."""
|
||||
# Mock create_index to not raise exception
|
||||
mock_database.users.create_index.return_value = None
|
||||
|
||||
repository = UserRepository(mock_database)
|
||||
|
||||
mock_database.users.create_index.assert_called_once_with("username", unique=True)
|
||||
|
||||
|
||||
def test_i_can_handle_index_creation_error(mock_database):
|
||||
"""Test that index creation errors are handled gracefully."""
|
||||
# Mock create_index to raise exception (index already exists)
|
||||
mock_database.users.create_index.side_effect = Exception("Index already exists")
|
||||
|
||||
# Should not raise exception
|
||||
repository = UserRepository(mock_database)
|
||||
|
||||
assert repository is not None
|
||||
mock_database.users.create_index.assert_called_once_with("username", unique=True)
|
||||
Reference in New Issue
Block a user