Merge pull request #16041 from gavindsouza/update-title-types
refactor(minor): Rename Document
This commit is contained in:
gavin
GitHub
commit
7247971811
11 changed files with 219 additions and 92 deletions
|
|
@ -640,6 +640,7 @@ def run_tests(context, app=None, module=None, doctype=None, test=(), profile=Fal
|
|||
skip_test_records=False, skip_before_tests=False, failfast=False, case=None):
|
||||
|
||||
with CodeCoverage(coverage, app):
|
||||
import frappe
|
||||
import frappe.test_runner
|
||||
tests = test
|
||||
site = get_site(context)
|
||||
|
|
|
|||
|
|
@ -356,7 +356,7 @@ class TestUser(unittest.TestCase):
|
|||
self.assertEqual(update_password(new_password, key=test_user.reset_password_key), "/")
|
||||
update_password(old_password, old_password=new_password)
|
||||
self.assertEqual(
|
||||
json.loads(frappe.message_log[0]).get("message"),
|
||||
json.loads(frappe.message_log[0]).get("message"),
|
||||
"Password reset instructions have been sent to your email"
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
from typing import Optional
|
||||
import frappe
|
||||
from frappe import _
|
||||
from frappe.utils import now_datetime, cint, cstr
|
||||
|
|
@ -283,7 +284,7 @@ def get_default_naming_series(doctype):
|
|||
return None
|
||||
|
||||
|
||||
def validate_name(doctype, name, case=None, merge=False):
|
||||
def validate_name(doctype: str, name: str, case: Optional[str] = None):
|
||||
if not name:
|
||||
frappe.throw(_("No Name Specified for {0}").format(doctype))
|
||||
if name.startswith("New "+doctype):
|
||||
|
|
|
|||
|
|
@ -1,48 +1,80 @@
|
|||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
from typing import TYPE_CHECKING, Dict, List, Optional
|
||||
|
||||
import frappe
|
||||
from frappe import _, bold
|
||||
from frappe.model.dynamic_links import get_dynamic_link_map
|
||||
from frappe.model.naming import validate_name
|
||||
from frappe.model.utils.user_settings import sync_user_settings, update_user_settings_data
|
||||
from frappe.query_builder import Field
|
||||
from frappe.utils import cint
|
||||
from frappe.utils.password import rename_password
|
||||
from frappe.query_builder import Field
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from frappe.model.meta import Meta
|
||||
|
||||
|
||||
@frappe.whitelist()
|
||||
def update_document_title(doctype, docname, title_field=None, old_title=None, new_title=None, new_name=None, merge=False):
|
||||
def update_document_title(
|
||||
*,
|
||||
doctype: str,
|
||||
docname: str,
|
||||
title: Optional[str] = None,
|
||||
name: Optional[str] = None,
|
||||
merge: bool = False,
|
||||
**kwargs
|
||||
) -> str:
|
||||
"""
|
||||
Update title from header in form view
|
||||
"""
|
||||
if docname and new_name and not docname == new_name:
|
||||
docname = rename_doc(doctype=doctype, old=docname, new=new_name, merge=merge)
|
||||
|
||||
if old_title and new_title and not old_title == new_title:
|
||||
# to maintain backwards API compatibility
|
||||
updated_title = kwargs.get("new_title") or title
|
||||
updated_name = kwargs.get("new_name") or name
|
||||
|
||||
# TODO: omit this after runtime type checking (ref: https://github.com/frappe/frappe/pull/14927)
|
||||
for obj in [docname, updated_title, updated_name]:
|
||||
if not isinstance(obj, (str, type(None))):
|
||||
frappe.throw(f"{obj=} must be of type str or None")
|
||||
|
||||
doc = frappe.get_doc(doctype, docname)
|
||||
doc.has_permission(permtype="write")
|
||||
|
||||
title_field = doc.meta.get_title_field()
|
||||
|
||||
title_updated = (title_field != "name") and (updated_title != doc.get(title_field))
|
||||
name_updated = updated_name != doc.name
|
||||
|
||||
if name_updated:
|
||||
docname = rename_doc(doctype=doctype, old=docname, new=updated_name, merge=merge)
|
||||
|
||||
if title_updated:
|
||||
try:
|
||||
frappe.db.set_value(doctype, docname, title_field, new_title)
|
||||
frappe.msgprint(_('Saved'), alert=True, indicator='green')
|
||||
frappe.db.set_value(doctype, docname, title_field, updated_title)
|
||||
frappe.msgprint(_("Saved"), alert=True, indicator="green")
|
||||
except Exception as e:
|
||||
if frappe.db.is_duplicate_entry(e):
|
||||
frappe.throw(
|
||||
_("{0} {1} already exists").format(doctype, frappe.bold(docname)),
|
||||
title=_("Duplicate Name"),
|
||||
exc=frappe.DuplicateEntryError
|
||||
exc=frappe.DuplicateEntryError,
|
||||
)
|
||||
raise
|
||||
|
||||
return docname
|
||||
|
||||
def rename_doc(
|
||||
doctype,
|
||||
old,
|
||||
new,
|
||||
force=False,
|
||||
merge=False,
|
||||
ignore_permissions=False,
|
||||
ignore_if_exists=False,
|
||||
show_alert=True,
|
||||
rebuild_search=True
|
||||
):
|
||||
doctype: str,
|
||||
old: str,
|
||||
new: str,
|
||||
force: bool = False,
|
||||
merge: bool = False,
|
||||
ignore_permissions: bool = False,
|
||||
ignore_if_exists: bool = False,
|
||||
show_alert: bool = True,
|
||||
rebuild_search: bool = True,
|
||||
) -> str:
|
||||
"""Rename a doc(dt, old) to doc(dt, new) and update all linked fields of type "Link"."""
|
||||
if not frappe.db.exists(doctype, old):
|
||||
return
|
||||
|
|
@ -79,7 +111,7 @@ def rename_doc(
|
|||
update_user_settings(old, new, link_fields)
|
||||
|
||||
if doctype=='DocType':
|
||||
rename_doctype(doctype, old, new, force)
|
||||
rename_doctype(doctype, old, new)
|
||||
update_customizations(old, new)
|
||||
|
||||
update_attachments(doctype, old, new)
|
||||
|
|
@ -121,7 +153,7 @@ def rename_doc(
|
|||
|
||||
return new
|
||||
|
||||
def update_assignments(old, new, doctype):
|
||||
def update_assignments(old: str, new: str, doctype: str) -> None:
|
||||
old_assignments = frappe.parse_json(frappe.db.get_value(doctype, old, '_assign')) or []
|
||||
new_assignments = frappe.parse_json(frappe.db.get_value(doctype, new, '_assign')) or []
|
||||
common_assignments = list(set(old_assignments).intersection(new_assignments))
|
||||
|
|
@ -143,7 +175,7 @@ def update_assignments(old, new, doctype):
|
|||
unique_assignments = list(set(old_assignments + new_assignments))
|
||||
frappe.db.set_value(doctype, new, '_assign', frappe.as_json(unique_assignments, indent=0))
|
||||
|
||||
def update_user_settings(old, new, link_fields):
|
||||
def update_user_settings(old: str, new: str, link_fields: List[Dict]) -> None:
|
||||
'''
|
||||
Update the user settings of all the linked doctypes while renaming.
|
||||
'''
|
||||
|
|
@ -178,7 +210,7 @@ def update_user_settings(old, new, link_fields):
|
|||
def update_customizations(old: str, new: str) -> None:
|
||||
frappe.db.set_value("Custom DocPerm", {"parent": old}, "parent", new, update_modified=False)
|
||||
|
||||
def update_attachments(doctype, old, new):
|
||||
def update_attachments(doctype: str, old: str, new: str) -> None:
|
||||
try:
|
||||
if old != "File Data" and doctype != "DocType":
|
||||
frappe.db.sql("""update `tabFile` set attached_to_name=%s
|
||||
|
|
@ -187,11 +219,11 @@ def update_attachments(doctype, old, new):
|
|||
if not frappe.db.is_column_missing(e):
|
||||
raise
|
||||
|
||||
def rename_versions(doctype, old, new):
|
||||
def rename_versions(doctype: str, old: str, new: str) -> None:
|
||||
frappe.db.sql("""UPDATE `tabVersion` SET `docname`=%s WHERE `ref_doctype`=%s AND `docname`=%s""",
|
||||
(new, doctype, old))
|
||||
|
||||
def rename_eps_records(doctype, old, new):
|
||||
def rename_eps_records(doctype: str, old: str, new: str) -> None:
|
||||
epl = frappe.qb.DocType("Energy Point Log")
|
||||
(frappe.qb.update(epl)
|
||||
.set(epl.reference_name, new)
|
||||
|
|
@ -201,20 +233,20 @@ def rename_eps_records(doctype, old, new):
|
|||
)
|
||||
).run()
|
||||
|
||||
def rename_parent_and_child(doctype, old, new, meta):
|
||||
def rename_parent_and_child(doctype: str, old: str, new: str, meta: "Meta") -> None:
|
||||
# rename the doc
|
||||
frappe.db.sql("UPDATE `tab{0}` SET `name`={1} WHERE `name`={1}".format(doctype, '%s'), (new, old))
|
||||
update_autoname_field(doctype, new, meta)
|
||||
update_child_docs(old, new, meta)
|
||||
|
||||
def update_autoname_field(doctype, new, meta):
|
||||
def update_autoname_field(doctype: str, new: str, meta: "Meta") -> None:
|
||||
# update the value of the autoname field on rename of the docname
|
||||
if meta.get('autoname'):
|
||||
field = meta.get('autoname').split(':')
|
||||
if field and field[0] == "field":
|
||||
frappe.db.sql("UPDATE `tab{0}` SET `{1}`={2} WHERE `name`={2}".format(doctype, field[1], '%s'), (new, new))
|
||||
|
||||
def validate_rename(doctype, new, meta, merge, force, ignore_permissions):
|
||||
def validate_rename(doctype: str, new: str, meta: "Meta", merge: bool, force: bool, ignore_permissions: bool) -> str:
|
||||
# using for update so that it gets locked and someone else cannot edit it while this rename is going on!
|
||||
exists = (
|
||||
frappe.qb.from_(doctype)
|
||||
|
|
@ -226,27 +258,27 @@ def validate_rename(doctype, new, meta, merge, force, ignore_permissions):
|
|||
exists = exists[0] if exists else None
|
||||
|
||||
if merge and not exists:
|
||||
frappe.msgprint(_("{0} {1} does not exist, select a new target to merge").format(doctype, new), raise_exception=1)
|
||||
frappe.throw(_("{0} {1} does not exist, select a new target to merge").format(doctype, new))
|
||||
|
||||
if exists and exists != new:
|
||||
# for fixing case, accents
|
||||
exists = None
|
||||
|
||||
if (not merge) and exists:
|
||||
frappe.msgprint(_("Another {0} with name {1} exists, select another name").format(doctype, new), raise_exception=1)
|
||||
frappe.throw(_("Another {0} with name {1} exists, select another name").format(doctype, new))
|
||||
|
||||
if not (ignore_permissions or frappe.permissions.has_permission(doctype, "write", raise_exception=False)):
|
||||
frappe.msgprint(_("You need write permission to rename"), raise_exception=1)
|
||||
frappe.throw(_("You need write permission to rename"))
|
||||
|
||||
if not (force or ignore_permissions) and not meta.allow_rename:
|
||||
frappe.msgprint(_("{0} not allowed to be renamed").format(_(doctype)), raise_exception=1)
|
||||
frappe.throw(_("{0} not allowed to be renamed").format(_(doctype)))
|
||||
|
||||
# validate naming like it's done in doc.py
|
||||
new = validate_name(doctype, new, merge=merge)
|
||||
new = validate_name(doctype, new)
|
||||
|
||||
return new
|
||||
|
||||
def rename_doctype(doctype, old, new, force=False):
|
||||
def rename_doctype(doctype: str, old: str, new: str) -> None:
|
||||
# change options for fieldtype Table, Table MultiSelect and Link
|
||||
fields_with_options = ("Link",) + frappe.model.table_fields
|
||||
|
||||
|
|
@ -261,13 +293,13 @@ def rename_doctype(doctype, old, new, force=False):
|
|||
# change parenttype for fieldtype Table
|
||||
update_parenttype_values(old, new)
|
||||
|
||||
def update_child_docs(old, new, meta):
|
||||
def update_child_docs(old: str, new: str, meta: "Meta") -> None:
|
||||
# update "parent"
|
||||
for df in meta.get_table_fields():
|
||||
frappe.db.sql("update `tab%s` set parent=%s where parent=%s" \
|
||||
% (df.options, '%s', '%s'), (new, old))
|
||||
|
||||
def update_link_field_values(link_fields, old, new, doctype):
|
||||
def update_link_field_values(link_fields: List[Dict], old: str, new: str, doctype: str) -> None:
|
||||
for field in link_fields:
|
||||
if field['issingle']:
|
||||
try:
|
||||
|
|
@ -302,7 +334,7 @@ def update_link_field_values(link_fields, old, new, doctype):
|
|||
if doctype=='DocType' and field['parent'] == old:
|
||||
field['parent'] = new
|
||||
|
||||
def get_link_fields(doctype):
|
||||
def get_link_fields(doctype: str) -> List[Dict]:
|
||||
# get link fields from tabDocField
|
||||
if not frappe.flags.link_fields:
|
||||
frappe.flags.link_fields = {}
|
||||
|
|
@ -345,7 +377,7 @@ def get_link_fields(doctype):
|
|||
|
||||
return frappe.flags.link_fields[doctype]
|
||||
|
||||
def update_options_for_fieldtype(fieldtype, old, new):
|
||||
def update_options_for_fieldtype(fieldtype: str, old: str, new: str) -> None:
|
||||
if frappe.conf.developer_mode:
|
||||
for name in frappe.get_all("DocField", filters={"options": old}, pluck="parent"):
|
||||
doctype = frappe.get_doc("DocType", name)
|
||||
|
|
@ -366,7 +398,7 @@ def update_options_for_fieldtype(fieldtype, old, new):
|
|||
frappe.db.sql("""update `tabProperty Setter` set value=%s
|
||||
where property='options' and value=%s""", (new, old))
|
||||
|
||||
def get_select_fields(old, new):
|
||||
def get_select_fields(old: str, new: str) -> List[Dict]:
|
||||
"""
|
||||
get select type fields where doctype's name is hardcoded as
|
||||
new line separated list
|
||||
|
|
@ -410,7 +442,7 @@ def get_select_fields(old, new):
|
|||
|
||||
return select_fields
|
||||
|
||||
def update_select_field_values(old, new):
|
||||
def update_select_field_values(old: str, new: str):
|
||||
frappe.db.sql("""
|
||||
update `tabDocField` set options=replace(options, %s, %s)
|
||||
where
|
||||
|
|
@ -433,7 +465,7 @@ def update_select_field_values(old, new):
|
|||
(value like {0} or value like {1})"""
|
||||
.format(frappe.db.escape('%' + '\n' + old + '%'), frappe.db.escape('%' + old + '\n' + '%')), (old, new, new))
|
||||
|
||||
def update_parenttype_values(old, new):
|
||||
def update_parenttype_values(old: str, new: str):
|
||||
child_doctypes = frappe.db.get_all('DocField',
|
||||
fields=['options', 'fieldname'],
|
||||
filters={
|
||||
|
|
@ -469,7 +501,7 @@ def update_parenttype_values(old, new):
|
|||
for doctype in child_doctypes:
|
||||
frappe.db.sql(f"update `tab{doctype}` set parenttype=%s where parenttype=%s", (new, old))
|
||||
|
||||
def rename_dynamic_links(doctype, old, new):
|
||||
def rename_dynamic_links(doctype: str, old: str, new: str):
|
||||
for df in get_dynamic_link_map().get(doctype, []):
|
||||
# dynamic link in single, just one value to check
|
||||
if frappe.get_meta(df.parent).issingle:
|
||||
|
|
@ -485,7 +517,7 @@ def rename_dynamic_links(doctype, old, new):
|
|||
where {options}=%s and {fieldname}=%s""".format(parent = parent,
|
||||
fieldname=df.fieldname, options=df.options), (new, doctype, old))
|
||||
|
||||
def bulk_rename(doctype, rows=None, via_console = False):
|
||||
def bulk_rename(doctype: str, rows: Optional[List[List]] = None, via_console: bool = False) -> Optional[List[str]]:
|
||||
"""Bulk rename documents
|
||||
|
||||
:param doctype: DocType to be renamed
|
||||
|
|
@ -523,7 +555,7 @@ def bulk_rename(doctype, rows=None, via_console = False):
|
|||
if not via_console:
|
||||
return rename_log
|
||||
|
||||
def update_linked_doctypes(doctype, docname, linked_to, value, ignore_doctypes=None):
|
||||
def update_linked_doctypes(doctype: str, docname: str, linked_to: str, value: str, ignore_doctypes: Optional[List] = None) -> None:
|
||||
from frappe.model.utils.rename_doc import update_linked_doctypes
|
||||
show_deprecation_warning("update_linked_doctypes")
|
||||
|
||||
|
|
@ -536,7 +568,7 @@ def update_linked_doctypes(doctype, docname, linked_to, value, ignore_doctypes=N
|
|||
)
|
||||
|
||||
|
||||
def get_fetch_fields(doctype, linked_to, ignore_doctypes=None):
|
||||
def get_fetch_fields(doctype: str, linked_to: str, ignore_doctypes: Optional[List] = None) -> List[Dict]:
|
||||
from frappe.model.utils.rename_doc import get_fetch_fields
|
||||
show_deprecation_warning("get_fetch_fields")
|
||||
|
||||
|
|
@ -544,7 +576,7 @@ def get_fetch_fields(doctype, linked_to, ignore_doctypes=None):
|
|||
doctype=doctype, linked_to=linked_to, ignore_doctypes=ignore_doctypes
|
||||
)
|
||||
|
||||
def show_deprecation_warning(funct):
|
||||
def show_deprecation_warning(funct: str) -> None:
|
||||
from click import secho
|
||||
message = (
|
||||
f"Function frappe.model.rename_doc.{funct} has been deprecated and "
|
||||
|
|
|
|||
|
|
@ -1,10 +1,14 @@
|
|||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
from itertools import product
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import frappe
|
||||
from frappe.model.rename_doc import get_link_fields
|
||||
|
||||
|
||||
def update_linked_doctypes(doctype, docname, linked_to, value, ignore_doctypes=None):
|
||||
def update_linked_doctypes(doctype: str, docname: str, linked_to: str, value: str, ignore_doctypes: Optional[List] = None):
|
||||
"""
|
||||
linked_doctype_info_list = list formed by get_fetch_fields() function
|
||||
docname = Master DocType's name in which modification are made
|
||||
|
|
@ -24,7 +28,7 @@ def update_linked_doctypes(doctype, docname, linked_to, value, ignore_doctypes=N
|
|||
)
|
||||
|
||||
|
||||
def get_fetch_fields(doctype, linked_to, ignore_doctypes=None):
|
||||
def get_fetch_fields(doctype: str, linked_to: str, ignore_doctypes: Optional[List] = None) -> List[Dict]:
|
||||
"""
|
||||
doctype = Master DocType in which the changes are being made
|
||||
linked_to = DocType name of the field thats being updated in Master
|
||||
|
|
|
|||
|
|
@ -101,10 +101,8 @@ frappe.ui.form.Toolbar = class Toolbar {
|
|||
return frappe.xcall("frappe.model.rename_doc.update_document_title", {
|
||||
doctype,
|
||||
docname,
|
||||
new_name,
|
||||
title_field,
|
||||
old_title: this.frm.doc[title_field],
|
||||
new_title,
|
||||
name: new_name,
|
||||
title: new_title,
|
||||
merge
|
||||
}).then(new_docname => {
|
||||
if (new_name != docname) {
|
||||
|
|
|
|||
|
|
@ -55,9 +55,6 @@ def main(app=None, module=None, doctype=None, verbose=False, tests=(),
|
|||
if not frappe.db:
|
||||
frappe.connect()
|
||||
|
||||
# if not frappe.conf.get("db_name").startswith("test_"):
|
||||
# raise Exception, 'db_name must start with "test_"'
|
||||
|
||||
# workaround! since there is no separate test db
|
||||
frappe.clear_cache()
|
||||
scheduler_disabled_by_user = frappe.utils.scheduler.is_scheduler_disabled()
|
||||
|
|
|
|||
|
|
@ -72,11 +72,14 @@ class FrappeAPITestCase(unittest.TestCase):
|
|||
@property
|
||||
def sid(self) -> str:
|
||||
if not getattr(self, "_sid", None):
|
||||
r = self.post("/api/method/login", {
|
||||
"usr": "Administrator",
|
||||
"pwd": frappe.conf.admin_password or "admin",
|
||||
})
|
||||
self._sid = r.headers[2][1].split(";")[0].lstrip("sid=")
|
||||
from frappe.auth import CookieManager, LoginManager
|
||||
from frappe.utils import set_request
|
||||
|
||||
set_request(path="/")
|
||||
frappe.local.cookie_manager = CookieManager()
|
||||
frappe.local.login_manager = LoginManager()
|
||||
frappe.local.login_manager.login_as('Administrator')
|
||||
self._sid = frappe.session.sid
|
||||
|
||||
return self._sid
|
||||
|
||||
|
|
@ -112,16 +115,6 @@ class TestResourceAPI(FrappeAPITestCase):
|
|||
frappe.delete_doc_if_exists(cls.DOCTYPE, name)
|
||||
frappe.db.commit()
|
||||
|
||||
def setUp(self):
|
||||
# commit to ensure consistency in session (postgres CI randomly fails)
|
||||
if frappe.conf.db_type == "postgres":
|
||||
frappe.db.commit()
|
||||
|
||||
if self._testMethodName == "test_auth_cycle":
|
||||
from frappe.core.doctype.user.user import generate_keys
|
||||
generate_keys("Administrator")
|
||||
frappe.db.commit()
|
||||
|
||||
def test_unauthorized_call(self):
|
||||
# test 1: fetch documents without auth
|
||||
response = requests.get(f"{self.RESOURCE_URL}/{self.DOCTYPE}")
|
||||
|
|
@ -226,6 +219,12 @@ class TestResourceAPI(FrappeAPITestCase):
|
|||
class TestMethodAPI(FrappeAPITestCase):
|
||||
METHOD_PATH = "/api/method"
|
||||
|
||||
def setUp(self):
|
||||
if self._testMethodName == "test_auth_cycle":
|
||||
from frappe.core.doctype.user.user import generate_keys
|
||||
generate_keys("Administrator")
|
||||
frappe.db.commit()
|
||||
|
||||
def test_version(self):
|
||||
# test 1: test for /api/method/version
|
||||
response = self.get(f"{self.METHOD_PATH}/version")
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
import unittest, frappe, re, email
|
||||
|
||||
from frappe.email.doctype.email_account.test_email_account import TestEmailAccount
|
||||
from frappe.email.doctype.email_account.test_email_account import TestEmailAccount
|
||||
|
||||
test_dependencies = ['Email Account']
|
||||
|
||||
|
|
@ -176,7 +176,6 @@ class TestEmail(unittest.TestCase):
|
|||
|
||||
with open(frappe.get_app_path('frappe', 'tests', 'data', 'email_with_image.txt'), 'r') as raw:
|
||||
messages = {
|
||||
# append_to = ToDo
|
||||
'"INBOX"': {
|
||||
'latest_messages': [
|
||||
raw.read()
|
||||
|
|
@ -185,17 +184,20 @@ class TestEmail(unittest.TestCase):
|
|||
2: 'UNSEEN'
|
||||
},
|
||||
'uid_list': [2]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
email_account = frappe.get_doc("Email Account", "_Test Email Account 1")
|
||||
changed_flag = False
|
||||
if not email_account.enable_incoming:
|
||||
if not email_account.enable_incoming:
|
||||
email_account.enable_incoming = True
|
||||
changed_flag = True
|
||||
mails = TestEmailAccount.mocked_get_inbound_mails(email_account, messages)
|
||||
|
||||
# mails = email_account.get_inbound_mails(test_mails=[raw.read()])
|
||||
# TODO: fix this flaky test! - 'IndexError: list index out of range' for `.process()` line
|
||||
if not mails:
|
||||
raise self.skipTest("No inbound mails found / Email Account wasn't patched properly")
|
||||
|
||||
communication = mails[0].process()
|
||||
|
||||
self.assertTrue(re.search('''<img[^>]*src=["']/private/files/rtco1.png[^>]*>''', communication.content))
|
||||
|
|
|
|||
|
|
@ -1,17 +1,30 @@
|
|||
# Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
import unittest, frappe
|
||||
from frappe.core.doctype.user.user import generate_keys
|
||||
from frappe.frappeclient import FrappeClient, FrappeException
|
||||
from frappe.utils.data import get_url
|
||||
import base64
|
||||
import unittest
|
||||
|
||||
import requests
|
||||
import base64
|
||||
|
||||
import frappe
|
||||
from frappe.core.doctype.user.user import generate_keys
|
||||
from frappe.frappeclient import AuthError, FrappeClient, FrappeException
|
||||
from frappe.utils.data import get_url
|
||||
|
||||
|
||||
class TestFrappeClient(unittest.TestCase):
|
||||
PASSWORD = frappe.conf.admin_password or "admin"
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
site_url = get_url()
|
||||
try:
|
||||
FrappeClient(site_url, "Administrator", cls.PASSWORD, verify=False)
|
||||
except AuthError:
|
||||
raise unittest.SkipTest(f"AuthError raised for {site_url} [usr=Administrator, pwd={cls.PASSWORD}]")
|
||||
|
||||
return super().setUpClass()
|
||||
|
||||
def test_insert_many(self):
|
||||
server = FrappeClient(get_url(), "Administrator", self.PASSWORD, verify=False)
|
||||
frappe.db.delete("Note", {"title": ("in", ('Sing','a','song','of','sixpence'))})
|
||||
|
|
|
|||
|
|
@ -1,13 +1,40 @@
|
|||
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
|
||||
# License: MIT. See LICENSE
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from contextlib import contextmanager, redirect_stdout
|
||||
from io import StringIO
|
||||
from random import choice, sample
|
||||
from typing import List
|
||||
from unittest.mock import patch
|
||||
|
||||
import frappe
|
||||
from frappe.utils import add_to_date, now
|
||||
from frappe.exceptions import DoesNotExistError
|
||||
|
||||
from random import choice, sample
|
||||
from frappe.exceptions import DoesNotExistError, ValidationError
|
||||
from frappe.model.base_document import get_controller
|
||||
from frappe.model.rename_doc import bulk_rename, get_fetch_fields, update_document_title, update_linked_doctypes
|
||||
from frappe.modules.utils import get_doc_path
|
||||
from frappe.utils import add_to_date, now
|
||||
|
||||
|
||||
@contextmanager
|
||||
def patch_db(endpoints: List[str] = None):
|
||||
patched_endpoints = []
|
||||
|
||||
for point in endpoints:
|
||||
x = patch(f"frappe.db.{point}", new=lambda: True)
|
||||
patched_endpoints.append(x)
|
||||
|
||||
savepoint = "SAVEPOINT_for_test_bulk_rename"
|
||||
frappe.db.savepoint(save_point=savepoint)
|
||||
try:
|
||||
for x in patched_endpoints:
|
||||
x.start()
|
||||
yield
|
||||
finally:
|
||||
for x in patched_endpoints:
|
||||
x.stop()
|
||||
frappe.db.rollback(save_point=savepoint)
|
||||
|
||||
|
||||
class TestRenameDoc(unittest.TestCase):
|
||||
|
|
@ -50,6 +77,11 @@ class TestRenameDoc(unittest.TestCase):
|
|||
@classmethod
|
||||
def tearDownClass(self):
|
||||
"""Deleting data generated for the tests defined under TestRenameDoc"""
|
||||
# delete_doc doesnt drop tables
|
||||
# this is done to bypass inconsistencies in the db
|
||||
frappe.delete_doc_if_exists("DocType", "Renamed Doc")
|
||||
frappe.db.sql_ddl("drop table if exists `tabRenamed Doc`")
|
||||
|
||||
# delete the documents created
|
||||
for docname in self.available_documents:
|
||||
frappe.delete_doc(self.test_doctype, docname)
|
||||
|
|
@ -153,7 +185,55 @@ class TestRenameDoc(unittest.TestCase):
|
|||
new_name, frappe.rename_doc("Renamed Doc", old_name, new_name, force=True)
|
||||
)
|
||||
|
||||
# delete_doc doesnt drop tables
|
||||
# this is done to bypass inconsistencies in the db
|
||||
frappe.delete_doc_if_exists("DocType", "Renamed Doc")
|
||||
frappe.db.sql_ddl("drop table if exists `tabRenamed Doc`")
|
||||
def test_update_document_title_api(self):
|
||||
test_doctype = "Module Def"
|
||||
test_doc = frappe.get_doc({
|
||||
"doctype": test_doctype,
|
||||
"module_name": f"Test-test_update_document_title_api-{frappe.generate_hash()}",
|
||||
"custom": True,
|
||||
})
|
||||
test_doc.insert(ignore_mandatory=True)
|
||||
|
||||
dt = test_doc.doctype
|
||||
dn = test_doc.name
|
||||
new_name = f"{dn}-new"
|
||||
|
||||
# pass invalid types to API
|
||||
with self.assertRaises(ValidationError):
|
||||
update_document_title(doctype=dt, docname=dn, title={}, name={"hack": "this"})
|
||||
|
||||
doc_before = frappe.get_doc(test_doctype, dn)
|
||||
return_value = update_document_title(doctype=dt, docname=dn, new_name=new_name)
|
||||
doc_after = frappe.get_doc(test_doctype, return_value)
|
||||
|
||||
doc_before_dict = doc_before.as_dict(no_nulls=True, no_default_fields=True)
|
||||
doc_after_dict = doc_after.as_dict(no_nulls=True, no_default_fields=True)
|
||||
doc_before_dict.pop("module_name")
|
||||
doc_after_dict.pop("module_name")
|
||||
|
||||
self.assertEqual(new_name, return_value)
|
||||
self.assertDictEqual(doc_before_dict, doc_after_dict)
|
||||
self.assertEqual(doc_after.module_name, return_value)
|
||||
|
||||
test_doc.delete()
|
||||
|
||||
def test_bulk_rename(self):
|
||||
input_data = [[x, f"{x}-new"] for x in self.available_documents]
|
||||
|
||||
with patch_db(["commit", "rollback"]), patch("frappe.enqueue") as enqueue:
|
||||
message_log = bulk_rename(self.test_doctype, input_data, via_console=False)
|
||||
self.assertEqual(len(message_log), len(self.available_documents))
|
||||
self.assertIsInstance(message_log, list)
|
||||
enqueue.assert_called_with(
|
||||
'frappe.utils.global_search.rebuild_for_doctype', doctype=self.test_doctype,
|
||||
)
|
||||
|
||||
def test_deprecated_utils(self):
|
||||
stdout = StringIO()
|
||||
|
||||
with redirect_stdout(stdout), patch_db(["set_value"]):
|
||||
get_fetch_fields("User", "ToDo", ["Activity Log"])
|
||||
self.assertTrue("Function frappe.model.rename_doc.get_fetch_fields" in stdout.getvalue())
|
||||
|
||||
update_linked_doctypes("User", "ToDo", "str", "str")
|
||||
self.assertTrue("Function frappe.model.rename_doc.update_linked_doctypes" in stdout.getvalue())
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue