mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
remove permission test
This commit is contained in:
@@ -1,17 +1,16 @@
|
||||
"""Tests for DocumentService."""
|
||||
|
||||
import os
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
import yaml
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
import stat
|
||||
|
||||
from basic_memory.services.document_service import (
|
||||
DocumentService,
|
||||
DocumentNotFoundError,
|
||||
DocumentWriteError
|
||||
DocumentWriteError,
|
||||
)
|
||||
|
||||
|
||||
@@ -33,15 +32,11 @@ async def test_doc_path(tmp_path) -> Path:
|
||||
async def test_create_document_with_frontmatter(document_service, test_doc_path):
|
||||
"""Test that created documents have proper frontmatter."""
|
||||
content = "# Test Document\n\nThis is a test."
|
||||
doc = await document_service.create_document(
|
||||
str(test_doc_path),
|
||||
content,
|
||||
{"type": "test"}
|
||||
)
|
||||
doc = await document_service.create_document(str(test_doc_path), content, {"type": "test"})
|
||||
|
||||
# Verify file content
|
||||
file_content = test_doc_path.read_text()
|
||||
|
||||
|
||||
# Parse frontmatter
|
||||
try:
|
||||
# Split content at the second "---" marker
|
||||
@@ -55,11 +50,11 @@ async def test_create_document_with_frontmatter(document_service, test_doc_path)
|
||||
assert metadata["type"] == "test"
|
||||
assert "created" in metadata
|
||||
assert "modified" in metadata
|
||||
|
||||
|
||||
# Verify timestamps are valid ISO format
|
||||
datetime.fromisoformat(metadata["created"])
|
||||
datetime.fromisoformat(metadata["modified"])
|
||||
|
||||
|
||||
# Verify original content is preserved
|
||||
assert content in doc_content
|
||||
|
||||
@@ -70,58 +65,33 @@ async def test_create_document_with_frontmatter(document_service, test_doc_path)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_document_unwriteable_directory(document_service, tmp_path):
|
||||
"""Test error when trying to write to an unwriteable directory."""
|
||||
# Create parent directory without write permissions
|
||||
parent_dir = tmp_path / "unwriteable"
|
||||
parent_dir.mkdir()
|
||||
parent_dir.chmod(stat.S_IREAD) # Read-only
|
||||
|
||||
bad_path = parent_dir / "test.md"
|
||||
content = "# Test"
|
||||
|
||||
with pytest.raises(DocumentWriteError):
|
||||
await document_service.create_document(str(bad_path), content)
|
||||
|
||||
# Verify directory is still read-only
|
||||
assert not os.access(parent_dir, os.W_OK)
|
||||
# Verify no database record
|
||||
doc = await document_service.repository.find_by_path(str(bad_path))
|
||||
assert doc is None
|
||||
|
||||
# Clean up - make writable again so it can be deleted
|
||||
parent_dir.chmod(stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_document_cleanup_on_write_failure(document_service, test_doc_path):
|
||||
"""Test that failed document creation cleans up DB record when file write fails."""
|
||||
|
||||
def fail_write(self, content):
|
||||
raise PermissionError("Mock write failure")
|
||||
|
||||
|
||||
# Store original write_text method
|
||||
original_write_text = Path.write_text
|
||||
|
||||
|
||||
try:
|
||||
# Replace write_text with our failing version
|
||||
Path.write_text = fail_write
|
||||
|
||||
|
||||
with pytest.raises(DocumentWriteError) as exc_info:
|
||||
await document_service.create_document(
|
||||
str(test_doc_path),
|
||||
"test content",
|
||||
{"type": "test"}
|
||||
str(test_doc_path), "test content", {"type": "test"}
|
||||
)
|
||||
|
||||
|
||||
assert "Mock write failure" in str(exc_info.value)
|
||||
|
||||
|
||||
# Verify DB record was cleaned up
|
||||
doc = await document_service.repository.find_by_path(str(test_doc_path))
|
||||
assert doc is None
|
||||
|
||||
|
||||
# Verify no file was created
|
||||
assert not test_doc_path.exists()
|
||||
|
||||
|
||||
finally:
|
||||
# Restore original write_text method
|
||||
Path.write_text = original_write_text
|
||||
@@ -138,10 +108,7 @@ async def test_delete_nonexistent_file(document_service, test_doc_path):
|
||||
async def test_update_nonexistent_file(document_service, test_doc_path):
|
||||
"""Test updating a file that doesn't exist."""
|
||||
with pytest.raises(DocumentNotFoundError):
|
||||
await document_service.update_document(
|
||||
str(test_doc_path),
|
||||
"new content"
|
||||
)
|
||||
await document_service.update_document(str(test_doc_path), "new content")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -152,7 +119,7 @@ async def test_read_file_exists_no_record(document_service, test_doc_path):
|
||||
|
||||
# Reading should create the record
|
||||
doc, read_content = await document_service.read_document(str(test_doc_path))
|
||||
|
||||
|
||||
assert read_content == content
|
||||
assert doc is not None
|
||||
assert doc.path == str(test_doc_path)
|
||||
@@ -174,12 +141,10 @@ async def test_update_file_exists_no_record(document_service, test_doc_path):
|
||||
# Update should create record
|
||||
new_content = "updated content"
|
||||
doc = await document_service.update_document(
|
||||
str(test_doc_path),
|
||||
new_content,
|
||||
{"status": "updated"}
|
||||
str(test_doc_path), new_content, {"status": "updated"}
|
||||
)
|
||||
|
||||
assert doc is not None
|
||||
assert doc.path == str(test_doc_path)
|
||||
assert doc.doc_metadata == {"status": "updated"}
|
||||
assert test_doc_path.read_text() == new_content
|
||||
assert test_doc_path.read_text() == new_content
|
||||
|
||||
Reference in New Issue
Block a user