"""Tests for file operations service.""" from pathlib import Path import pytest from basic_memory.services.exceptions import FileOperationError from basic_memory.services.file_service import FileService @pytest.mark.asyncio async def test_exists(tmp_path: Path, file_service: FileService): """Test file existence checking.""" # Test path test_path = tmp_path / "test.md" # Should not exist initially assert not await file_service.exists(test_path) # Create file test_path.write_text("test content") assert await file_service.exists(test_path) # Delete file test_path.unlink() assert not await file_service.exists(test_path) @pytest.mark.asyncio async def test_exists_error_handling(tmp_path: Path, file_service: FileService, monkeypatch): """Test error handling in exists() method.""" test_path = tmp_path / "test.md" def boom(*args, **kwargs): raise PermissionError("Access denied") monkeypatch.setattr(Path, "exists", boom) with pytest.raises(FileOperationError) as exc_info: await file_service.exists(test_path) assert "Failed to check file existence" in str(exc_info.value) @pytest.mark.asyncio async def test_write_read_file(tmp_path: Path, file_service: FileService): """Test basic write/read operations with checksums.""" test_path = tmp_path / "test.md" test_content = "test content\nwith multiple lines" # Write file and get checksum checksum = await file_service.write_file(test_path, test_content) assert test_path.exists() # Read back and verify content/checksum content, read_checksum = await file_service.read_file(test_path) assert content == test_content assert read_checksum == checksum @pytest.mark.asyncio async def test_write_creates_directories(tmp_path: Path, file_service: FileService): """Test directory creation on write.""" test_path = tmp_path / "subdir" / "nested" / "test.md" test_content = "test content" # Write should create directories await file_service.write_file(test_path, test_content) assert test_path.exists() assert test_path.parent.is_dir() @pytest.mark.asyncio async def test_write_atomic(tmp_path: Path, file_service: FileService, monkeypatch): """Test atomic write with no partial files.""" test_path = tmp_path / "test.md" temp_path = test_path.with_suffix(".tmp") from basic_memory import file_utils async def fake_write_file_atomic(*args, **kwargs): raise Exception("Write failed") monkeypatch.setattr(file_utils, "write_file_atomic", fake_write_file_atomic) # Attempt write that will fail with pytest.raises(FileOperationError): await file_service.write_file(test_path, "test content") # No partial files should exist assert not test_path.exists() assert not temp_path.exists() @pytest.mark.asyncio async def test_delete_file(tmp_path: Path, file_service: FileService): """Test file deletion.""" test_path = tmp_path / "test.md" test_content = "test content" # Create then delete await file_service.write_file(test_path, test_content) assert test_path.exists() await file_service.delete_file(test_path) assert not test_path.exists() # Delete non-existent file should not error await file_service.delete_file(test_path) @pytest.mark.asyncio async def test_checksum_consistency(tmp_path: Path, file_service: FileService): """Test checksum remains consistent.""" test_path = tmp_path / "test.md" test_content = "test content\n" * 10 # Get checksum from write checksum1 = await file_service.write_file(test_path, test_content) # Get checksum from read _, checksum2 = await file_service.read_file(test_path) # Write again and get new checksum checksum3 = await file_service.write_file(test_path, test_content) # All should match assert checksum1 == checksum2 == checksum3 @pytest.mark.asyncio async def test_error_handling_missing_file(tmp_path: Path, file_service: FileService): """Test error handling for missing files.""" test_path = tmp_path / "missing.md" with pytest.raises(FileOperationError): await file_service.read_file(test_path) @pytest.mark.asyncio async def test_error_handling_invalid_path(tmp_path: Path, file_service: FileService): """Test error handling for invalid paths.""" # Try to write to a directory instead of file test_path = tmp_path / "test.md" test_path.mkdir() # Create a directory instead of a file with pytest.raises(FileOperationError): await file_service.write_file(test_path, "test") @pytest.mark.asyncio async def test_write_unicode_content(tmp_path: Path, file_service: FileService): """Test handling of unicode content.""" test_path = tmp_path / "test.md" test_content = """ # Test Unicode - Emoji: 🚀 ⭐️ 🔥 - Chinese: 你好世界 - Arabic: مرحبا بالعالم - Russian: Привет, мир """ # Write and read back await file_service.write_file(test_path, test_content) content, _ = await file_service.read_file(test_path) assert content == test_content @pytest.mark.asyncio async def test_read_file_content(tmp_path: Path, file_service: FileService): """Test read_file_content returns just the content without checksum.""" test_path = tmp_path / "test.md" test_content = "test content\nwith multiple lines" # Write file await file_service.write_file(test_path, test_content) # Read content only content = await file_service.read_file_content(test_path) assert content == test_content @pytest.mark.asyncio async def test_read_file_content_missing_file(tmp_path: Path, file_service: FileService): """Test read_file_content raises error for missing files.""" test_path = tmp_path / "missing.md" # FileNotFoundError is preserved so callers can treat missing files specially (e.g. sync). with pytest.raises(FileNotFoundError): await file_service.read_file_content(test_path) @pytest.mark.asyncio async def test_read_file_content_raises_file_operation_error_for_directory( tmp_path: Path, file_service: FileService ): """read_file_content should wrap non-FileNotFound errors in FileOperationError.""" dir_path = tmp_path / "not-a-file" dir_path.mkdir() with pytest.raises(FileOperationError) as exc_info: await file_service.read_file_content(dir_path) assert "Failed to read file" in str(exc_info.value) @pytest.mark.asyncio async def test_read_file_bytes(tmp_path: Path, file_service: FileService): """Test read_file_bytes for binary file reading.""" test_path = tmp_path / "test.bin" # Create binary content with non-UTF8 bytes binary_content = b"\x00\x01\x02\x03\xff\xfe\xfd" # Write binary file directly test_path.write_bytes(binary_content) # Read back using read_file_bytes content = await file_service.read_file_bytes(test_path) assert content == binary_content @pytest.mark.asyncio async def test_read_file_bytes_image(tmp_path: Path, file_service: FileService): """Test read_file_bytes with image-like binary content.""" test_path = tmp_path / "test.png" # PNG header signature png_header = b"\x89PNG\r\n\x1a\n" fake_image_content = png_header + b"\x00" * 100 test_path.write_bytes(fake_image_content) content = await file_service.read_file_bytes(test_path) assert content == fake_image_content assert content.startswith(png_header) @pytest.mark.asyncio async def test_read_file_bytes_missing_file(tmp_path: Path, file_service: FileService): """Test read_file_bytes raises error for missing files.""" test_path = tmp_path / "missing.bin" with pytest.raises(FileOperationError): await file_service.read_file_bytes(test_path) @pytest.mark.asyncio async def test_read_file_bytes_text_file(tmp_path: Path, file_service: FileService): """Test read_file_bytes can read text files as bytes.""" test_path = tmp_path / "test.txt" text_content = "Hello, World!" test_path.write_text(text_content) content = await file_service.read_file_bytes(test_path) assert content == text_content.encode("utf-8")