CoreRasurae
344 lines
13 KiB
Python
344 lines
13 KiB
Python
"""
|
|
Local Provider Tests
|
|
=================
|
|
|
|
This module contains tests for the LocalStorageProvider class.
|
|
"""
|
|
|
|
import os
|
|
import pytest
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
from clevercloud_storage_framework.providers.local_provider import LocalStorageProvider
|
|
from clevercloud_storage_framework.exceptions import (
|
|
StorageError, InvalidPathError, ResourceNotFoundError, StoragePermissionError,
|
|
UnsupportedError
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def local_provider():
|
|
"""Create a LocalStorageProvider instance for testing."""
|
|
return LocalStorageProvider()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for testing."""
|
|
temp_dir = tempfile.mkdtemp()
|
|
yield temp_dir
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
class TestLocalStorageProvider:
|
|
"""Tests for the LocalStorageProvider class."""
|
|
|
|
def test_parse_path(self, local_provider):
|
|
"""Test parsing local paths."""
|
|
# Valid paths
|
|
assert local_provider.parse_path('/path/to/file') == ('/path/to/file', '')
|
|
assert local_provider.parse_path('relative/path') == ('relative/path', '')
|
|
|
|
# Invalid paths
|
|
with pytest.raises(InvalidPathError):
|
|
local_provider.parse_path('')
|
|
|
|
def test_supports_path(self, local_provider):
|
|
"""Test checking if a path is supported."""
|
|
assert local_provider.supports_path('/local/path') is True
|
|
assert local_provider.supports_path('relative/path') is True
|
|
assert local_provider.supports_path('s3://bucket/key') is False
|
|
assert local_provider.supports_path('efs://filesystem/path') is False
|
|
assert local_provider.supports_path('') is False
|
|
|
|
def test_get_provider_name(self, local_provider):
|
|
"""Test getting the provider name."""
|
|
assert local_provider.get_provider_name() == 'local'
|
|
|
|
def test_file_exists(self, local_provider, temp_dir):
|
|
"""Test checking if a file exists."""
|
|
# Create a test file
|
|
test_file = os.path.join(temp_dir, 'test-file.txt')
|
|
with open(test_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
assert local_provider.file_exists(test_file) is True
|
|
assert local_provider.file_exists(os.path.join(temp_dir, 'nonexistent.txt')) is False
|
|
|
|
def test_get_file_size(self, local_provider, temp_dir):
|
|
"""Test getting a file's size."""
|
|
# Create a test file
|
|
test_file = os.path.join(temp_dir, 'test-file.txt')
|
|
with open(test_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
assert local_provider.get_file_size(test_file) == 12 # Length of 'test content'
|
|
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.get_file_size(os.path.join(temp_dir, 'nonexistent.txt'))
|
|
|
|
def test_get_file_metadata(self, local_provider, temp_dir):
|
|
"""Test getting a file's metadata."""
|
|
# Create a test file
|
|
test_file = os.path.join(temp_dir, 'test-file.txt')
|
|
with open(test_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
metadata = local_provider.get_file_metadata(test_file)
|
|
assert metadata['name'] == 'test-file.txt'
|
|
assert metadata['path'] == test_file
|
|
assert metadata['size'] == 12
|
|
assert metadata['type'] == 'file'
|
|
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.get_file_metadata(os.path.join(temp_dir, 'nonexistent.txt'))
|
|
|
|
def test_list_files(self, local_provider, temp_dir):
|
|
"""Test listing files in a directory."""
|
|
# Create test files and directories
|
|
os.makedirs(os.path.join(temp_dir, 'subdir'))
|
|
with open(os.path.join(temp_dir, 'file1.txt'), 'w') as f:
|
|
f.write('file1 content')
|
|
with open(os.path.join(temp_dir, 'file2.txt'), 'w') as f:
|
|
f.write('file2 content')
|
|
with open(os.path.join(temp_dir, 'subdir', 'file3.txt'), 'w') as f:
|
|
f.write('file3 content')
|
|
|
|
# List files in the root directory (non-recursive)
|
|
files = list(local_provider.list_files(temp_dir))
|
|
assert len(files) == 3 # 2 files + 1 directory
|
|
|
|
# Check file metadata
|
|
file_names = [f['name'] for f in files]
|
|
assert 'file1.txt' in file_names
|
|
assert 'file2.txt' in file_names
|
|
assert 'subdir' in file_names
|
|
|
|
# List files recursively
|
|
files = list(local_provider.list_files(temp_dir, recursive=True))
|
|
assert len(files) == 4 # 3 files + 1 directory
|
|
|
|
# Check that the nested file is included
|
|
nested_file = next((f for f in files if f['name'] == 'file3.txt'), None)
|
|
assert nested_file is not None
|
|
|
|
# Test listing a non-existent directory
|
|
with pytest.raises(ResourceNotFoundError):
|
|
list(local_provider.list_files(os.path.join(temp_dir, 'nonexistent')))
|
|
|
|
def test_read_file(self, local_provider, temp_dir):
|
|
"""Test reading a file."""
|
|
# Create a test file
|
|
test_file = os.path.join(temp_dir, 'test-file.txt')
|
|
with open(test_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
# Read the entire file
|
|
content = b''.join(local_provider.read_file(test_file))
|
|
assert content == b'test content'
|
|
|
|
# Read with offset
|
|
content = b''.join(local_provider.read_file(test_file, offset=5))
|
|
assert content == b'content'
|
|
|
|
# Read with length
|
|
content = b''.join(local_provider.read_file(test_file, length=4))
|
|
assert content == b'test'
|
|
|
|
# Read with offset and length
|
|
content = b''.join(local_provider.read_file(test_file, offset=5, length=4))
|
|
assert content == b'cont'
|
|
|
|
# Test reading a non-existent file
|
|
with pytest.raises(ResourceNotFoundError):
|
|
list(local_provider.read_file(os.path.join(temp_dir, 'nonexistent.txt')))
|
|
|
|
def test_write_file(self, local_provider, temp_dir):
|
|
"""Test writing a file."""
|
|
# Write a file with bytes
|
|
test_file = os.path.join(temp_dir, 'write-test.txt')
|
|
local_provider.write_file(test_file, b'test content')
|
|
|
|
# Check if the file was written correctly
|
|
with open(test_file, 'rb') as f:
|
|
assert f.read() == b'test content'
|
|
|
|
# Write a file with a file-like object
|
|
test_file2 = os.path.join(temp_dir, 'write-test2.txt')
|
|
with tempfile.NamedTemporaryFile(delete=False) as f:
|
|
f.write(b'file object content')
|
|
f.flush()
|
|
f.close()
|
|
|
|
with open(f.name, 'rb') as f2:
|
|
local_provider.write_file(test_file2, f2)
|
|
|
|
os.unlink(f.name)
|
|
|
|
# Check if the file was written correctly
|
|
with open(test_file2, 'rb') as f:
|
|
assert f.read() == b'file object content'
|
|
|
|
# Write a file with an iterator of bytes chunks
|
|
test_file3 = os.path.join(temp_dir, 'write-test3.txt')
|
|
chunks = [b'chunk1', b'chunk2', b'chunk3']
|
|
local_provider.write_file(test_file3, iter(chunks))
|
|
|
|
# Check if the file was written correctly
|
|
with open(test_file3, 'rb') as f:
|
|
assert f.read() == b'chunk1chunk2chunk3'
|
|
|
|
# Test with invalid content type
|
|
with pytest.raises(StorageError):
|
|
local_provider.write_file(test_file, 123) # Not a valid content type
|
|
|
|
def test_delete_file(self, local_provider, temp_dir):
|
|
"""Test deleting a file."""
|
|
# Create a test file
|
|
test_file = os.path.join(temp_dir, 'delete-test.txt')
|
|
with open(test_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
# Delete the file
|
|
local_provider.delete_file(test_file)
|
|
|
|
# Check if the file was deleted
|
|
assert not os.path.exists(test_file)
|
|
|
|
# Test deleting a non-existent file
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.delete_file(os.path.join(temp_dir, 'nonexistent.txt'))
|
|
|
|
# Test deleting a directory as a file
|
|
dir_path = os.path.join(temp_dir, 'test-dir')
|
|
os.makedirs(dir_path)
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.delete_file(dir_path)
|
|
|
|
def test_copy_file(self, local_provider, temp_dir):
|
|
"""Test copying a file."""
|
|
# Create a test file
|
|
source_file = os.path.join(temp_dir, 'source.txt')
|
|
with open(source_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
# Copy the file
|
|
dest_file = os.path.join(temp_dir, 'dest.txt')
|
|
local_provider.copy_file(source_file, dest_file)
|
|
|
|
# Check if the file was copied correctly
|
|
assert os.path.exists(dest_file)
|
|
with open(dest_file, 'r') as f:
|
|
assert f.read() == 'test content'
|
|
|
|
# Test copying a non-existent file
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.copy_file(os.path.join(temp_dir, 'nonexistent.txt'), dest_file)
|
|
|
|
# Test copying a directory as a file
|
|
dir_path = os.path.join(temp_dir, 'test-dir')
|
|
os.makedirs(dir_path)
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.copy_file(dir_path, dest_file)
|
|
|
|
def test_move_file(self, local_provider, temp_dir):
|
|
"""Test moving a file."""
|
|
# Create a test file
|
|
source_file = os.path.join(temp_dir, 'move-source.txt')
|
|
with open(source_file, 'w') as f:
|
|
f.write('test content')
|
|
|
|
# Move the file
|
|
dest_file = os.path.join(temp_dir, 'move-dest.txt')
|
|
local_provider.move_file(source_file, dest_file)
|
|
|
|
# Check if the file was moved correctly
|
|
assert not os.path.exists(source_file)
|
|
assert os.path.exists(dest_file)
|
|
with open(dest_file, 'r') as f:
|
|
assert f.read() == 'test content'
|
|
|
|
# Test moving a non-existent file
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.move_file(os.path.join(temp_dir, 'nonexistent.txt'), dest_file)
|
|
|
|
# Test moving a directory as a file
|
|
dir_path = os.path.join(temp_dir, 'test-dir')
|
|
os.makedirs(dir_path)
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.move_file(dir_path, dest_file)
|
|
|
|
def test_create_directory(self, local_provider, temp_dir):
|
|
"""Test creating a directory."""
|
|
# Create a directory
|
|
test_dir = os.path.join(temp_dir, 'test-dir')
|
|
local_provider.create_directory(test_dir)
|
|
|
|
# Check if the directory was created
|
|
assert os.path.exists(test_dir)
|
|
assert os.path.isdir(test_dir)
|
|
|
|
# Create a nested directory
|
|
nested_dir = os.path.join(temp_dir, 'parent', 'child')
|
|
local_provider.create_directory(nested_dir)
|
|
|
|
# Check if the nested directory was created
|
|
assert os.path.exists(nested_dir)
|
|
assert os.path.isdir(nested_dir)
|
|
|
|
# Create a directory that already exists
|
|
local_provider.create_directory(test_dir) # Should not raise an error
|
|
|
|
def test_delete_directory(self, local_provider, temp_dir):
|
|
"""Test deleting a directory."""
|
|
# Create a directory with files
|
|
test_dir = os.path.join(temp_dir, 'delete-dir')
|
|
os.makedirs(test_dir)
|
|
with open(os.path.join(test_dir, 'file.txt'), 'w') as f:
|
|
f.write('test content')
|
|
|
|
# Try to delete the directory without recursive flag
|
|
with pytest.raises(StorageError):
|
|
local_provider.delete_directory(test_dir)
|
|
|
|
# Delete the directory with recursive flag
|
|
local_provider.delete_directory(test_dir, recursive=True)
|
|
|
|
# Check if the directory was deleted
|
|
assert not os.path.exists(test_dir)
|
|
|
|
# Create an empty directory
|
|
empty_dir = os.path.join(temp_dir, 'empty-dir')
|
|
os.makedirs(empty_dir)
|
|
|
|
# Delete the empty directory
|
|
local_provider.delete_directory(empty_dir)
|
|
|
|
# Check if the directory was deleted
|
|
assert not os.path.exists(empty_dir)
|
|
|
|
# Test deleting a non-existent directory
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.delete_directory(os.path.join(temp_dir, 'nonexistent'))
|
|
|
|
# Test deleting a file as a directory
|
|
file_path = os.path.join(temp_dir, 'test-file.txt')
|
|
with open(file_path, 'w') as f:
|
|
f.write('test content')
|
|
with pytest.raises(ResourceNotFoundError):
|
|
local_provider.delete_directory(file_path)
|
|
|
|
def test_supports_presigned_urls(self, local_provider):
|
|
"""Test that Local provider does not support pre-signed URLs."""
|
|
assert local_provider.supports_presigned_urls() is False
|
|
|
|
def test_generate_presigned_url(self, local_provider):
|
|
"""Test that Local provider raises UnsupportedError for pre-signed URLs."""
|
|
with pytest.raises(UnsupportedError) as exc_info:
|
|
local_provider.generate_presigned_url('/path/to/file.txt')
|
|
|
|
assert "Pre-signed URLs are not available for this provider" in str(exc_info.value)
|
|
assert exc_info.value.provider == local_provider
|