mirror of
https://github.com/arc53/DocsGPT.git
synced 2025-11-29 16:43:16 +00:00
d6c49bdbf0
- Add 104 comprehensive tests for agent system - Integrate agent tests into CI/CD pipeline - Standardize tests with @pytest.mark.unit markers - Fix cross-platform path compatibility - Clean up unused imports and dependencies
402 lines
14 KiB
Python
402 lines
14 KiB
Python
import io
|
|
import os
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
|
|
import pytest
|
|
from application.storage.local import LocalStorage
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_base_dir():
|
|
return "/tmp/test_storage"
|
|
|
|
|
|
@pytest.fixture
|
|
def local_storage(temp_base_dir):
|
|
return LocalStorage(base_dir=temp_base_dir)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageInitialization:
|
|
|
|
def test_init_with_custom_base_dir(self):
|
|
storage = LocalStorage(base_dir="/custom/path")
|
|
assert storage.base_dir == "/custom/path"
|
|
|
|
def test_init_with_default_base_dir(self):
|
|
storage = LocalStorage()
|
|
assert storage.base_dir is not None
|
|
assert isinstance(storage.base_dir, str)
|
|
|
|
def test_get_full_path_with_relative_path(self, local_storage):
|
|
result = local_storage._get_full_path("documents/test.txt")
|
|
expected = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert os.path.normpath(result) == os.path.normpath(expected)
|
|
|
|
def test_get_full_path_with_absolute_path(self, local_storage):
|
|
result = local_storage._get_full_path("/absolute/path/test.txt")
|
|
assert result == "/absolute/path/test.txt"
|
|
|
|
@patch("os.makedirs")
|
|
@patch("builtins.open", new_callable=mock_open)
|
|
@patch("shutil.copyfileobj")
|
|
def test_save_file_creates_directory_and_saves(
|
|
self, mock_copyfileobj, mock_file, mock_makedirs, local_storage
|
|
):
|
|
file_data = io.BytesIO(b"test content")
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.save_file(file_data, path)
|
|
|
|
expected_dir = os.path.join("/tmp/test_storage", "documents")
|
|
expected_file = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
|
|
assert mock_makedirs.call_count == 1
|
|
assert os.path.normpath(mock_makedirs.call_args[0][0]) == os.path.normpath(
|
|
expected_dir
|
|
)
|
|
assert mock_makedirs.call_args[1] == {"exist_ok": True}
|
|
|
|
assert mock_file.call_count == 1
|
|
assert os.path.normpath(mock_file.call_args[0][0]) == os.path.normpath(
|
|
expected_file
|
|
)
|
|
assert mock_file.call_args[0][1] == "wb"
|
|
|
|
mock_copyfileobj.assert_called_once_with(file_data, mock_file())
|
|
assert result == {"storage_type": "local"}
|
|
|
|
@patch("os.makedirs")
|
|
def test_save_file_with_save_method(self, mock_makedirs, local_storage):
|
|
file_data = MagicMock()
|
|
file_data.save = MagicMock()
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.save_file(file_data, path)
|
|
|
|
expected_file = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert file_data.save.call_count == 1
|
|
assert os.path.normpath(file_data.save.call_args[0][0]) == os.path.normpath(
|
|
expected_file
|
|
)
|
|
assert result == {"storage_type": "local"}
|
|
|
|
@patch("os.makedirs")
|
|
@patch("builtins.open", new_callable=mock_open)
|
|
def test_save_file_with_absolute_path(
|
|
self, mock_file, mock_makedirs, local_storage
|
|
):
|
|
file_data = io.BytesIO(b"test content")
|
|
path = "/absolute/path/test.txt"
|
|
|
|
local_storage.save_file(file_data, path)
|
|
|
|
mock_makedirs.assert_called_once_with("/absolute/path", exist_ok=True)
|
|
mock_file.assert_called_once_with("/absolute/path/test.txt", "wb")
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageGetFile:
|
|
|
|
@patch("os.path.exists", return_value=True)
|
|
@patch("builtins.open", new_callable=mock_open, read_data=b"file content")
|
|
def test_get_file_returns_file_handle(self, mock_file, mock_exists, local_storage):
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.get_file(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert mock_file.call_count == 1
|
|
assert os.path.normpath(mock_file.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert result is not None
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_get_file_raises_error_when_not_found(self, mock_exists, local_storage):
|
|
path = "documents/nonexistent.txt"
|
|
|
|
with pytest.raises(FileNotFoundError, match="File not found"):
|
|
local_storage.get_file(path)
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/nonexistent.txt")
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageDeleteFile:
|
|
|
|
@patch("os.remove")
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_delete_file_removes_existing_file(
|
|
self, mock_exists, mock_remove, local_storage
|
|
):
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.delete_file(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert result is True
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert mock_remove.call_count == 1
|
|
assert os.path.normpath(mock_remove.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_delete_file_returns_false_when_not_found(self, mock_exists, local_storage):
|
|
path = "documents/nonexistent.txt"
|
|
|
|
result = local_storage.delete_file(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/nonexistent.txt")
|
|
assert result is False
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageFileExists:
|
|
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_file_exists_returns_true_when_file_found(self, mock_exists, local_storage):
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.file_exists(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert result is True
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_file_exists_returns_false_when_not_found(self, mock_exists, local_storage):
|
|
path = "documents/nonexistent.txt"
|
|
|
|
result = local_storage.file_exists(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/nonexistent.txt")
|
|
assert result is False
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageListFiles:
|
|
|
|
@patch("os.walk")
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_list_files_returns_all_files_in_directory(
|
|
self, mock_exists, mock_walk, local_storage
|
|
):
|
|
directory = "documents"
|
|
base_dir = os.path.join("/tmp/test_storage", "documents")
|
|
|
|
mock_walk.return_value = [
|
|
(base_dir, ["subdir"], ["file1.txt", "file2.txt"]),
|
|
(os.path.join(base_dir, "subdir"), [], ["file3.txt"]),
|
|
]
|
|
|
|
result = local_storage.list_files(directory)
|
|
|
|
assert len(result) == 3
|
|
result_normalized = [os.path.normpath(f) for f in result]
|
|
assert os.path.normpath("documents/file1.txt") in result_normalized
|
|
assert os.path.normpath("documents/file2.txt") in result_normalized
|
|
assert os.path.normpath("documents/subdir/file3.txt") in result_normalized
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_list_files_returns_empty_list_when_directory_not_found(
|
|
self, mock_exists, local_storage
|
|
):
|
|
directory = "nonexistent"
|
|
|
|
result = local_storage.list_files(directory)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "nonexistent")
|
|
assert result == []
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageProcessFile:
|
|
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_process_file_calls_processor_with_full_path(
|
|
self, mock_exists, local_storage
|
|
):
|
|
path = "documents/test.txt"
|
|
processor_func = MagicMock(return_value="processed")
|
|
|
|
result = local_storage.process_file(path, processor_func, extra_arg="value")
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert result == "processed"
|
|
assert processor_func.call_count == 1
|
|
call_kwargs = processor_func.call_args[1]
|
|
assert os.path.normpath(call_kwargs["local_path"]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert call_kwargs["extra_arg"] == "value"
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_process_file_raises_error_when_file_not_found(
|
|
self, mock_exists, local_storage
|
|
):
|
|
path = "documents/nonexistent.txt"
|
|
processor_func = MagicMock()
|
|
|
|
with pytest.raises(FileNotFoundError, match="File not found"):
|
|
local_storage.process_file(path, processor_func)
|
|
processor_func.assert_not_called()
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageIsDirectory:
|
|
|
|
@patch("os.path.isdir", return_value=True)
|
|
def test_is_directory_returns_true_when_directory_exists(
|
|
self, mock_isdir, local_storage
|
|
):
|
|
path = "documents"
|
|
|
|
result = local_storage.is_directory(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents")
|
|
assert result is True
|
|
assert mock_isdir.call_count == 1
|
|
assert os.path.normpath(mock_isdir.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("os.path.isdir", return_value=False)
|
|
def test_is_directory_returns_false_when_not_directory(
|
|
self, mock_isdir, local_storage
|
|
):
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.is_directory(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert result is False
|
|
assert mock_isdir.call_count == 1
|
|
assert os.path.normpath(mock_isdir.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestLocalStorageRemoveDirectory:
|
|
|
|
@patch("shutil.rmtree")
|
|
@patch("os.path.isdir", return_value=True)
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_remove_directory_deletes_directory(
|
|
self, mock_exists, mock_isdir, mock_rmtree, local_storage
|
|
):
|
|
directory = "documents"
|
|
|
|
result = local_storage.remove_directory(directory)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents")
|
|
assert result is True
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert mock_isdir.call_count == 1
|
|
assert os.path.normpath(mock_isdir.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert mock_rmtree.call_count == 1
|
|
assert os.path.normpath(mock_rmtree.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("os.path.exists", return_value=False)
|
|
def test_remove_directory_returns_false_when_not_exists(
|
|
self, mock_exists, local_storage
|
|
):
|
|
directory = "nonexistent"
|
|
|
|
result = local_storage.remove_directory(directory)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "nonexistent")
|
|
assert result is False
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("os.path.isdir", return_value=False)
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_remove_directory_returns_false_when_not_directory(
|
|
self, mock_exists, mock_isdir, local_storage
|
|
):
|
|
path = "documents/test.txt"
|
|
|
|
result = local_storage.remove_directory(path)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents/test.txt")
|
|
assert result is False
|
|
assert mock_exists.call_count == 1
|
|
assert os.path.normpath(mock_exists.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
assert mock_isdir.call_count == 1
|
|
assert os.path.normpath(mock_isdir.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("shutil.rmtree", side_effect=OSError("Permission denied"))
|
|
@patch("os.path.isdir", return_value=True)
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_remove_directory_returns_false_on_os_error(
|
|
self, mock_exists, mock_isdir, mock_rmtree, local_storage
|
|
):
|
|
directory = "documents"
|
|
|
|
result = local_storage.remove_directory(directory)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents")
|
|
assert result is False
|
|
assert mock_rmtree.call_count == 1
|
|
assert os.path.normpath(mock_rmtree.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|
|
|
|
@patch("shutil.rmtree", side_effect=PermissionError("Access denied"))
|
|
@patch("os.path.isdir", return_value=True)
|
|
@patch("os.path.exists", return_value=True)
|
|
def test_remove_directory_returns_false_on_permission_error(
|
|
self, mock_exists, mock_isdir, mock_rmtree, local_storage
|
|
):
|
|
directory = "documents"
|
|
|
|
result = local_storage.remove_directory(directory)
|
|
|
|
expected_path = os.path.join("/tmp/test_storage", "documents")
|
|
assert result is False
|
|
assert mock_rmtree.call_count == 1
|
|
assert os.path.normpath(mock_rmtree.call_args[0][0]) == os.path.normpath(
|
|
expected_path
|
|
)
|