gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(tests): mock log_task_progress in S3 upload tests

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-17 04:35:00 +00:00
parent e7ae88457d
commit 3d3d68f7a7
1 changed files with 34 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_upload_to_s3.py
+34 -20
View File
@@ -12,9 +12,10 @@ from app.tasks.upload_to_s3 import upload_to_s3
class TestUploadToS3:
"""Tests for S3 upload functionality."""
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_success_with_folder_prefix(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_success_with_folder_prefix(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test successful S3 upload with folder prefix."""
# Setup
test_file = tmp_path / "test.pdf"
@@ -47,9 +48,10 @@ class TestUploadToS3:
assert call_args[1]["ExtraArgs"]["StorageClass"] == "STANDARD"
assert call_args[1]["ExtraArgs"]["ACL"] == "private"
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_without_folder_prefix(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_without_folder_prefix(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test S3 upload without folder prefix."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -72,9 +74,10 @@ class TestUploadToS3:
extra_args = mock_s3.upload_file.call_args[1]["ExtraArgs"]
assert "ACL" not in extra_args
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_with_folder_prefix_no_trailing_slash(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_with_folder_prefix_no_trailing_slash(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test that folder prefix gets trailing slash added."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -95,18 +98,20 @@ class TestUploadToS3:
# Should add trailing slash
assert result.result["s3_key"] == "uploads/docs/test.pdf"
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_file_not_found(self, mock_settings):
def test_upload_file_not_found(self, mock_settings, mock_log):
"""Test S3 upload with non-existent file."""
mock_settings.s3_bucket_name = "my-bucket"
mock_settings.aws_access_key_id = "key"
mock_settings.aws_secret_access_key = "secret"
with pytest.raises(FileNotFoundError):
upload_to_s3.apply(args=["/nonexistent/file.pdf"])
result = upload_to_s3.apply(args=["/nonexistent/file.pdf"])
assert isinstance(result.result, FileNotFoundError)
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_missing_bucket_name(self, mock_settings, tmp_path):
def test_upload_missing_bucket_name(self, mock_settings, mock_log, tmp_path):
"""Test S3 upload with missing bucket name."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -115,11 +120,13 @@ class TestUploadToS3:
mock_settings.aws_access_key_id = "key"
mock_settings.aws_secret_access_key = "secret"
with pytest.raises(ValueError, match="bucket name"):
upload_to_s3.apply(args=[str(test_file)])
result = upload_to_s3.apply(args=[str(test_file)])
assert isinstance(result.result, ValueError)
assert "bucket name" in str(result.result)
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_missing_credentials(self, mock_settings, tmp_path):
def test_upload_missing_credentials(self, mock_settings, mock_log, tmp_path):
"""Test S3 upload with missing AWS credentials."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -128,12 +135,14 @@ class TestUploadToS3:
mock_settings.aws_access_key_id = None
mock_settings.aws_secret_access_key = None
with pytest.raises(ValueError, match="credentials"):
upload_to_s3.apply(args=[str(test_file)])
result = upload_to_s3.apply(args=[str(test_file)])
assert isinstance(result.result, ValueError)
assert "credentials" in str(result.result)
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_client_error(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_client_error(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test S3 upload with boto3 ClientError."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -151,12 +160,14 @@ class TestUploadToS3:
mock_s3.upload_file.side_effect = ClientError(error_response, "upload_file")
mock_boto_client.return_value = mock_s3
with pytest.raises(Exception, match="Failed to upload"):
upload_to_s3.apply(args=[str(test_file)])
result = upload_to_s3.apply(args=[str(test_file)])
assert isinstance(result.result, Exception)
assert "Failed to upload" in str(result.result)
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_generic_exception(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_generic_exception(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test S3 upload with generic exception."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -173,12 +184,14 @@ class TestUploadToS3:
mock_s3.upload_file.side_effect = Exception("Network error")
mock_boto_client.return_value = mock_s3
with pytest.raises(Exception, match="Error uploading"):
upload_to_s3.apply(args=[str(test_file)])
result = upload_to_s3.apply(args=[str(test_file)])
assert isinstance(result.result, Exception)
assert "Error uploading" in str(result.result)
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_with_different_storage_classes(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_with_different_storage_classes(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test S3 upload with different storage classes."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")
@@ -199,9 +212,10 @@ class TestUploadToS3:
extra_args = mock_s3.upload_file.call_args[1]["ExtraArgs"]
assert extra_args["StorageClass"] == "INTELLIGENT_TIERING"
@patch("app.tasks.upload_to_s3.log_task_progress")
@patch("app.tasks.upload_to_s3.boto3.client")
@patch("app.tasks.upload_to_s3.settings")
def test_upload_url_generation(self, mock_settings, mock_boto_client, tmp_path):
def test_upload_url_generation(self, mock_settings, mock_boto_client, mock_log, tmp_path):
"""Test that the S3 URL is properly generated."""
test_file = tmp_path / "test.pdf"
test_file.write_text("test content")