311 lines
12 KiB
Python
311 lines
12 KiB
Python
"""
|
|
Test suite for DocumentContentRepository with async/await support.
|
|
|
|
This module contains comprehensive tests for all DocumentContentRepository methods
|
|
using mongomock-motor for in-memory MongoDB testing.
|
|
"""
|
|
|
|
import pytest
|
|
import hashlib
|
|
from datetime import datetime
|
|
|
|
import pytest_asyncio
|
|
from bson import ObjectId
|
|
from pymongo.errors import DuplicateKeyError
|
|
from mongomock_motor import AsyncMongoMockClient
|
|
|
|
from app.database.repositories.document_content_repository import DocumentContentRepository
|
|
from app.models.document import DocumentContent
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def in_memory_repository():
|
|
"""Create an in-memory DocumentContentRepository for testing."""
|
|
client = AsyncMongoMockClient()
|
|
db = client.test_database
|
|
repo = DocumentContentRepository(db)
|
|
await repo.initialize()
|
|
return repo
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_document_content():
|
|
"""Sample DocumentContent data for testing."""
|
|
content = "This is sample document content for testing purposes."
|
|
file_hash = hashlib.sha256(content.encode()).hexdigest()
|
|
|
|
return DocumentContent(
|
|
file_hash=file_hash,
|
|
content=content,
|
|
encoding="utf-8",
|
|
file_size=len(content.encode()),
|
|
mime_type="text/plain"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def another_document_content():
|
|
"""Another sample DocumentContent data for testing."""
|
|
content = "This is another sample document with different content."
|
|
file_hash = hashlib.sha256(content.encode()).hexdigest()
|
|
|
|
return DocumentContent(
|
|
file_hash=file_hash,
|
|
content=content,
|
|
encoding="utf-8",
|
|
file_size=len(content.encode()),
|
|
mime_type="text/plain"
|
|
)
|
|
|
|
|
|
class TestDocumentContentRepositoryCreation:
|
|
"""Tests for document content creation functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_create_document_content(self, in_memory_repository, sample_document_content):
|
|
"""Test successful document content creation."""
|
|
# Act
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Assert
|
|
assert created_content is not None
|
|
assert created_content.file_hash == sample_document_content.file_hash
|
|
assert created_content.content == sample_document_content.content
|
|
assert created_content.encoding == sample_document_content.encoding
|
|
assert created_content.file_size == sample_document_content.file_size
|
|
assert created_content.mime_type == sample_document_content.mime_type
|
|
assert created_content.id is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_create_document_content_with_duplicate_file_hash(self, in_memory_repository,
|
|
sample_document_content):
|
|
"""Test that creating document content with duplicate file_hash raises DuplicateKeyError."""
|
|
# Arrange
|
|
await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Act & Assert
|
|
with pytest.raises(DuplicateKeyError) as exc_info:
|
|
await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
assert "already exists" in str(exc_info.value)
|
|
|
|
|
|
class TestDocumentContentRepositoryFinding:
|
|
"""Tests for document content finding functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_find_document_content_by_id(self, in_memory_repository, sample_document_content):
|
|
"""Test finding document content by valid ID."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Act
|
|
found_content = await in_memory_repository.find_document_content_by_id(str(created_content.id))
|
|
|
|
# Assert
|
|
assert found_content is not None
|
|
assert found_content.id == created_content.id
|
|
assert found_content.file_hash == created_content.file_hash
|
|
assert found_content.content == created_content.content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_find_document_content_by_invalid_id(self, in_memory_repository):
|
|
"""Test that invalid ObjectId returns None."""
|
|
# Act
|
|
found_content = await in_memory_repository.find_document_content_by_id("invalid_id")
|
|
|
|
# Assert
|
|
assert found_content is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_find_document_content_by_nonexistent_id(self, in_memory_repository):
|
|
"""Test that nonexistent but valid ObjectId returns None."""
|
|
# Arrange
|
|
nonexistent_id = str(ObjectId())
|
|
|
|
# Act
|
|
found_content = await in_memory_repository.find_document_content_by_id(nonexistent_id)
|
|
|
|
# Assert
|
|
assert found_content is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_find_document_content_by_file_hash(self, in_memory_repository, sample_document_content):
|
|
"""Test finding document content by file hash."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Act
|
|
found_content = await in_memory_repository.find_document_content_by_file_hash(sample_document_content.file_hash)
|
|
|
|
# Assert
|
|
assert found_content is not None
|
|
assert found_content.file_hash == created_content.file_hash
|
|
assert found_content.id == created_content.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_find_document_content_by_nonexistent_file_hash(self, in_memory_repository):
|
|
"""Test that nonexistent file hash returns None."""
|
|
# Act
|
|
found_content = await in_memory_repository.find_document_content_by_file_hash("nonexistent_hash")
|
|
|
|
# Assert
|
|
assert found_content is None
|
|
|
|
|
|
class TestDocumentContentRepositoryUpdate:
|
|
"""Tests for document content update functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_update_document_content(self, in_memory_repository, sample_document_content):
|
|
"""Test successful document content update."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
update_data = {
|
|
"content": "Updated content for testing",
|
|
"encoding": "utf-16",
|
|
"mime_type": "text/html"
|
|
}
|
|
|
|
# Act
|
|
updated_content = await in_memory_repository.update_document_content(str(created_content.id), update_data)
|
|
|
|
# Assert
|
|
assert updated_content is not None
|
|
assert updated_content.content == update_data["content"]
|
|
assert updated_content.encoding == update_data["encoding"]
|
|
assert updated_content.mime_type == update_data["mime_type"]
|
|
assert updated_content.id == created_content.id
|
|
assert updated_content.file_hash == created_content.file_hash # Should remain unchanged
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_update_document_content_with_invalid_id(self, in_memory_repository):
|
|
"""Test that updating with invalid ID returns None."""
|
|
# Act
|
|
result = await in_memory_repository.update_document_content("invalid_id", {"content": "test"})
|
|
|
|
# Assert
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_update_document_content_with_partial_data(self, in_memory_repository, sample_document_content):
|
|
"""Test updating document content with partial data."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
partial_update = {"encoding": "iso-8859-1"}
|
|
|
|
# Act
|
|
updated_content = await in_memory_repository.update_document_content(str(created_content.id), partial_update)
|
|
|
|
# Assert
|
|
assert updated_content is not None
|
|
assert updated_content.encoding == "iso-8859-1"
|
|
assert updated_content.content == created_content.content # Should remain unchanged
|
|
assert updated_content.mime_type == created_content.mime_type # Should remain unchanged
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_update_document_content_with_empty_data(self, in_memory_repository, sample_document_content):
|
|
"""Test updating document content with empty data returns current content."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
empty_update = {}
|
|
|
|
# Act
|
|
result = await in_memory_repository.update_document_content(str(created_content.id), empty_update)
|
|
|
|
# Assert
|
|
assert result is not None
|
|
assert result.content == created_content.content
|
|
assert result.encoding == created_content.encoding
|
|
assert result.mime_type == created_content.mime_type
|
|
|
|
|
|
class TestDocumentContentRepositoryDeletion:
|
|
"""Tests for document content deletion functionality."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_delete_document_content(self, in_memory_repository, sample_document_content):
|
|
"""Test successful document content deletion."""
|
|
# Arrange
|
|
created_content = await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Act
|
|
deletion_result = await in_memory_repository.delete_document_content(str(created_content.id))
|
|
|
|
# Assert
|
|
assert deletion_result is True
|
|
|
|
# Verify content is actually deleted
|
|
found_content = await in_memory_repository.find_document_content_by_id(str(created_content.id))
|
|
assert found_content is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_delete_document_content_with_invalid_id(self, in_memory_repository):
|
|
"""Test that deleting with invalid ID returns False."""
|
|
# Act
|
|
result = await in_memory_repository.delete_document_content("invalid_id")
|
|
|
|
# Assert
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_cannot_delete_nonexistent_document_content(self, in_memory_repository):
|
|
"""Test that deleting nonexistent document content returns False."""
|
|
# Arrange
|
|
nonexistent_id = str(ObjectId())
|
|
|
|
# Act
|
|
result = await in_memory_repository.delete_document_content(nonexistent_id)
|
|
|
|
# Assert
|
|
assert result is False
|
|
|
|
|
|
class TestDocumentContentRepositoryUtilities:
|
|
"""Tests for utility methods."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_check_content_exists(self, in_memory_repository, sample_document_content):
|
|
"""Test checking if document content exists by file hash."""
|
|
# Arrange
|
|
await in_memory_repository.create_document_content(sample_document_content)
|
|
|
|
# Act
|
|
exists = await in_memory_repository.content_exists(sample_document_content.file_hash)
|
|
not_exists = await in_memory_repository.content_exists("nonexistent_hash")
|
|
|
|
# Assert
|
|
assert exists is True
|
|
assert not_exists is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_list_document_contents(self, in_memory_repository, sample_document_content,
|
|
another_document_content):
|
|
"""Test listing document contents with pagination."""
|
|
# Arrange
|
|
await in_memory_repository.create_document_content(sample_document_content)
|
|
await in_memory_repository.create_document_content(another_document_content)
|
|
|
|
# Act
|
|
all_contents = await in_memory_repository.list_document_contents()
|
|
limited_contents = await in_memory_repository.list_document_contents(skip=0, limit=1)
|
|
|
|
# Assert
|
|
assert len(all_contents) == 2
|
|
assert len(limited_contents) == 1
|
|
assert all(isinstance(content, DocumentContent) for content in all_contents)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_i_can_count_document_contents(self, in_memory_repository, sample_document_content,
|
|
another_document_content):
|
|
"""Test counting document contents."""
|
|
# Arrange
|
|
initial_count = await in_memory_repository.count_document_contents()
|
|
await in_memory_repository.create_document_content(sample_document_content)
|
|
await in_memory_repository.create_document_content(another_document_content)
|
|
|
|
# Act
|
|
final_count = await in_memory_repository.count_document_contents()
|
|
|
|
# Assert
|
|
assert final_count == initial_count + 2 |