Spaces:
Sleeping
Sleeping
File size: 2,850 Bytes
db4ba8d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """
Tests for storage service security
"""
from unittest.mock import MagicMock, patch
from src.services.ingest_svc import StorageService
def test_minio_bucket_created_private():
"""Test that MinIO bucket is created without adding a public bucket policy."""
with patch("src.services.ingest_svc.boto3.client") as mock_s3:
# Mock S3 client
mock_client = MagicMock()
mock_s3.return_value = mock_client
# Mock head_bucket to raise (bucket doesn't exist)
mock_client.head_bucket.side_effect = Exception("Not found")
# Create storage service
storage = StorageService()
# Verify bucket creation was attempted
mock_client.create_bucket.assert_called_once()
# Verify no bucket policy was applied by default
mock_client.put_bucket_policy.assert_not_called()
def test_minio_bucket_created_without_policy():
"""Test that MinIO bucket creation does not require an API-local policy."""
with patch("src.services.ingest_svc.boto3.client") as mock_s3:
mock_client = MagicMock()
mock_s3.return_value = mock_client
mock_client.head_bucket.side_effect = Exception("Not found")
storage = StorageService()
# Policy should not be created because MinIO is private by default.
mock_client.put_bucket_policy.assert_not_called()
def test_minio_bucket_already_exists():
"""Test that storage service works with existing bucket."""
with patch("src.services.ingest_svc.boto3.client") as mock_s3:
mock_client = MagicMock()
mock_s3.return_value = mock_client
# Mock successful head_bucket (bucket exists)
mock_client.head_bucket.return_value = True
storage = StorageService()
# Should not call create_bucket or put_bucket_policy
mock_client.create_bucket.assert_not_called()
mock_client.put_bucket_policy.assert_not_called()
def test_upload_document_streams_bytes_to_minio():
"""Test that upload_document writes bytes through the S3 client."""
with patch("src.services.ingest_svc.boto3.client") as mock_s3:
mock_client = MagicMock()
mock_s3.return_value = mock_client
mock_client.head_bucket.return_value = True
storage = StorageService()
with patch("src.services.ingest_svc.io.BytesIO") as mock_bytesio:
mock_bytesio.return_value = MagicMock()
with patch.object(storage, "compute_hash") as mock_hash:
mock_hash.return_value = "abc123"
import asyncio
asyncio.run(storage.upload_document("batch", "doc", "file.txt", b"hello"))
mock_client.upload_fileobj.assert_called_once()
mock_bytesio.assert_called_once()
|