"""Unit tests for P1-6: DMS file processing — chunked streaming, SHA-256, sanitization. These tests verify the new functionality without requiring the full CSRF-protected HTTP stack. They test the helper functions and storage backend directly. """ from __future__ import annotations import hashlib import os import tempfile from unittest.mock import AsyncMock, MagicMock import pytest from app.core.storage import LocalStorage, StorageBackend from app.plugins.builtins.dms.routes import _sanitize_filename class TestSanitizeFilename: """Test the _sanitize_filename helper.""" def test_simple_filename(self): assert _sanitize_filename("document.pdf") == "document.pdf" def test_strips_path_separators(self): result = _sanitize_filename("../../etc/passwd") assert "/" not in result assert ".." not in result assert result == "passwd" def test_strips_backslashes(self): result = _sanitize_filename("..\\..\\windows\\system32") assert ".." not in result # On Linux, backslash is not a path separator, so it's stripped by the safe-filename regex assert "\\" not in result def test_strips_control_chars(self): result = _sanitize_filename("file\x00name.txt") assert "\x00" not in result assert "file" in result def test_empty_filename(self): assert _sanitize_filename("") == "file" def test_none_like_filename(self): assert _sanitize_filename(" ") == "file" def test_preserves_extension(self): result = _sanitize_filename("report.pdf") assert result.endswith(".pdf") def test_strips_leading_dots(self): result = _sanitize_filename(".hidden") assert not result.startswith(".") def test_collapses_multiple_dots(self): result = _sanitize_filename("file...txt") assert "..." not in result def test_collapses_multiple_spaces(self): result = _sanitize_filename("file name.pdf") assert " " not in result def test_truncates_long_filename(self): long_name = "a" * 250 + ".pdf" result = _sanitize_filename(long_name) assert len(result) <= 200 def test_dangerous_chars_removed(self): result = _sanitize_filename("file;rm -rf /.txt") assert ";" not in result assert "rm" not in result or result == "file-rm-rf.txt" class TestChunkSize: """Verify chunk size constant.""" def test_chunk_size_is_1mb(self): assert CHUNK_SIZE == 1024 * 1024 class TestLocalStorageStreaming: """Test LocalStorage.save_stream for chunked writes.""" @pytest.mark.asyncio async def test_save_stream_writes_all_chunks(self, tmp_path): storage = LocalStorage(base_path=str(tmp_path)) chunks = [b"chunk1_", b"chunk2_", b"chunk3"] async def chunk_iter(): for c in chunks: yield c total = await storage.save_stream("test/stream_file.bin", chunk_iter()) assert total == sum(len(c) for c in chunks) # Verify file content full_path = os.path.join(str(tmp_path), "test", "stream_file.bin") with open(full_path, "rb") as f: content = f.read() assert content == b"chunk1_chunk2_chunk3" @pytest.mark.asyncio async def test_save_stream_empty_file(self, tmp_path): storage = LocalStorage(base_path=str(tmp_path)) async def chunk_iter(): return yield # make it an async generator total = await storage.save_stream("empty.bin", chunk_iter()) assert total == 0 @pytest.mark.asyncio async def test_save_stream_creates_directories(self, tmp_path): storage = LocalStorage(base_path=str(tmp_path)) async def chunk_iter(): yield b"data" await storage.save_stream("deep/nested/path/file.bin", chunk_iter()) full_path = os.path.join(str(tmp_path), "deep", "nested", "path", "file.bin") assert os.path.exists(full_path) class TestStreamingHashIntegration: """Test that streaming produces correct SHA-256 hash.""" @pytest.mark.asyncio async def test_streaming_hash_matches_full_read(self, tmp_path): """Verify that chunked streaming produces the same SHA-256 as reading the full file.""" storage = LocalStorage(base_path=str(tmp_path)) content = b"x" * (CHUNK_SIZE * 2 + 12345) # ~2MB + some # Compute expected hash expected_hash = hashlib.sha256(content).hexdigest() # Simulate chunked upload hasher = hashlib.sha256() total_size = 0 async def chunk_iter(): nonlocal total_size offset = 0 while offset < len(content): chunk = content[offset : offset + CHUNK_SIZE] total_size += len(chunk) hasher.update(chunk) yield chunk offset += CHUNK_SIZE await storage.save_stream("hash_test.bin", chunk_iter()) assert total_size == len(content) assert hasher.hexdigest() == expected_hash @pytest.mark.asyncio async def test_streaming_hash_small_file(self, tmp_path): """Verify hash for a small file that fits in one chunk.""" storage = LocalStorage(base_path=str(tmp_path)) content = b"small file content" expected_hash = hashlib.sha256(content).hexdigest() hasher = hashlib.sha256() async def chunk_iter(): hasher.update(content) yield content await storage.save_stream("small.bin", chunk_iter()) assert hasher.hexdigest() == expected_hash class TestStoragePathNotInSchema: """Verify storage_path is not in the API response schema.""" def test_file_metadata_response_no_storage_path(self): from app.plugins.builtins.dms.schemas import FileMetadataResponse fields = FileMetadataResponse.model_fields assert "storage_path" not in fields assert "content_hash" in fields class TestModelHasContentHash: """Verify DmsFile model has content_hash column.""" def test_model_has_content_hash_column(self): from app.plugins.builtins.dms.models import File as DmsFile assert hasattr(DmsFile, "content_hash") col = DmsFile.__table__.columns.get("content_hash") assert col is not None assert col.type.length == 64 assert col.nullable is True class TestMigrationContentHash: """Verify migration 0038 exists and has correct revision chain.""" def test_migration_file_exists(self): path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "alembic", "versions", "0038_dms_content_hash.py", ) assert os.path.exists(path) def test_migration_revision_id(self): import importlib.util path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "alembic", "versions", "0038_dms_content_hash.py", ) spec = importlib.util.spec_from_file_location("migration_0038", path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) assert module.revision == "0038_dms_content_hash" assert module.down_revision == "0037_user_tenant_model"