leocrm/tests/test_storage.py

"""Tests for app/core/storage.py — validation, hashing, metadata, LocalStorage, factory."""

from __future__ import annotations

import asyncio
import hashlib
import os
import tempfile

import pytest

# Set test env BEFORE importing app modules
os.environ.setdefault("SECRET_KEY", "test-secret-key-with-at-least-32-characters-for-testing-only!!")
os.environ.setdefault("ENVIRONMENT", "testing")
os.environ.setdefault("SESSION_COOKIE_SECURE", "false")

from app.core.storage import (  # noqa: E402
    LocalStorage,
    compute_hash,
    get_file_metadata,
    get_storage_backend,
    reset_storage_backend,
    save_with_metadata,
    validate_mime,
    validate_size,
)


# ─── Fixtures ───


@pytest.fixture
def tmp_storage(tmp_path, monkeypatch):
    """Provide a LocalStorage backed by a temp directory."""
    monkeypatch.setenv("STORAGE_PATH", str(tmp_path))
    monkeypatch.setenv("STORAGE_BACKEND", "local")
    reset_storage_backend()
    backend = LocalStorage(base_path=str(tmp_path))
    yield backend
    reset_storage_backend()


# ─── 1. Path-Traversal Tests ───


class TestPathTraversal:
    @pytest.mark.asyncio
    async def test_save_traversal_relative_parent(self, tmp_storage):
        with pytest.raises(ValueError, match="[Pp]ath traversal"):
            await tmp_storage.save("../etc/passwd", b"data")

    @pytest.mark.asyncio
    async def test_save_traversal_double_parent(self, tmp_storage):
        with pytest.raises(ValueError, match="[Pp]ath traversal"):
            await tmp_storage.save("../../secret", b"data")

    @pytest.mark.asyncio
    async def test_save_normal_path_ok(self, tmp_storage):
        result = await tmp_storage.save("normal/path/file.txt", b"hello")
        assert result == "normal/path/file.txt"


# ─── 2. MIME Tests ───


class TestValidateMime:
    def test_text_plain_allowed(self):
        mime = validate_mime("file.txt", b"hello", ["text/plain"])
        assert mime == "text/plain"

    def test_exe_rejected(self):
        with pytest.raises(ValueError, match="not allowed"):
            validate_mime("file.exe", b"MZ\x90\x00", ["text/plain"])

    def test_pdf_default_allowlist(self):
        # PDF magic bytes
        mime = validate_mime("file.pdf", b"%PDF-1.4 test", None)
        assert mime is not None

    def test_empty_allowlist_all_allowed(self):
        mime = validate_mime("file.xyz", b"data", [])
        assert mime is not None

    def test_custom_allowlist_reject(self):
        with pytest.raises(ValueError, match="not allowed"):
            validate_mime("file.txt", b"hello", ["application/pdf"])


# ─── 3. Size-Limit Tests ───


class TestValidateSize:
    def test_small_file_ok(self):
        validate_size(b"x" * 100, max_size_mb=1)

    def test_oversized_file_raises(self):
        with pytest.raises(ValueError, match="exceeds limit"):
            validate_size(b"x" * (60 * 1024 * 1024), max_size_mb=50)

    def test_exact_limit_ok(self):
        # Exactly at the limit should pass (<=)
        validate_size(b"x" * (1024 * 1024), max_size_mb=1)

    def test_default_from_config(self):
        # Should use config default (50 MB) — small data passes
        validate_size(b"x" * 100)


# ─── 4. Hash Tests ───


class TestComputeHash:
    def test_sha256_known_value(self):
        result = compute_hash(b"hello")
        expected = hashlib.sha256(b"hello").hexdigest()
        assert result == expected
        assert result == "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"

    def test_md5_known_value(self):
        result = compute_hash(b"hello", algorithm="md5")
        expected = hashlib.md5(b"hello").hexdigest()
        assert result == expected
        assert result == "5d41402abc4b2a76b9719d911017c592"

    def test_empty_data(self):
        result = compute_hash(b"")
        assert result == hashlib.sha256(b"").hexdigest()

    def test_sha1(self):
        result = compute_hash(b"test", algorithm="sha1")
        assert result == hashlib.sha1(b"test").hexdigest()


# ─── 5. save_with_metadata Tests ───


class TestSaveWithMetadata:
    @pytest.mark.asyncio
    async def test_save_returns_full_metadata(self, tmp_storage, monkeypatch):
        monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        reset_storage_backend()
        # Patch the singleton to use our tmp backend
        import app.core.storage as storage_mod
        storage_mod._storage_backend = tmp_storage

        data = b"hello world"
        result = await save_with_metadata("test/file.txt", data)

        assert result["path"] == "test/file.txt"
        assert result["size"] == len(data)
        assert result["hash"] == compute_hash(data)
        assert "mime_type" in result
        assert result["storage_path"] == "test/file.txt"
        reset_storage_backend()

    @pytest.mark.asyncio
    async def test_save_oversized_raises(self, tmp_storage, monkeypatch):
        monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        reset_storage_backend()
        import app.core.storage as storage_mod
        storage_mod._storage_backend = tmp_storage

        big_data = b"x" * (60 * 1024 * 1024)
        with pytest.raises(ValueError, match="exceeds limit"):
            await save_with_metadata("big.txt", big_data, max_size_mb=50)
        reset_storage_backend()

    @pytest.mark.asyncio
    async def test_save_invalid_mime_raises(self, tmp_storage, monkeypatch):
        monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        reset_storage_backend()
        import app.core.storage as storage_mod
        storage_mod._storage_backend = tmp_storage

        with pytest.raises(ValueError, match="not allowed"):
            await save_with_metadata("file.xyz", b"data", allowed_mimes=["application/pdf"])
        reset_storage_backend()


# ─── 6. get_file_metadata Tests ───


class TestGetFileMetadata:
    def test_existing_file(self, tmp_storage, monkeypatch):
        monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        reset_storage_backend()
        import app.core.storage as storage_mod
        storage_mod._storage_backend = tmp_storage

        # Save a file first
        asyncio.run(tmp_storage.save("meta/test.txt", b"metadata test"))

        meta = get_file_metadata("meta/test.txt")
        assert meta["exists"] is True
        assert meta["size"] == len(b"metadata test")
        assert meta["modified"] is not None
        reset_storage_backend()

    def test_non_existing_file(self, tmp_storage, monkeypatch):
        monkeypatch.setenv("STORAGE_PATH", str(tmp_storage.base_path))
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        reset_storage_backend()
        import app.core.storage as storage_mod
        storage_mod._storage_backend = tmp_storage

        meta = get_file_metadata("nonexistent/file.txt")
        assert meta["exists"] is False
        assert meta["size"] is None
        assert meta["modified"] is None
        reset_storage_backend()


# ─── 7. LocalStorage Tests ───


class TestLocalStorage:
    @pytest.mark.asyncio
    async def test_save_read_delete_cycle(self, tmp_storage):
        data = b"cycle test data"
        path = await tmp_storage.save("cycle/test.txt", data)
        assert path == "cycle/test.txt"

        read_data = await tmp_storage.read("cycle/test.txt")
        assert read_data == data

        deleted = await tmp_storage.delete("cycle/test.txt")
        assert deleted is True

        deleted_again = await tmp_storage.delete("cycle/test.txt")
        assert deleted_again is False

    @pytest.mark.asyncio
    async def test_save_stream_read_cycle(self, tmp_storage):
        async def chunk_gen():
            yield b"chunk1-"
            yield b"chunk2-"
            yield b"chunk3"

        total = await tmp_storage.save_stream("stream/test.bin", chunk_gen())
        assert total == len(b"chunk1-chunk2-chunk3")

        read_data = await tmp_storage.read("stream/test.bin")
        assert read_data == b"chunk1-chunk2-chunk3"

    @pytest.mark.asyncio
    async def test_exists_check(self, tmp_storage):
        assert await tmp_storage.exists("exists/no.txt") is False
        await tmp_storage.save("exists/yes.txt", b"yes")
        assert await tmp_storage.exists("exists/yes.txt") is True

    @pytest.mark.asyncio
    async def test_list_files(self, tmp_storage):
        await tmp_storage.save("list/a.txt", b"a")
        await tmp_storage.save("list/sub/b.txt", b"b")
        files = await tmp_storage.list_files("list")
        assert len(files) == 2
        # Paths are relative to base_path
        assert any("a.txt" in f for f in files)
        assert any("b.txt" in f for f in files)


# ─── 8. Backend Factory Tests ───


class TestBackendFactory:
    def test_get_storage_backend_local_default(self, monkeypatch):
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp())
        reset_storage_backend()
        backend = get_storage_backend()
        assert isinstance(backend, LocalStorage)
        reset_storage_backend()

    def test_reset_storage_backend(self, monkeypatch):
        monkeypatch.setenv("STORAGE_BACKEND", "local")
        monkeypatch.setenv("STORAGE_PATH", tempfile.mkdtemp())
        reset_storage_backend()
        backend1 = get_storage_backend()
        reset_storage_backend()
        backend2 = get_storage_backend()
        assert backend1 is not backend2
        reset_storage_backend()