diff --git a/frappe/exceptions.py b/frappe/exceptions.py index 1f74dcc485..3f59b7a865 100644 --- a/frappe/exceptions.py +++ b/frappe/exceptions.py @@ -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 diff --git a/frappe/model/document.py b/frappe/model/document.py index a782f7c21c..7def467f1c 100644 --- a/frappe/model/document.py +++ b/frappe/model/document.py @@ -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`.""" diff --git a/frappe/tests/test_document_ro_mode.py b/frappe/tests/test_document_ro_mode.py new file mode 100644 index 0000000000..bf0767d094 --- /dev/null +++ b/frappe/tests/test_document_ro_mode.py @@ -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() diff --git a/tests/test_document_ro_mode.py b/tests/test_document_ro_mode.py new file mode 100644 index 0000000000..e69de29bb2