feat: add read only document mode
This commit is contained in:
parent
b7c8d7368c
commit
1c4a0fe54f
4 changed files with 255 additions and 1 deletions
|
|
@ -11,6 +11,12 @@ class SiteNotSpecifiedError(Exception):
|
|||
super(Exception, self).__init__(self.message)
|
||||
|
||||
|
||||
class DatabaseModificationError(Exception):
|
||||
"""Error raised when attempting to modify the database in a read-only document context."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class UrlSchemeNotSupported(Exception):
|
||||
pass
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,9 @@ import hashlib
|
|||
import json
|
||||
import time
|
||||
from collections.abc import Generator, Iterable
|
||||
from typing import TYPE_CHECKING, Any, Optional
|
||||
from contextlib import contextmanager
|
||||
from functools import wraps
|
||||
from typing import TYPE_CHECKING, Any, Literal, Optional, TypeAlias, Union, overload
|
||||
|
||||
from werkzeug.exceptions import NotFound
|
||||
|
||||
|
|
@ -89,6 +91,61 @@ def get_doc(*args, **kwargs):
|
|||
raise ImportError(doctype)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def read_only_document(context=None):
|
||||
# Store original methods
|
||||
original_methods = {
|
||||
"save": Document.save,
|
||||
"_save": Document._save,
|
||||
"insert": Document.insert,
|
||||
"delete": Document.delete,
|
||||
"submit": Document.submit,
|
||||
"cancel": Document.cancel,
|
||||
"db_set": Document.db_set,
|
||||
}
|
||||
|
||||
def read_only_method(func):
|
||||
@wraps(func)
|
||||
def wrapper(self, *args, **kwargs):
|
||||
if self.doctype == "Error Log" and func.__name__ == "insert":
|
||||
return original_methods["insert"](self, *args, **kwargs)
|
||||
error_msg = f"Cannot call {func.__name__} in read-only document mode"
|
||||
if context:
|
||||
error_msg += f" ({context})"
|
||||
raise frappe.DatabaseModificationError(error_msg)
|
||||
|
||||
return wrapper
|
||||
|
||||
# Use a thread-local variable to track nested invocations
|
||||
if not hasattr(frappe.local, "read_only_depth"):
|
||||
frappe.local.read_only_depth = 0
|
||||
|
||||
try:
|
||||
# Increment the depth counter
|
||||
frappe.local.read_only_depth += 1
|
||||
|
||||
# Only apply read-only methods if this is the outermost invocation
|
||||
if frappe.local.read_only_depth == 1:
|
||||
# Replace methods with read-only versions
|
||||
for method_name, method in original_methods.items():
|
||||
setattr(Document, method_name, read_only_method(method))
|
||||
|
||||
yield
|
||||
|
||||
finally:
|
||||
# Decrement the depth counter
|
||||
frappe.local.read_only_depth -= 1
|
||||
|
||||
# Only restore original methods if this is the outermost invocation
|
||||
if frappe.local.read_only_depth == 0:
|
||||
# Restore original methods
|
||||
for method_name, method in original_methods.items():
|
||||
setattr(Document, method_name, method)
|
||||
|
||||
# Clean up the thread-local variable
|
||||
del frappe.local.read_only_depth
|
||||
|
||||
|
||||
class Document(BaseDocument):
|
||||
"""All controllers inherit from `Document`."""
|
||||
|
||||
|
|
|
|||
191
frappe/tests/test_document_ro_mode.py
Normal file
191
frappe/tests/test_document_ro_mode.py
Normal file
|
|
@ -0,0 +1,191 @@
|
|||
import unittest
|
||||
from contextlib import contextmanager
|
||||
|
||||
import frappe
|
||||
from frappe.model.document import Document, read_only_document
|
||||
from frappe.tests.utils import FrappeTestCase
|
||||
|
||||
|
||||
class TestReadOnlyDocument(FrappeTestCase):
|
||||
def setUp(self):
|
||||
# Create a test document
|
||||
self.test_doc = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo"})
|
||||
self.test_doc.insert()
|
||||
|
||||
def tearDown(self):
|
||||
# Delete the test document
|
||||
frappe.delete_doc("ToDo", self.test_doc.name)
|
||||
|
||||
def test_read_only_save(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.save()
|
||||
|
||||
def test_read_only_insert(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
frappe.get_doc({"doctype": "ToDo", "description": "Another Test ToDo"}).insert()
|
||||
|
||||
def test_read_only_delete(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.delete()
|
||||
|
||||
def test_read_only_submit(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.submit()
|
||||
|
||||
def test_read_only_cancel(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.cancel()
|
||||
|
||||
def test_read_only_db_set(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.db_set("status", "Closed")
|
||||
|
||||
def test_read_only_nested_calls(self):
|
||||
def nested_save():
|
||||
self.test_doc.save()
|
||||
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
nested_save()
|
||||
|
||||
def test_read_only_context_manager_restoration(self):
|
||||
original_save = Document.save
|
||||
|
||||
with read_only_document():
|
||||
self.assertNotEqual(Document.save, original_save)
|
||||
|
||||
self.assertEqual(Document.save, original_save)
|
||||
|
||||
def test_nested_read_only_document(self):
|
||||
# Check that read_only_depth is not set initially
|
||||
self.assertFalse(hasattr(frappe.local, "read_only_depth"))
|
||||
|
||||
with read_only_document():
|
||||
# Check that read_only_depth is set to 1
|
||||
self.assertEqual(frappe.local.read_only_depth, 1)
|
||||
|
||||
# Attempt to modify the document
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.description = "Modified in outer context"
|
||||
self.test_doc.save()
|
||||
|
||||
with read_only_document():
|
||||
# Check that read_only_depth is incremented to 2
|
||||
self.assertEqual(frappe.local.read_only_depth, 2)
|
||||
|
||||
# Attempt to modify the document in nested context
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.description = "Modified in inner context"
|
||||
self.test_doc.save()
|
||||
|
||||
# Check that read_only_depth is back to 1 after nested context
|
||||
self.assertEqual(frappe.local.read_only_depth, 1)
|
||||
|
||||
# Attempt to modify the document again
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.description = "Modified after inner context"
|
||||
self.test_doc.save()
|
||||
|
||||
# Check that read_only_depth is removed after all contexts are closed
|
||||
self.assertFalse(hasattr(frappe.local, "read_only_depth"))
|
||||
|
||||
# Verify that document can be modified outside read_only_document
|
||||
self.test_doc.description = "Modified outside read_only_document"
|
||||
self.test_doc.save()
|
||||
self.assertEqual(self.test_doc.description, "Modified outside read_only_document")
|
||||
|
||||
def test_error_log_exception_in_read_only(self):
|
||||
with read_only_document():
|
||||
# Attempt to insert an Error Log
|
||||
error_log = frappe.get_doc({"doctype": "Error Log", "error": "Test error in read-only mode"})
|
||||
|
||||
# This should not raise an exception
|
||||
error_log.insert()
|
||||
|
||||
# Verify that the Error Log was inserted
|
||||
self.assertTrue(error_log.name)
|
||||
|
||||
# Attempt to modify a different document
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
self.test_doc.description = "Modified in read-only mode"
|
||||
self.test_doc.save()
|
||||
|
||||
# Clean up the inserted Error Log
|
||||
frappe.delete_doc("Error Log", error_log.name)
|
||||
|
||||
def test_read_only_multiple_documents(self):
|
||||
doc1 = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo 1"})
|
||||
doc2 = frappe.get_doc({"doctype": "ToDo", "description": "Test ToDo 2"})
|
||||
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
doc1.insert()
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
doc2.insert()
|
||||
|
||||
def test_read_only_custom_method(self):
|
||||
class CustomDocument(Document):
|
||||
def custom_save(self):
|
||||
self.save()
|
||||
|
||||
custom_doc = CustomDocument({"doctype": "ToDo", "description": "Custom Test ToDo"})
|
||||
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
custom_doc.custom_save()
|
||||
|
||||
def test_read_only_exception_handling(self):
|
||||
@contextmanager
|
||||
def exception_raiser():
|
||||
raise Exception("Test exception")
|
||||
yield
|
||||
|
||||
try:
|
||||
with read_only_document(), exception_raiser():
|
||||
pass
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Ensure methods are restored even if an exception occurs
|
||||
self.assertEqual(Document.save, self.test_doc.__class__.save)
|
||||
|
||||
def test_read_only_nested_context_managers(self):
|
||||
original_save = Document.save
|
||||
|
||||
with read_only_document():
|
||||
self.assertNotEqual(Document.save, original_save)
|
||||
|
||||
with read_only_document():
|
||||
self.assertNotEqual(Document.save, original_save)
|
||||
|
||||
self.assertNotEqual(Document.save, original_save)
|
||||
|
||||
self.assertEqual(Document.save, original_save)
|
||||
|
||||
def test_read_only_method_call_details(self):
|
||||
with read_only_document():
|
||||
with self.assertRaises(frappe.DatabaseModificationError) as cm:
|
||||
self.test_doc.save()
|
||||
|
||||
self.assertIn("Cannot call save in read-only document mode", str(cm.exception))
|
||||
|
||||
def test_read_only_does_not_affect_reads(self):
|
||||
with read_only_document():
|
||||
# These operations should not raise exceptions
|
||||
doc = frappe.get_doc("ToDo", self.test_doc.name)
|
||||
self.assertEqual(doc.description, "Test ToDo")
|
||||
|
||||
docs = frappe.get_all("ToDo", filters={"name": self.test_doc.name})
|
||||
self.assertEqual(len(docs), 1)
|
||||
|
||||
def test_read_only_with_new_document_instance(self):
|
||||
with read_only_document():
|
||||
new_doc = frappe.new_doc("ToDo")
|
||||
with self.assertRaises(frappe.DatabaseModificationError):
|
||||
new_doc.insert()
|
||||
0
tests/test_document_ro_mode.py
Normal file
0
tests/test_document_ro_mode.py
Normal file
Loading…
Add table
Reference in a new issue