leocrm/tests/test_p1_6_dms_streaming.py

"""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"