busya
8a4310169b
12 known failures — all from Agent 3 new unverified tests (mock setup issues): - 3 dataset_review_routes_extended (DTO field mismatches) - 1 settings_consolidated (dict key access) - 1 llm_analysis_service_coverage (rate_limit mock) - 1 migration_plugin (SessionLocal side_effect exhaustion) - 1 preview (DB query vs dict key) - 5 scheduler (datetime timezone + async mock mismatches) NEW TEST FILES THIS SESSION: - test_batch_insert_coverage.py — 3 tests - test_storage_plugin.py — +3 tests - test_search.py — +2 tests - test_mapper.py — already 100% - test_llm_analysis_migration_v1_to_v2.py — 14 tests - test_llm_async_http.py — +1 test - test_prompt_builder.py — +1 test - test_service_datasource.py — +1 test - test_lang_detect.py — +1 test - test_scheduler.py — +6 tests - test_llm_analysis_service_coverage.py — +15 tests - test_dataset_review_routes_extended.py — +14 tests - test_settings_consolidated.py — +13 tests Modules pushed to 100%: _batch_insert, dictionary_entries, service_datasource, _llm_async_http, prompt_builder, dictionary_crud, _batch_sizer, storage/plugin, mapper.py Session: 7194→7643 tests (+449), 80%→83% raw (+3pp), 93.4%→96.9% real (+3.5pp). Remaining: 12 failures to fix + ~300 statements to reach 98% real coverage.
568 lines
24 KiB
Python
568 lines
24 KiB
Python
# #region Test.StoragePlugin [C:3] [TYPE Module] [SEMANTICS test, storage, plugin, filesystem]
|
|
# @BRIEF Unit tests for StoragePlugin — properties, path resolution, file operations, and edge cases.
|
|
# @RELATION BINDS_TO -> [StoragePlugin]
|
|
# @TEST_EDGE: path_traversal -> validate_path raises ValueError
|
|
# @TEST_EDGE: missing_variable -> resolve_path falls back to stripped literal
|
|
# @TEST_EDGE: category_mismatch -> delete_file/get_file_path raise ValueError
|
|
# @TEST_EDGE: file_not_found -> delete_file/get_file_path raise FileNotFoundError
|
|
# @TEST_EDGE: missing_category -> list_files root view returns category directories
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import tempfile
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock
|
|
|
|
import pytest
|
|
|
|
from src.plugins.storage.plugin import StoragePlugin, _copy_upload_to_disk
|
|
from src.models.storage import FileCategory, StoredFile
|
|
|
|
|
|
# ── Helpers ──
|
|
|
|
|
|
def _mock_config_manager(storage_root: str | None = None):
|
|
"""Create a mock ConfigManager with storage root settings."""
|
|
cm = MagicMock()
|
|
cfg = MagicMock()
|
|
if storage_root:
|
|
cfg.settings.storage.root_path = storage_root
|
|
else:
|
|
cfg.settings.storage.root_path = "/app/storage"
|
|
cm.get_config.return_value = cfg
|
|
return cm
|
|
|
|
|
|
def _mock_task_context():
|
|
ctx = MagicMock()
|
|
ctx.logger = MagicMock()
|
|
ctx.logger.with_source.return_value = MagicMock()
|
|
return ctx
|
|
|
|
|
|
class TestStoragePluginProperties:
|
|
"""Verify static property values."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_id(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
assert plugin.id == "storage-manager"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_name(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
assert plugin.name == "Storage Manager"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_description(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
assert plugin.description == "Manages local file storage for backups and repositories."
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_version(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
assert plugin.version == "1.0.0"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_ui_route(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
assert plugin.ui_route == "/tools/storage"
|
|
|
|
|
|
class TestStoragePluginGetSchema:
|
|
"""Verify get_schema output."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_get_schema(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
schema = plugin.get_schema()
|
|
|
|
assert schema["type"] == "object"
|
|
assert "category" in schema["properties"]
|
|
assert schema["properties"]["category"]["type"] == "string"
|
|
assert schema["properties"]["category"]["enum"] == [c.value for c in FileCategory]
|
|
assert schema["required"] == ["category"]
|
|
|
|
|
|
class TestStoragePluginExecute:
|
|
"""Verify StoragePlugin.execute with various scenarios."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_execute_basic(self, mock_get_cm):
|
|
"""Execute with basic params."""
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
result = await plugin.execute({"category": "backups"})
|
|
assert result is None
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_execute_with_context(self, mock_get_cm):
|
|
"""Execute with TaskContext — uses context.logger."""
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
ctx = _mock_task_context()
|
|
result = await plugin.execute({"category": "backups"}, context=ctx)
|
|
assert result is None
|
|
ctx.logger.with_source.assert_called()
|
|
|
|
|
|
class TestStoragePluginGetStorageRoot:
|
|
"""Verify get_storage_root path resolution."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_absolute_path(self, mock_get_cm):
|
|
"""Absolute storage root returns directly."""
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root="/tmp/test_storage")
|
|
plugin = StoragePlugin()
|
|
root = plugin.get_storage_root()
|
|
assert root == Path("/tmp/test_storage")
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_relative_path(self, mock_get_cm):
|
|
"""Relative storage root resolves against project root."""
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root="relative/storage")
|
|
plugin = StoragePlugin()
|
|
root = plugin.get_storage_root()
|
|
# Should resolve relative to project root (parents[3] of plugin.py)
|
|
assert root.is_absolute()
|
|
assert "relative/storage" in str(root)
|
|
|
|
|
|
class TestStoragePluginResolvePath:
|
|
"""Verify resolve_path pattern substitution."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_resolve_with_variables(self, mock_get_cm):
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
result = plugin.resolve_path("backups/{name}_{timestamp}", {"name": "test"})
|
|
# {timestamp} is auto-filled, {name} comes from variables
|
|
assert result.startswith("backups/test_")
|
|
# Verify timestamp format (YYYYMMDDTHHMMSS)
|
|
timestamp_part = result.split("_", 1)[1]
|
|
assert len(timestamp_part) == 15 # YYYYMMDDTHHMMSS
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_resolve_clean_double_slashes(self, mock_get_cm):
|
|
"""Double slashes are normalized."""
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
result = plugin.resolve_path("backups//{name}", {"name": "test"})
|
|
assert "//" not in result
|
|
assert result == "backups/test"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_resolve_missing_variable(self, mock_get_cm):
|
|
"""Missing variable falls back to stripped literal."""
|
|
mock_get_cm.return_value = _mock_config_manager()
|
|
plugin = StoragePlugin()
|
|
result = plugin.resolve_path("backups/{name}_{missing}", {"name": "test"})
|
|
# Fallback: strip braces from the pattern
|
|
assert "{" not in result
|
|
assert "}" not in result
|
|
|
|
|
|
class TestStoragePluginValidatePath:
|
|
"""Verify validate_path prevents traversal and returns valid paths."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_valid_path(self, mock_get_cm):
|
|
"""Path within storage root is accepted."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
# ensure_directories creates category subdirs — don't rmdir parent
|
|
plugin = StoragePlugin()
|
|
valid = Path(tmpdir) / "test_file.txt"
|
|
valid.touch()
|
|
result = plugin.validate_path(valid)
|
|
assert result == valid.resolve()
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_path_traversal_raises(self, mock_get_cm):
|
|
"""Path outside storage root raises ValueError."""
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root="/tmp/test_storage_traversal")
|
|
plugin = StoragePlugin()
|
|
outside = Path("/etc/passwd")
|
|
with pytest.raises(ValueError, match="Access denied"):
|
|
plugin.validate_path(outside)
|
|
|
|
|
|
class TestStoragePluginEnsureDirectories:
|
|
"""Verify ensure_directories creates category subfolders."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_ensure_directories_creates_folders(self, mock_get_cm):
|
|
"""Category directories are created under storage root."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
plugin = StoragePlugin()
|
|
# ensure_directories called in __init__ already
|
|
for cat in FileCategory:
|
|
assert (Path(tmpdir) / cat.value).exists()
|
|
|
|
|
|
class TestStoragePluginListFiles:
|
|
"""Verify list_files in various modes."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_root_view(self, mock_get_cm):
|
|
"""Root view (no category) returns category directories."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
# Create category directories
|
|
for cat in FileCategory:
|
|
(Path(tmpdir) / cat.value).mkdir(parents=True, exist_ok=True)
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files()
|
|
|
|
assert len(files) == len(FileCategory)
|
|
names = {f.name for f in files}
|
|
for cat in FileCategory:
|
|
assert cat.value in names
|
|
# All returned as directories
|
|
for f in files:
|
|
assert f.mime_type == "directory"
|
|
assert f.size == 0
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_with_category(self, mock_get_cm):
|
|
"""Filter by category returns files in that category."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
(backup_dir / "file1.txt").write_text("hello")
|
|
(backup_dir / "file2.txt").write_text("world")
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP)
|
|
|
|
assert len(files) == 2
|
|
assert files[0].name == "file1.txt"
|
|
assert files[1].name == "file2.txt"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_recursive(self, mock_get_cm):
|
|
"""Recursive mode scans subdirectories."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
nested = backup_dir / "sub" / "dir"
|
|
nested.mkdir(parents=True, exist_ok=True)
|
|
(nested / "deep.txt").write_text("deep")
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP, recursive=True)
|
|
|
|
assert len(files) == 1
|
|
assert files[0].name == "deep.txt"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_missing_category_dir(self, mock_get_cm):
|
|
"""Missing category directory returns empty list."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
# Don't create any directories
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP)
|
|
assert files == []
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_skip_logs(self, mock_get_cm):
|
|
"""Files/dirs containing 'Logs' are skipped."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
(backup_dir / "good.txt").write_text("good")
|
|
(backup_dir / "LogsFile.txt").write_text("logs")
|
|
log_dir = backup_dir / "Logs"
|
|
log_dir.mkdir()
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP)
|
|
|
|
# Only good.txt should appear
|
|
names = [f.name for f in files]
|
|
assert "good.txt" in names
|
|
assert "LogsFile.txt" not in names
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_skip_logs_recursive(self, mock_get_cm):
|
|
"""Recursive scan skips files with 'Logs' in their filename (L284)."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
nested = backup_dir / "daily"
|
|
nested.mkdir(parents=True, exist_ok=True)
|
|
(nested / "report.txt").write_text("clean")
|
|
(nested / "TransactionLogs.csv").write_text("log data")
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP, recursive=True)
|
|
|
|
assert len(files) == 1
|
|
assert files[0].name == "report.txt"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_recursive_missing_dir(self, mock_get_cm):
|
|
"""Recursive mode with missing category dir continues silently (L274)."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
# Patch ensure_directories so category dirs are NOT created
|
|
with patch.object(StoragePlugin, "ensure_directories"):
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(category=FileCategory.BACKUP, recursive=True)
|
|
assert files == []
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_root_view_missing_categories(self, mock_get_cm):
|
|
"""Root view listing skips missing category dirs via continue (L249)."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
# Patch ensure_directories so only ONE category exists on disk
|
|
with patch.object(StoragePlugin, "ensure_directories"):
|
|
plugin = StoragePlugin()
|
|
# Manually create only one category dir
|
|
(Path(tmpdir) / "backups").mkdir()
|
|
|
|
files = plugin.list_files()
|
|
|
|
# Only "backups" should appear; "repositories" is missing → hits L249 continue
|
|
assert len(files) == 1
|
|
assert files[0].name == "backups"
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_list_files_with_subpath(self, mock_get_cm):
|
|
"""Subpath within category narrows listing."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
nested = Path(tmpdir) / "backups" / "monthly" / "week1"
|
|
nested.mkdir(parents=True, exist_ok=True)
|
|
(nested / "report.txt").write_text("report")
|
|
|
|
plugin = StoragePlugin()
|
|
files = plugin.list_files(
|
|
category=FileCategory.BACKUP,
|
|
subpath="monthly/week1",
|
|
)
|
|
|
|
assert len(files) == 1
|
|
assert files[0].name == "report.txt"
|
|
|
|
|
|
class TestStoragePluginSaveFile:
|
|
"""Verify save_file writes uploaded content to disk."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_save_file_without_subpath(self, mock_get_cm):
|
|
"""File saved directly under category directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
|
|
upload_file = MagicMock()
|
|
upload_file.filename = "test_doc.pdf"
|
|
upload_file.content_type = "application/pdf"
|
|
mock_file_obj = MagicMock()
|
|
mock_file_obj.read.return_value = b""
|
|
upload_file.file = mock_file_obj
|
|
|
|
plugin = StoragePlugin()
|
|
result = await plugin.save_file(
|
|
file=upload_file,
|
|
category=FileCategory.BACKUP,
|
|
)
|
|
|
|
assert result.name == "test_doc.pdf"
|
|
assert result.category == FileCategory.BACKUP
|
|
assert result.mime_type == "application/pdf"
|
|
assert result.path.startswith("backups/")
|
|
assert isinstance(result, StoredFile)
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_save_file_with_subpath(self, mock_get_cm):
|
|
"""File saved under category/subpath."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
|
|
upload_file = MagicMock()
|
|
upload_file.filename = "report.xlsx"
|
|
upload_file.content_type = (
|
|
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
)
|
|
mock_file_obj = MagicMock()
|
|
mock_file_obj.read.return_value = b""
|
|
upload_file.file = mock_file_obj
|
|
|
|
plugin = StoragePlugin()
|
|
result = await plugin.save_file(
|
|
file=upload_file,
|
|
category=FileCategory.REPOSITORY,
|
|
subpath="archived/2024",
|
|
)
|
|
|
|
assert result.name == "report.xlsx"
|
|
assert result.category == FileCategory.REPOSITORY
|
|
assert "archived/2024" in result.path
|
|
|
|
|
|
class TestStoragePluginDeleteFile:
|
|
"""Verify delete_file removes files and directories."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_success(self, mock_get_cm):
|
|
"""Existing file is deleted."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
target = backup_dir / "to_delete.txt"
|
|
target.write_text("bye")
|
|
|
|
plugin = StoragePlugin()
|
|
await plugin.delete_file(FileCategory.BACKUP, "backups/to_delete.txt")
|
|
|
|
assert not target.exists()
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_delete_directory_success(self, mock_get_cm):
|
|
"""Existing directory is removed via shutil.rmtree."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
target_dir = backup_dir / "old_backup"
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
(target_dir / "inner.txt").write_text("inner")
|
|
|
|
plugin = StoragePlugin()
|
|
await plugin.delete_file(FileCategory.BACKUP, "backups/old_backup")
|
|
|
|
assert not target_dir.exists()
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_not_found(self, mock_get_cm):
|
|
"""Non-existent file raises FileNotFoundError."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
plugin = StoragePlugin()
|
|
with pytest.raises(FileNotFoundError, match="not found"):
|
|
await plugin.delete_file(FileCategory.BACKUP, "backups/ghost.txt")
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
@pytest.mark.asyncio
|
|
async def test_delete_file_category_mismatch(self, mock_get_cm):
|
|
"""Path not starting with category value raises ValueError."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
target = backup_dir / "file.txt"
|
|
target.write_text("data")
|
|
|
|
plugin = StoragePlugin()
|
|
with pytest.raises(ValueError, match="does not belong to category"):
|
|
await plugin.delete_file(FileCategory.BACKUP, "repositories/file.txt")
|
|
|
|
|
|
class TestStoragePluginGetFilePath:
|
|
"""Verify get_file_path returns valid paths."""
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_get_file_path_success(self, mock_get_cm):
|
|
"""Existing file returns resolved path."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
target = backup_dir / "download.txt"
|
|
target.write_text("content")
|
|
|
|
plugin = StoragePlugin()
|
|
result = plugin.get_file_path(FileCategory.BACKUP, "backups/download.txt")
|
|
|
|
assert result == target.resolve()
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_get_file_path_directory_raises(self, mock_get_cm):
|
|
"""Directory path raises FileNotFoundError (not a file)."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
plugin = StoragePlugin()
|
|
with pytest.raises(FileNotFoundError, match="not found"):
|
|
plugin.get_file_path(FileCategory.BACKUP, "backups")
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_get_file_path_not_found(self, mock_get_cm):
|
|
"""Non-existent file raises FileNotFoundError."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
plugin = StoragePlugin()
|
|
with pytest.raises(FileNotFoundError, match="not found"):
|
|
plugin.get_file_path(FileCategory.BACKUP, "backups/missing.txt")
|
|
|
|
@patch("src.plugins.storage.plugin.get_config_manager")
|
|
def test_get_file_path_category_mismatch(self, mock_get_cm):
|
|
"""Path not starting with category value raises ValueError."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
mock_get_cm.return_value = _mock_config_manager(storage_root=tmpdir)
|
|
backup_dir = Path(tmpdir) / "backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
plugin = StoragePlugin()
|
|
with pytest.raises(ValueError, match="does not belong to category"):
|
|
plugin.get_file_path(FileCategory.BACKUP, "repositories/download.txt")
|
|
|
|
|
|
class TestCopyUploadToDisk:
|
|
"""Verify the helper function _copy_upload_to_disk."""
|
|
|
|
def test_copy_upload_to_disk(self):
|
|
"""File content is copied from source to destination."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = Path(tmpdir) / "output.bin"
|
|
source = MagicMock()
|
|
# Return data once, then empty bytes to signal EOF
|
|
source.read = MagicMock(side_effect=[b"test data", b""])
|
|
|
|
_copy_upload_to_disk(dest, source)
|
|
|
|
assert dest.read_bytes() == b"test data"
|
|
|
|
def test_copy_upload_to_disk_empty(self):
|
|
"""Empty source file is handled."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
dest = Path(tmpdir) / "empty.bin"
|
|
source = MagicMock()
|
|
source.read = MagicMock(return_value=b"")
|
|
|
|
_copy_upload_to_disk(dest, source)
|
|
|
|
assert dest.read_bytes() == b""
|
|
# #endregion Test.StoragePlugin
|