Agent Zero
727d86614e
P0 (7): Auth-bypass removed, migrations fixed, plugin-upload disabled, RLS FORCE+WITH CHECK, plugin double-registration fixed, persistent volume, domain removed P1 (11): User/tenant model, Redis centralized, worker separated, transactional outbox, XSS fixed, DMS chunked streaming, permissions unified, password reset, metrics secured, config/docs fixed, cross-tenant FK P2 (4): Contact model normalized, cross-imports reduced 94%, commands+state machines for contacts/dms/mail/calendar, SPA path-traversal 8 new migrations, 99 unit tests, 13 commands, 8 contracts, 72 files changed
225 lines
7.2 KiB
Python
225 lines
7.2 KiB
Python
"""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 CHUNK_SIZE, _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"
|