287 lines
No EOL
11 KiB
Python
287 lines
No EOL
11 KiB
Python
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from io import BytesIO
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException, UploadFile
|
|
from google.cloud.exceptions import NotFound
|
|
|
|
from app.services.gcs import (
|
|
GCSService,
|
|
upload_file_to_gcs,
|
|
upload_vtt_to_gcs,
|
|
get_signed_download_url,
|
|
generate_signed_upload_url
|
|
)
|
|
|
|
|
|
class TestGCSService:
|
|
"""Test Google Cloud Storage service functionality"""
|
|
|
|
@pytest.fixture
|
|
def mock_storage_client(self):
|
|
"""Mock Google Cloud Storage client"""
|
|
with patch('app.services.gcs.storage.Client') as mock_client:
|
|
mock_bucket = MagicMock()
|
|
mock_client.return_value.bucket.return_value = mock_bucket
|
|
yield mock_client, mock_bucket
|
|
|
|
@pytest.fixture
|
|
def gcs_service(self, mock_storage_client):
|
|
"""Create GCS service instance with mocked client"""
|
|
mock_client, mock_bucket = mock_storage_client
|
|
service = GCSService()
|
|
service.bucket = mock_bucket
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def sample_upload_file(self):
|
|
"""Create a sample UploadFile for testing"""
|
|
file_content = b"test video content"
|
|
file_obj = BytesIO(file_content)
|
|
|
|
upload_file = UploadFile(
|
|
filename="test.mp4",
|
|
file=file_obj,
|
|
content_type="video/mp4"
|
|
)
|
|
return upload_file
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_to_gcs_success(self, gcs_service, sample_upload_file):
|
|
"""Test successful file upload to GCS"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
# Mock the upload operation
|
|
def mock_upload(file_obj):
|
|
mock_blob.upload_from_file(file_obj)
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result("gs://test-bucket/test/path.mp4")
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.upload_file_to_gcs(
|
|
sample_upload_file,
|
|
"test/path.mp4"
|
|
)
|
|
|
|
assert result == "gs://test-bucket/test/path.mp4"
|
|
assert mock_blob.content_type == "video/mp4"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_with_custom_content_type(self, gcs_service, sample_upload_file):
|
|
"""Test file upload with custom content type"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result("gs://test-bucket/test/path.mp4")
|
|
mock_submit.return_value = future
|
|
|
|
await gcs_service.upload_file_to_gcs(
|
|
sample_upload_file,
|
|
"test/path.mp4",
|
|
content_type="application/octet-stream"
|
|
)
|
|
|
|
assert mock_blob.content_type == "application/octet-stream"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_file_failure(self, gcs_service, sample_upload_file):
|
|
"""Test file upload failure handling"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(Exception("Upload failed"))
|
|
mock_submit.return_value = future
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await gcs_service.upload_file_to_gcs(sample_upload_file, "test/path.mp4")
|
|
|
|
assert exc_info.value.status_code == 500
|
|
assert "File upload failed" in str(exc_info.value.detail)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_text_to_gcs_success(self, gcs_service):
|
|
"""Test successful text upload to GCS"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result("gs://test-bucket/test/file.txt")
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.upload_text_to_gcs(
|
|
"test content",
|
|
"test/file.txt",
|
|
"text/plain"
|
|
)
|
|
|
|
assert result == "gs://test-bucket/test/file.txt"
|
|
assert mock_blob.content_type == "text/plain"
|
|
mock_blob.upload_from_string.assert_called_once_with("test content")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_url_success(self, gcs_service):
|
|
"""Test successful signed URL generation"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = True
|
|
mock_blob.generate_signed_url.return_value = "https://signed-url.example.com"
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result("https://signed-url.example.com")
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.get_signed_url("test/file.mp4")
|
|
|
|
assert result == "https://signed-url.example.com"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_url_file_not_found(self, gcs_service):
|
|
"""Test signed URL generation for non-existent file"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = False
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(NotFound("File not found"))
|
|
mock_submit.return_value = future
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await gcs_service.get_signed_url("test/nonexistent.mp4")
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "File not found" in str(exc_info.value.detail)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_success(self, gcs_service):
|
|
"""Test successful file deletion"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result(True)
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.delete_file("test/file.mp4")
|
|
|
|
assert result is True
|
|
mock_blob.delete.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_not_found(self, gcs_service):
|
|
"""Test deleting non-existent file"""
|
|
mock_blob = MagicMock()
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_exception(NotFound("File not found"))
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.delete_file("test/nonexistent.mp4")
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_exists_true(self, gcs_service):
|
|
"""Test checking if file exists (true case)"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = True
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result(True)
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.file_exists("test/file.mp4")
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_exists_false(self, gcs_service):
|
|
"""Test checking if file exists (false case)"""
|
|
mock_blob = MagicMock()
|
|
mock_blob.exists.return_value = False
|
|
gcs_service.bucket.blob.return_value = mock_blob
|
|
|
|
with patch.object(gcs_service.executor, 'submit') as mock_submit:
|
|
future = asyncio.Future()
|
|
future.set_result(False)
|
|
mock_submit.return_value = future
|
|
|
|
result = await gcs_service.file_exists("test/file.mp4")
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestGCSConvenienceFunctions:
|
|
"""Test convenience functions for GCS operations"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_vtt_to_gcs(self):
|
|
"""Test VTT upload convenience function"""
|
|
vtt_content = """WEBVTT
|
|
|
|
00:00:01.000 --> 00:00:03.000
|
|
Hello world
|
|
"""
|
|
|
|
with patch('app.services.gcs.gcs_service.upload_text_to_gcs') as mock_upload:
|
|
mock_upload.return_value = "gs://bucket/test.vtt"
|
|
|
|
result = await upload_vtt_to_gcs(vtt_content, "test.vtt")
|
|
|
|
assert result == "gs://bucket/test.vtt"
|
|
mock_upload.assert_called_once_with(vtt_content, "test.vtt", "text/vtt")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_signed_download_url(self):
|
|
"""Test signed download URL convenience function"""
|
|
with patch('app.services.gcs.gcs_service.get_signed_url') as mock_get_url:
|
|
mock_get_url.return_value = "https://signed-url.example.com"
|
|
|
|
result = await get_signed_download_url("test/file.mp4", 12)
|
|
|
|
assert result == "https://signed-url.example.com"
|
|
mock_get_url.assert_called_once_with("test/file.mp4", 12)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_signed_upload_url(self):
|
|
"""Test signed upload URL generation"""
|
|
expected_result = {
|
|
"url": "https://upload-url.example.com",
|
|
"fields": {"Content-Type": "video/mp4"}
|
|
}
|
|
|
|
with patch('app.services.gcs.gcs_service.bucket') as mock_bucket:
|
|
mock_blob = MagicMock()
|
|
mock_blob.generate_signed_post_policy_v4.return_value = (
|
|
expected_result["url"],
|
|
expected_result["fields"]
|
|
)
|
|
mock_bucket.blob.return_value = mock_blob
|
|
|
|
with patch('app.services.gcs.gcs_service.executor') as mock_executor:
|
|
future = asyncio.Future()
|
|
future.set_result(expected_result)
|
|
mock_executor.submit.return_value = future
|
|
|
|
result = await generate_signed_upload_url(
|
|
"test/file.mp4",
|
|
"video/mp4"
|
|
)
|
|
|
|
assert result == expected_result |