feat: global frappe.in_test flag (#32960)

* feat: global `frappe.in_test` flag

* feat: helper utility to toggle `frappe.in_test`

* fix: use `toggle_test_mode` util

* fix: use `frappe.in_test`

* chore: add comment explaining global `in_test`

* chore: ignore commit replacing flag usage

* test: temporarily disable `frappe.in_test`

this worked earlier because flag was set in werkzeug.local which was separate for API test client

* test: add comment explaining change

.git-blame-ignore-revs

@@ -58,3 +58,6 @@ e9bbe03354079cfcef65a77b0c33f57b047a7c93
 
 # ruff update
 84ef6ec677c8657c3243ac456a1ef794bfb34a50
+
+# replace `frappe.flags.in_test` with `frappe.in_test`
+653c80b8483cc41aef25cd7d66b9b6bb188bf5f8

frappe/__init__.py

@@ -83,6 +83,9 @@ cache: Optional["RedisWrapper"] = None
 client_cache: Optional["ClientCache"] = None
 STANDARD_USERS = ("Guest", "Administrator")
 
+# this global may be subsequently changed by frappe.tests.utils.toggle_test_mode()
+in_test = False
+
 _dev_server = int(sbool(os.environ.get("DEV_SERVER", False)))
 
 if _dev_server:
@@ -219,7 +222,7 @@ def init(site: str, sites_path: str = ".", new_site: bool = False, force: bool =
 			"in_install_db": False,
 			"in_install_app": False,
 			"in_import": False,
-			"in_test": False,
+			"in_test": in_test,
 			"mute_messages": False,
 			"ignore_links": False,
 			"mute_emails": False,
@@ -640,7 +643,7 @@ def whitelist(allow_guest=False, xss_safe=False, methods=None):
 		global whitelisted, guest_methods, xss_safe_methods, allowed_http_methods_for_whitelisted_func
 
 		# validate argument types only if request is present
-		in_request_or_test = lambda: getattr(local, "request", None) or local.flags.in_test  # noqa: E731
+		in_request_or_test = lambda: getattr(local, "request", None) or in_test  # noqa: E731
 
 		# get function from the unbound / bound method
 		# this is needed because functions can be compared, but not methods
@@ -740,7 +743,7 @@ def only_for(roles: list[str] | tuple[str] | str, message=False):
 	:param roles: Permitted role(s)
 	"""
 
-	if local.flags.in_test or local.session.user == "Administrator":
+	if in_test or local.session.user == "Administrator":
 		return
 
 	if isinstance(roles, str):
@@ -767,7 +770,7 @@ def get_domain_data(module):
 		else:
 			return _dict()
 	except ImportError:
-		if local.flags.in_test:
+		if in_test:
 			return _dict()
 		else:
 			raise
@@ -1595,7 +1598,7 @@ def copy_doc(doc: "Document", ignore_no_copy: bool = True) -> "Document":
 
 	fields_to_clear = ["name", "owner", "creation", "modified", "modified_by"]
 
-	if not local.flags.in_test:
+	if not in_test:
 		fields_to_clear.append("docstatus")
 
 	if isinstance(doc, BaseDocument) or hasattr(doc, "as_dict"):

frappe/automation/doctype/auto_repeat/auto_repeat.py

@@ -86,7 +86,7 @@ class AutoRepeat(Document):
 		validate_template(self.message or "")
 
 	def before_insert(self):
-		if not frappe.flags.in_test:
+		if not frappe.in_test:
 			start_date = getdate(self.start_date)
 			today_date = getdate(today())
 			if start_date <= today_date:
@@ -112,7 +112,7 @@ class AutoRepeat(Document):
 			frappe.db.set_value(self.reference_doctype, self.reference_document, "auto_repeat", "")
 
 	def validate_reference_doctype(self):
-		if frappe.flags.in_test or frappe.flags.in_patch:
+		if frappe.in_test or frappe.flags.in_patch:
 			return
 		if not frappe.get_meta(self.reference_doctype).allow_auto_repeat:
 			frappe.throw(
@@ -229,7 +229,7 @@ class AutoRepeat(Document):
 
 			self.disable_auto_repeat()
 
-			if self.reference_document and not frappe.flags.in_test:
+			if self.reference_document and not frappe.in_test:
 				self.notify_error_to_user(error_log)
 
 	def make_new_document(self):

frappe/core/doctype/access_log/access_log.py

@@ -81,7 +81,7 @@ def make_access_log(
 	# `frappe.db.commit` added because insert doesnt `commit` when called in GET requests like `printview`
 	# dont commit in test mode. It must be tempting to put this block along with the in_request in the
 	# whitelisted method...yeah, don't do it. That part would be executed possibly on a read only DB conn
-	if not frappe.flags.in_test or in_request:
+	if not frappe.in_test or in_request:
 		frappe.db.commit()
 
 

frappe/core/doctype/data_import/data_import.py

@@ -101,7 +101,7 @@ class DataImport(Document):
 	def start_import(self):
 		from frappe.utils.scheduler import is_scheduler_inactive
 
-		run_now = frappe.flags.in_test or frappe.conf.developer_mode
+		run_now = frappe.in_test or frappe.conf.developer_mode
 		if is_scheduler_inactive() and not run_now:
 			frappe.throw(_("Scheduler is inactive. Cannot import data."), title=_("Scheduler Inactive"))
 

frappe/core/doctype/doctype/doctype.py

@@ -322,7 +322,7 @@ class DocType(Document):
 
 	def check_developer_mode(self):
 		"""Throw exception if not developer mode or via patch"""
-		if frappe.flags.in_patch or frappe.flags.in_test:
+		if frappe.flags.in_patch or frappe.in_test:
 			return
 
 		if not frappe.conf.get("developer_mode") and not self.custom:
@@ -594,7 +594,7 @@ class DocType(Document):
 			global_search_fields_after_update.append("name")
 
 		if set(global_search_fields_before_update) != set(global_search_fields_after_update):
-			now = (not frappe.request) or frappe.flags.in_test or frappe.flags.in_install
+			now = (not frappe.request) or frappe.in_test or frappe.flags.in_install
 			frappe.enqueue("frappe.utils.global_search.rebuild_for_doctype", now=now, doctype=self.name)
 
 	def set_base_class_for_controller(self):

frappe/core/doctype/role_profile/role_profile.py

@@ -29,7 +29,7 @@ class RoleProfile(Document):
 		self.clear_cache()
 		self.queue_action(
 			"update_all_users",
-			now=frappe.flags.in_test or frappe.flags.in_install,
+			now=frappe.in_test or frappe.flags.in_install,
 			enqueue_after_commit=True,
 			queue="long",
 		)

frappe/core/doctype/user/test_user.py

@@ -25,6 +25,7 @@ from frappe.model.delete_doc import delete_doc
 from frappe.tests import IntegrationTestCase
 from frappe.tests.classes.context_managers import change_settings
 from frappe.tests.test_api import FrappeAPITestCase
+from frappe.tests.utils import toggle_test_mode
 from frappe.utils import get_url
 
 user_module = frappe.core.doctype.user.user
@@ -212,13 +213,15 @@ class TestUser(IntegrationTestCase):
 
 			# test password strength while saving user with new password
 			user = frappe.get_doc("User", "test@example.com")
-			frappe.flags.in_test = False
+			toggle_test_mode(False)
+			try:
 				user.new_password = "password"
 				self.assertRaises(frappe.exceptions.ValidationError, user.save)
 				user.reload()
 				user.new_password = "Eastern_43A1W"
 				user.save()
-			frappe.flags.in_test = True
+			finally:
+				toggle_test_mode(True)
 
 	def test_comment_mentions(self):
 		comment = """

frappe/core/doctype/user/user.py

@@ -173,7 +173,7 @@ class User(Document):
 		self.__new_password = self.new_password
 		self.new_password = ""
 
-		if not frappe.flags.in_test:
+		if not frappe.in_test:
 			self.password_strength_test()
 
 		if self.name not in STANDARD_USERS:
@@ -269,7 +269,7 @@ class User(Document):
 		self.share_with_self()
 		clear_notifications(user=self.name)
 		frappe.clear_cache(user=self.name)
-		now = frappe.flags.in_test or frappe.flags.in_install
+		now = frappe.in_test or frappe.flags.in_install
 		self.send_password_notification(self.__new_password)
 		frappe.enqueue(
 			"frappe.core.doctype.user.user.create_contact",

frappe/desk/doctype/notification_log/notification_log.py

@@ -94,8 +94,8 @@ def enqueue_create_notification(users: list[str] | str, doc: dict):
 		"frappe.desk.doctype.notification_log.notification_log.make_notification_logs",
 		doc=doc,
 		users=users,
-		now=frappe.flags.in_test,
-		enqueue_after_commit=not frappe.flags.in_test,
+		now=frappe.in_test,
+		enqueue_after_commit=not frappe.in_test,
 	)
 
 
@@ -141,7 +141,7 @@ def send_notification_email(doc: NotificationLog):
 		template="new_notification",
 		args=args,
 		header=[header, "orange"],
-		now=frappe.flags.in_test,
+		now=frappe.in_test,
 	)
 
 

frappe/desk/doctype/system_health_report/system_health_report.py

@@ -57,7 +57,7 @@ def health_check(step: str):
 			try:
 				return func(*args, **kwargs)
 			except Exception as e:
-				if frappe.flags.in_test:
+				if frappe.in_test:
 					raise
 				frappe.log(frappe.get_traceback())
 				# nosemgrep

frappe/desk/doctype/workspace/workspace.py

@@ -235,7 +235,7 @@ def disable_saving_as_public():
 		frappe.flags.in_install
 		or frappe.flags.in_uninstall
 		or frappe.flags.in_patch
-		or frappe.flags.in_test
+		or frappe.in_test
 		or frappe.flags.in_fixtures
 		or frappe.flags.in_migrate
 	)

frappe/desk/page/setup_wizard/setup_wizard.py

@@ -244,7 +244,7 @@ def update_system_settings(args):  # nosemgrep
 			"date_format": frappe.db.get_value("Country", args.get("country"), "date_format"),
 			"time_format": frappe.db.get_value("Country", args.get("country"), "time_format"),
 			"number_format": number_format,
-			"enable_scheduler": 1 if not frappe.flags.in_test else 0,
+			"enable_scheduler": 1 if not frappe.in_test else 0,
 			"backup_limit": 3,  # Default for downloadable backups
 			"enable_telemetry": cint(args.get("enable_telemetry")),
 		}

frappe/email/doctype/email_account/email_account.py

@@ -159,7 +159,7 @@ class EmailAccount(Document):
 		if self.enable_incoming and self.use_imap and len(self.imap_folder) <= 0:
 			frappe.throw(_("You need to set one IMAP folder for {0}").format(frappe.bold(self.email_id)))
 
-		if frappe.local.flags.in_patch or frappe.local.flags.in_test:
+		if frappe.local.flags.in_patch or frappe.in_test:
 			return
 
 		use_oauth = self.auth_method == "OAuth"
@@ -363,9 +363,7 @@ class EmailAccount(Document):
 
 	@property
 	def _password(self):
-		raise_exception = not (
-			self.auth_method == "OAuth" or self.no_smtp_authentication or frappe.flags.in_test
-		)
+		raise_exception = not (self.auth_method == "OAuth" or self.no_smtp_authentication or frappe.in_test)
 		return self.get_password(raise_exception=raise_exception)
 
 	@property
@@ -565,7 +563,7 @@ class EmailAccount(Document):
 			self.set_failed_attempts_count(self.get_failed_attempts_count() + 1)
 
 	def _disable_broken_incoming_account(self, description):
-		if frappe.flags.in_test:
+		if frappe.in_test:
 			return
 		self.db_set("enable_incoming", 0)
 

frappe/email/doctype/email_domain/email_domain.py

@@ -81,7 +81,7 @@ class EmailDomain(Document):
 	def validate(self):
 		"""Validate POP3/IMAP and SMTP connections."""
 
-		if frappe.local.flags.in_patch or frappe.local.flags.in_test or frappe.local.flags.in_install:
+		if frappe.local.flags.in_patch or frappe.in_test or frappe.local.flags.in_install:
 			return
 
 		self.validate_incoming_server_conn()

frappe/email/doctype/email_queue/email_queue.py

@@ -182,7 +182,7 @@ class EmailQueue(Document):
 				message = ctx.build_message(recipient.recipient)
 				if method := get_hook_method("override_email_send"):
 					method(self, self.sender, recipient.recipient, message)
-				elif not frappe.flags.in_test or frappe.flags.testing_email:
+				elif not frappe.in_test or frappe.flags.testing_email:
 					if ctx.email_account_doc.service == "Frappe Mail":
 						is_newsletter = self.reference_doctype == "Newsletter"
 						ctx.frappe_mail_client.send_raw(
@@ -200,7 +200,7 @@ class EmailQueue(Document):
 
 				ctx.update_recipient_status_to_sent(recipient)
 
-			if frappe.flags.in_test and not frappe.flags.testing_email:
+			if frappe.in_test and not frappe.flags.testing_email:
 				frappe.flags.sent_mail = message
 				return
 
@@ -773,7 +773,7 @@ class QueueBuilder:
 					job_name=frappe.utils.get_job_name(
 						"send_bulk_emails_for", self.reference_doctype, self.reference_name
 					),
-					now=frappe.flags.in_test or send_now,
+					now=frappe.in_test or send_now,
 					queue="long",
 				)
 

frappe/email/doctype/newsletter/newsletter.py

@@ -210,7 +210,7 @@ class Newsletter(WebsiteGenerator):
 		args["message"] = self.get_message(medium="email")
 
 		is_auto_commit_set = bool(frappe.db.auto_commit_on_many_writes)
-		frappe.db.auto_commit_on_many_writes = not frappe.flags.in_test
+		frappe.db.auto_commit_on_many_writes = not frappe.in_test
 
 		frappe.sendmail(
 			subject=self.subject,
@@ -421,7 +421,7 @@ def send_scheduled_email():
 			frappe.db.set_value("Newsletter", newsletter_name, "email_sent", 0)
 			newsletter.log_error("Failed to send newsletter")
 
-		if not frappe.flags.in_test:
+		if not frappe.in_test:
 			frappe.db.commit()
 
 	frappe.flags.is_scheduler_running = False

frappe/email/doctype/notification/notification.py

@@ -322,7 +322,7 @@ def get_context(context):
 			"frappe.email.doctype.notification.notification.evaluate_alert",
 			doc=doc,
 			alert=self,
-			now=frappe.flags.in_test,
+			now=frappe.in_test,
 			enqueue_after_commit=enqueue_after_commit,
 		)
 

frappe/email/queue.py

@@ -95,7 +95,7 @@ def get_unsubcribed_url(reference_doctype, reference_name, email, unsubscribe_me
 @frappe.whitelist(allow_guest=True)
 def unsubscribe(doctype, name, email):
 	# unsubsribe from comments and communications
-	if not frappe.flags.in_test and not verify_request():
+	if not frappe.in_test and not verify_request():
 		return
 
 	try:

frappe/email/receive.py

@@ -632,7 +632,7 @@ class InboundMail(Email):
 	def process(self):
 		"""Create communication record from email."""
 		if self.is_sender_same_as_receiver() and not self.is_reply():
-			if frappe.flags.in_test:
+			if frappe.in_test:
 				print("WARN: Cannot pull email. Sender same as recipient inbox")
 			raise SentEmailInInboxError
 

frappe/email/smtp.py

@@ -109,7 +109,7 @@ class SMTPServer:
 			frappe.request.after_response.add(self.quit)
 		elif frappe.job:
 			frappe.job.after_job.add(self.quit)
-		elif not frappe.flags.in_test:
+		elif not frappe.in_test:
 			# Console?
 			import atexit
 

frappe/integrations/doctype/webhook/__init__.py

@@ -115,6 +115,6 @@ def flush_webhook_execution_queue():
 			"frappe.integrations.doctype.webhook.webhook.enqueue_webhook",
 			doc=instance.doc,
 			webhook=instance.webhook,
-			now=frappe.flags.in_test,
+			now=frappe.in_test,
 			queue=instance.webhook.background_jobs_queue or "default",
 		)

frappe/model/base_document.py

@@ -977,7 +977,7 @@ class BaseDocument:
 			self.set(df.fieldname, cstr(self.get(df.fieldname)).strip())
 			value = self.get(df.fieldname)
 
-			if value not in options and not (frappe.flags.in_test and value.startswith("_T-")):
+			if value not in options and not (frappe.in_test and value.startswith("_T-")):
 				# show an elaborate message
 				prefix = _("Row #{0}:").format(self.idx) if self.get("parentfield") else ""
 				label = _(self.meta.get_label(df.fieldname))

frappe/model/delete_doc.py

@@ -153,7 +153,7 @@ def delete_doc(
 					"frappe.model.delete_doc.delete_dynamic_links",
 					doctype=doc.doctype,
 					name=doc.name,
-					now=frappe.flags.in_test,
+					now=frappe.in_test,
 					enqueue_after_commit=True,
 				)
 

frappe/model/document.py

@@ -499,7 +499,7 @@ class Document(BaseDocument):
 		if ignore_permissions is not None:
 			self.flags.ignore_permissions = ignore_permissions
 
-		self.flags.ignore_version = frappe.flags.in_test if ignore_version is None else ignore_version
+		self.flags.ignore_version = frappe.in_test if ignore_version is None else ignore_version
 
 		if self.get("__islocal") or not self.get("name"):
 			return self.insert()

frappe/model/dynamic_links.py

@@ -32,7 +32,7 @@ def get_dynamic_link_map(for_delete=False):
 
 	Note: Will not map single doctypes
 	"""
-	if getattr(frappe.local, "dynamic_link_map", None) is None or frappe.flags.in_test:
+	if getattr(frappe.local, "dynamic_link_map", None) is None or frappe.in_test:
 		# Build from scratch
 		dynamic_link_map = {}
 		for df in get_dynamic_links():

frappe/model/trace.py

@@ -90,7 +90,7 @@ class TracedValue:
 
 		"""
 		if value in self.forbidden_values:
-			if frappe.flags.in_test:
+			if frappe.in_test:
 				frappe.throw(f"{self.field_name} cannot be set to {value}", AssertionError)
 			else:
 				frappe.throw(f"{self.field_name} cannot be set to {value}")
@@ -99,7 +99,7 @@ class TracedValue:
 			try:
 				self.custom_validation(obj, value)
 			except Exception as e:
-				if frappe.flags.in_test:
+				if frappe.in_test:
 					frappe.throw(str(e), AssertionError)
 				else:
 					frappe.throw(str(e))

frappe/parallel_test_runner.py

@@ -12,7 +12,7 @@ import click
 import requests
 
 import frappe
-from frappe.tests.utils import make_test_records
+from frappe.tests.utils import make_test_records, toggle_test_mode
 
 from .testing.environment import _decorate_all_methods_and_functions_with_type_checker
 from .testing.result import TestResult
@@ -53,7 +53,7 @@ class ParallelTestRunner:
 		if self.dry_run:
 			return
 
-		frappe.flags.in_test = True
+		toggle_test_mode(True)
 		frappe.clear_cache()
 		frappe.utils.scheduler.disable_scheduler()
 		_decorate_all_methods_and_functions_with_type_checker()

frappe/printing/doctype/print_format/print_format.py

@@ -70,7 +70,7 @@ class PrintFormat(Document):
 			and not frappe.local.conf.get("developer_mode")
 			and not frappe.flags.in_migrate
 			and not frappe.flags.in_install
-			and not frappe.flags.in_test
+			and not frappe.in_test
 		):
 			frappe.throw(frappe._("Standard Print Format cannot be updated"))
 

frappe/printing/doctype/print_style/print_style.py

@@ -26,7 +26,7 @@ class PrintStyle(Document):
 			self.standard == 1
 			and not frappe.local.conf.get("developer_mode")
 			and not frappe.flags.in_import
-			and not frappe.flags.in_test
+			and not frappe.in_test
 		):
 			frappe.throw(frappe._("Standard Print Style cannot be changed. Please duplicate to edit."))
 

frappe/recorder.py

@@ -360,7 +360,7 @@ def start(
 @administrator_only
 def stop(*args, **kwargs):
 	frappe.client_cache.set_value(RECORDER_INTERCEPT_FLAG, False)
-	frappe.enqueue(post_process, now=frappe.flags.in_test)
+	frappe.enqueue(post_process, now=frappe.in_test)
 
 
 @frappe.whitelist()

frappe/sessions.py

@@ -199,7 +199,7 @@ def get_csrf_token():
 
 def generate_csrf_token():
 	frappe.local.session.data.csrf_token = frappe.generate_hash()
-	if not frappe.flags.in_test:
+	if not frappe.in_test:
 		frappe.local.session_obj.update(force=True)
 
 

frappe/testing/environment.py

@@ -29,7 +29,7 @@ import tomli
 
 import frappe
 import frappe.utils.scheduler
-from frappe.tests.utils import make_test_records
+from frappe.tests.utils import make_test_records, toggle_test_mode
 
 from .runner import TestRunnerError
 from .utils import debug_timer
@@ -53,7 +53,7 @@ def _initialize_test_environment(site, config):
 		raise TestRunnerError(f"Failed to connect to the database: {e}") from e
 
 	# Set various test-related flags
-	frappe.flags.in_test = True
+	toggle_test_mode(True)
 	frappe.flags.print_messages = logger.getEffectiveLevel() < logging.INFO
 	frappe.flags.tests_verbose = logger.getEffectiveLevel() < logging.INFO
 

frappe/tests/test_api_v2.py

@@ -6,6 +6,7 @@ import requests
 import frappe
 from frappe.installer import update_site_config
 from frappe.tests.test_api import FrappeAPITestCase, suppress_stdout
+from frappe.tests.utils import toggle_test_mode
 
 authorization_token = None
 
@@ -88,7 +89,13 @@ class TestResourceAPIV2(FrappeAPITestCase):
 	def test_copy_document(self):
 		doc = frappe.get_doc(self.DOCTYPE, self.GENERATED_DOCUMENTS[0])
 
+		# disabled temporarily to assert that `docstatus` is not copied outside of tests
+		toggle_test_mode(False)
+		try:
 			response = self.get(self.resource(self.DOCTYPE, doc.name, "copy"))
+		finally:
+			toggle_test_mode(True)
+
 		self.assertEqual(response.status_code, 200)
 		data = response.json["data"]
 

frappe/tests/test_trace.py

@@ -5,6 +5,7 @@ import frappe
 from frappe.model.document import Document
 from frappe.model.trace import TracedDocument, traced_field
 from frappe.tests import UnitTestCase
+from frappe.tests.utils import toggle_test_mode
 
 
 def create_mock_meta(doctype):
@@ -107,9 +108,9 @@ class TestTracedFieldContext(UnitTestCase):
 	def test_traced_field_context_not_in_test_mode(self):
 		doc = TestDocument()
 
-		# Temporarily set frappe.flags.in_test to False
-		original_in_test = frappe.flags.in_test
-		frappe.flags.in_test = False
+		# Temporarily set frappe.in_test to False
+		original_in_test = frappe.in_test
+		toggle_test_mode(False)
 
 		try:
 			with self.trace_fields(TestDocument, test_field={"forbidden_values": ["forbidden"]}):
@@ -120,7 +121,7 @@ class TestTracedFieldContext(UnitTestCase):
 				self.assertEqual(doc.test_field, "allowed")
 		finally:
 			# Restore the original in_test flag
-			frappe.flags.in_test = original_in_test
+			toggle_test_mode(original_in_test)
 
 		# After context
 		doc.test_field = "forbidden"

frappe/tests/test_utils.py

@@ -20,6 +20,7 @@ import frappe
 from frappe.installer import parse_app_name
 from frappe.model.document import Document
 from frappe.tests import IntegrationTestCase, MockedRequestTestCase
+from frappe.tests.utils import toggle_test_mode
 from frappe.utils import (
 	add_trackers_to_url,
 	ceil,
@@ -988,9 +989,9 @@ class TestLazyLoader(IntegrationTestCase):
 class TestIdenticon(IntegrationTestCase):
 	def test_get_gravatar(self):
 		# developers@frappe.io has a gravatar linked so str URL will be returned
-		frappe.flags.in_test = False
+		toggle_test_mode(False)
 		gravatar_url = get_gravatar("developers@frappe.io")
-		frappe.flags.in_test = True
+		toggle_test_mode(True)
 		self.assertIsInstance(gravatar_url, str)
 		self.assertTrue(gravatar_url.startswith("http"))
 

frappe/tests/ui_test_helpers.py

@@ -7,7 +7,7 @@ UI_TEST_USER = "frappe@example.com"
 
 
 def whitelist_for_tests(fn):
-	if frappe.request and not frappe.flags.in_test and not frappe._dev_server:
+	if frappe.request and not frappe.in_test and not frappe._dev_server:
 		frappe.throw("Cannot run UI tests. Use a development server with `bench start`")
 
 	return frappe.whitelist()(fn)

frappe/tests/utils/__init__.py

@@ -26,6 +26,12 @@ def check_orpahned_doctypes():
 		)
 
 
+def toggle_test_mode(enable: bool):
+	"""Enable or disable `frappe.in_test` (and related deprecated flag)"""
+	frappe.in_test = enable
+	frappe.local.flags.in_test = enable
+
+
 from frappe.deprecation_dumpster import (
 	get_tests_CompatFrappeTestCase,
 )

frappe/utils/__init__.py

@@ -269,7 +269,7 @@ def has_gravatar(email: str) -> str:
 	"""Return gravatar url if user has set an avatar at gravatar.com."""
 	import requests
 
-	if frappe.flags.in_import or frappe.flags.in_install or frappe.flags.in_test:
+	if frappe.flags.in_import or frappe.flags.in_install or frappe.in_test:
 		# no gravatar if via upload
 		# since querying gravatar for every item will be slow
 		return ""

frappe/utils/background_jobs.py

@@ -139,7 +139,7 @@ def enqueue(
 			"unknown", "v17", "Using enqueue with `job_name` is deprecated, use `job_id` instead."
 		)
 
-	if not is_async and not frappe.flags.in_test:
+	if not is_async and not frappe.in_test:
 		from frappe.deprecation_dumpster import deprecation_warning
 
 		deprecation_warning(
@@ -148,7 +148,7 @@ def enqueue(
 			"Using enqueue with is_async=False outside of tests is not recommended, use now=True instead.",
 		)
 
-	call_directly = now or (not is_async and not frappe.flags.in_test)
+	call_directly = now or (not is_async and not frappe.in_test)
 	if call_directly:
 		return frappe.call(method, **kwargs)
 
@@ -243,7 +243,9 @@ def execute_job(site, method, event, job_name, kwargs, user=None, is_async=True,
 		frappe.init(site, force=True)
 		frappe.connect()
 		if os.environ.get("CI"):
-			frappe.flags.in_test = True
+			from frappe.tests.utils import toggle_test_mode
+
+			toggle_test_mode(True)
 
 		if user:
 			frappe.set_user(user)

frappe/utils/user.py

@@ -59,7 +59,7 @@ class UserPermissions:
 
 			return user
 
-		if not frappe.flags.in_install_db and not frappe.flags.in_test:
+		if not frappe.flags.in_install_db and not frappe.in_test:
 			user_doc = frappe.cache.hget("user_doc", self.name, get_user_doc)
 			if user_doc:
 				self.doc = frappe.get_doc(user_doc)

frappe/website/doctype/personal_data_deletion_request/personal_data_deletion_request.py

@@ -136,7 +136,7 @@ class PersonalDataDeletionRequest(Document):
 			"_anonymize_data",
 			queue="long",
 			timeout=3000,
-			now=frappe.flags.in_test,
+			now=frappe.in_test,
 		)
 
 	def notify_user_after_deletion(self):

frappe/website/doctype/personal_data_download_request/personal_data_download_request.py

@@ -32,7 +32,7 @@ class PersonalDataDownloadRequest(Document):
 			"generate_file_and_send_mail",
 			queue="short",
 			personal_data=personal_data,
-			now=frappe.flags.in_test,
+			now=frappe.in_test,
 		)
 
 	def generate_file_and_send_mail(self, personal_data):

frappe/website/doctype/web_form/web_form.py

@@ -79,10 +79,7 @@ class WebForm(WebsiteGenerator):
 			self.module = frappe.db.get_value("DocType", self.doc_type, "module")
 
 		in_user_env = not (
-			frappe.flags.in_install
-			or frappe.flags.in_patch
-			or frappe.flags.in_test
-			or frappe.flags.in_fixtures
+			frappe.flags.in_install or frappe.flags.in_patch or frappe.in_test or frappe.flags.in_fixtures
 		)
 		if in_user_env and self.is_standard and not frappe.conf.developer_mode:
 			# only published can be changed for standard web forms

frappe/website/doctype/website_theme/website_theme.py

@@ -55,7 +55,7 @@ class WebsiteTheme(Document):
 			not self.custom
 			and frappe.local.conf.get("developer_mode")
 			and not frappe.flags.in_import
-			and not frappe.flags.in_test
+			and not frappe.in_test
 		):
 			self.export_doc()
 
@@ -65,7 +65,7 @@ class WebsiteTheme(Document):
 		return (
 			not self.custom
 			and not frappe.local.conf.get("developer_mode")
-			and not (frappe.flags.in_import or frappe.flags.in_test or frappe.flags.in_migrate)
+			and not (frappe.flags.in_import or frappe.in_test or frappe.flags.in_migrate)
 		)
 
 	def on_trash(self):

frappe/website/utils.py

@@ -94,7 +94,7 @@ def get_comment_list(doctype, name):
 
 
 def get_home_page():
-	if frappe.local.flags.home_page and not frappe.flags.in_test:
+	if frappe.local.flags.home_page and not frappe.in_test:
 		return frappe.local.flags.home_page
 
 	def _get_home_page():

frappe/website/website_generator.py

@@ -155,7 +155,7 @@ class WebsiteGenerator(Document):
 
 	def remove_old_route_from_index(self):
 		"""Remove page from the website index if the route has changed."""
-		if self.allow_website_search_indexing() or frappe.flags.in_test:
+		if self.allow_website_search_indexing() or frappe.in_test:
 			return
 		old_doc = self.get_doc_before_save()
 		# Check if the route is changed
@@ -169,7 +169,7 @@ class WebsiteGenerator(Document):
 		- remove document from index if document is unpublished
 		- update index otherwise
 		"""
-		if not self.allow_website_search_indexing() or frappe.flags.in_test:
+		if not self.allow_website_search_indexing() or frappe.in_test:
 			return
 
 		if self.is_website_published():

frappe/workflow/doctype/workflow_action/workflow_action.py

@@ -122,7 +122,7 @@ def process_workflow_actions(doc, state):
 			doc=doc,
 			transitions=next_possible_transitions,
 			enqueue_after_commit=True,
-			now=frappe.flags.in_test,
+			now=frappe.in_test,
 		)
 
 

frappe/www/list.py

@@ -56,7 +56,7 @@ def get(doctype, txt=None, limit_start=0, limit=20, pathname=None, **kwargs):
 			new_context.doc = frappe.get_doc(doc.doctype, doc.name)
 			new_context.update(new_context.doc.as_dict())
 
-		if not frappe.flags.in_test:
+		if not frappe.in_test:
 			pathname = pathname or frappe.local.request.path
 			new_context["pathname"] = pathname.strip("/ ")
 		new_context.update(list_context)