Merge pull request #57323 from frappe/version-15-hotfix

chore: release v15
This commit is contained in:
Diptanil Saha
2026-07-22 05:45:46 +05:30
committed by GitHub
60 changed files with 2648 additions and 444 deletions

View File

@@ -119,6 +119,7 @@ class Account(NestedSet):
self.validate_account_currency()
self.validate_root_company_and_sync_account_to_children()
self.validate_receivable_payable_account_type()
self.validate_stock_account_type_change()
def validate_parent_child_account_type(self):
if self.parent_account:
@@ -207,6 +208,36 @@ class Account(NestedSet):
frappe.msgprint(msg)
self.add_comment("Comment", msg)
def validate_stock_account_type_change(self):
doc_before_save = self.get_doc_before_save()
if not (doc_before_save and doc_before_save.account_type == "Stock"):
return
if self.account_type == "Stock":
return
if self.stock_ledger_entry_exists():
frappe.throw(
_(
"The account type of {0} cannot be changed from {1} because stock ledger entries exist against it."
).format(frappe.bold(self.name), frappe.bold(_("Stock")))
)
def stock_ledger_entry_exists(self):
from erpnext.stock import get_warehouse_account_map
warehouse_account = get_warehouse_account_map(self.company)
warehouses = [wh for wh, details in warehouse_account.items() if details.account == self.name]
if not warehouses:
return False
return bool(
frappe.db.count(
"Stock Ledger Entry",
filters={"warehouse": ("in", warehouses), "is_cancelled": 0},
)
)
def validate_root_details(self):
doc_before_save = self.get_doc_before_save()

View File

@@ -313,6 +313,31 @@ class TestAccount(unittest.TestCase):
acc.account_currency = "USD"
self.assertRaises(frappe.ValidationError, acc.save)
def test_stock_account_type_change_with_ledger_entries(self):
from erpnext.stock.doctype.stock_entry.stock_entry_utils import make_stock_entry
company = "_Test Company with perpetual inventory"
warehouse = "Stores - TCP1"
stock_account = get_warehouse_account(frappe.get_doc("Warehouse", warehouse))
make_stock_entry(
item_code="_Test Item",
target=warehouse,
company=company,
qty=5,
basic_rate=100,
)
account = frappe.get_doc("Account", stock_account)
self.assertEqual(account.account_type, "Stock")
account.account_type = ""
self.assertRaises(frappe.ValidationError, account.save)
account.reload()
account.account_name = f"{account.account_name} Updated"
account.save() # non-type change stays allowed
def test_account_balance(self):
from erpnext.accounts.utils import get_balance_on

View File

@@ -215,6 +215,7 @@ def build_forest(data):
for row in data:
account_name, parent_account, account_number, parent_account_number = row[0:4]
if account_number:
account_number = cstr(account_number).strip()
account_name = f"{account_number} - {account_name}"
if parent_account_number:
parent_account_number = cstr(parent_account_number).strip()

View File

@@ -169,23 +169,10 @@ frappe.ui.form.on("Dunning", {
},
get_dunning_letter_text: function (frm) {
if (frm.doc.dunning_type) {
frappe.call({
method: "erpnext.accounts.doctype.dunning.dunning.get_dunning_letter_text",
args: {
dunning_type: frm.doc.dunning_type,
language: frm.doc.language,
doc: frm.doc,
},
callback: function (r) {
if (r.message) {
frm.set_value("body_text", r.message.body_text);
frm.set_value("closing_text", r.message.closing_text);
frm.set_value("language", r.message.language);
} else {
frm.set_value("body_text", "");
frm.set_value("closing_text", "");
}
},
frm.call("get_dunning_letter_text").then((r) => {
if (!r.exc) {
frm.refresh_fields();
}
});
}
},

View File

@@ -156,6 +156,46 @@ class Dunning(AccountsController):
"Serial and Batch Bundle",
]
@frappe.whitelist()
def get_dunning_letter_text(self):
DOCTYPE = "Dunning Letter Text"
FIELDS = ["body_text", "closing_text", "language"]
if not self.dunning_type:
return
filters = {"parent": self.dunning_type, "is_default_language": 1}
if self.language:
filters.pop("is_default_language")
filters["language"] = self.language
letter_text = frappe.db.get_value(DOCTYPE, filters, FIELDS, as_dict=True)
if not letter_text:
msg = (
_("Dunning Letter for Dunning Type {0} in language '{1}' not found.").format(
frappe.bold(self.dunning_type), frappe.bold(self.language)
)
if self.language
else _("Dunning Letter for Dunning Type {0} not found.").format(
frappe.bold(self.dunning_type)
)
)
frappe.msgprint(msg, alert=True, indicator="yellow")
self.body_text = (
frappe.render_template(letter_text.body_text, self.as_dict(), restrict_globals=True)
if letter_text
else None
)
self.closing_text = (
frappe.render_template(letter_text.closing_text, self.as_dict(), restrict_globals=True)
if letter_text
else None
)
self.language = letter_text.language if letter_text else self.language
def update_linked_dunnings(doc, previous_outstanding_amount):
if (
@@ -234,35 +274,3 @@ def get_linked_dunnings_as_per_state(sales_invoice, state):
& (overdue_payment.sales_invoice == sales_invoice)
)
).run(as_dict=True)
@frappe.whitelist()
def get_dunning_letter_text(dunning_type: str, doc: str | dict, language: str | None = None) -> dict:
DOCTYPE = "Dunning Letter Text"
FIELDS = ["body_text", "closing_text", "language"]
if isinstance(doc, str):
doc = json.loads(doc)
if not language:
language = doc.get("language")
letter_text = None
if language:
letter_text = frappe.db.get_value(
DOCTYPE, {"parent": dunning_type, "language": language}, FIELDS, as_dict=1
)
if not letter_text:
letter_text = frappe.db.get_value(
DOCTYPE, {"parent": dunning_type, "is_default_language": 1}, FIELDS, as_dict=1
)
if not letter_text:
return {}
return {
"body_text": frappe.render_template(letter_text.body_text, doc),
"closing_text": frappe.render_template(letter_text.closing_text, doc),
"language": letter_text.language,
}

View File

@@ -3,7 +3,10 @@
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import comma_and
from frappe.utils.jinja import validate_template
class DunningType(Document):
@@ -30,3 +33,134 @@ class DunningType(Document):
def autoname(self):
company_abbr = frappe.get_value("Company", self.company, "abbr")
self.name = f"{self.dunning_type} - {company_abbr}"
def validate(self):
self.validate_dunning_letter_text()
self.validate_income_account()
self.validate_cost_center()
self.set_default_dunning_type()
def validate_dunning_letter_text(self):
self.validate_languages()
self.validate_is_default_language()
self.validate_dunning_letter_text_templates()
def validate_income_account(self):
if not self.income_account:
return
account = frappe.get_cached_doc("Account", self.income_account)
msg = []
if account.company != self.company:
msg.append(
_(
"{0} doesn't belong to Company {1}. Please select an Income Account that belongs to Company {1}."
).format(frappe.bold(self.income_account), frappe.bold(self.company))
)
if account.disabled:
msg.append(
_("{0} is disabled. Please select a valid Income Account.").format(
frappe.bold(self.income_account)
)
)
if account.root_type != "Income":
msg.append(
_("{0} is not an Income Account. Please select a valid Income Account.").format(
frappe.bold(self.income_account)
)
)
if account.is_group:
msg.append(
_("{0} is a group account. Please select a non-group Income Account.").format(
frappe.bold(self.income_account)
)
)
if msg:
frappe.msgprint(
msg,
title=_("Income Account Validation Error"),
as_list=True,
raise_exception=frappe.ValidationError,
)
def validate_cost_center(self):
if not self.cost_center:
return
cost_center = frappe.get_cached_doc("Cost Center", self.cost_center)
msg = []
if cost_center.company != self.company:
msg.append(
_(
"{0} doesn't belong to Company {1}. Please select a Cost Center that belongs to Company {1}."
).format(frappe.bold(self.cost_center), frappe.bold(self.company))
)
if cost_center.disabled:
msg.append(
_("{0} is disabled. Please select an enabled Cost Center.").format(
frappe.bold(self.cost_center)
)
)
if cost_center.is_group:
msg.append(
_("{0} is a group Cost Center. Please select a non-group Cost Center.").format(
frappe.bold(self.cost_center)
)
)
if msg:
frappe.msgprint(
msg,
title=_("Cost Center Validation Error"),
as_list=True,
raise_exception=frappe.ValidationError,
)
def validate_languages(self):
languages = [d.language for d in self.dunning_letter_text]
if len(languages) == len(set(languages)):
return
frappe.throw(_("Duplicate languages found on Dunning Letter Text. Keep only one of them."))
def validate_is_default_language(self):
is_default_language_list = [
d.language for d in self.dunning_letter_text if d.is_default_language == 1
]
if len(is_default_language_list) <= 1:
return
frappe.throw(
_("{0} languages are marked as default languages. Please select only one of them.").format(
comma_and(is_default_language_list, add_quotes=True)
)
)
def validate_dunning_letter_text_templates(self):
for d in self.dunning_letter_text:
if d.body_text:
validate_template(d.body_text, restrict_globals=True)
if d.closing_text:
validate_template(d.closing_text, restrict_globals=True)
def set_default_dunning_type(self):
if self.is_default != 1:
return
frappe.db.set_value(
"Dunning Type",
{"company": self.company, "is_default": 1, "name": ["!=", self.name]},
"is_default",
0,
)

View File

@@ -1,9 +1,199 @@
# Copyright (c) 2020, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
# import frappe
import unittest
import frappe
from frappe.tests.utils import FrappeTestCase
class TestDunningType(unittest.TestCase):
pass
def make_dunning_type(dunning_type, company="_Test Company", **kwargs):
doc = frappe.new_doc("Dunning Type")
doc.dunning_type = dunning_type
doc.company = company
doc.dunning_fee = kwargs.get("dunning_fee", 100)
doc.rate_of_interest = kwargs.get("rate_of_interest", 5)
doc.is_default = kwargs.get("is_default", 0)
if "income_account" in kwargs:
doc.income_account = kwargs["income_account"]
elif kwargs.get("income_account") is not False:
doc.income_account = "Sales - _TC" if company == "_Test Company" else "Sales - _TC1"
if "cost_center" in kwargs:
doc.cost_center = kwargs["cost_center"]
elif kwargs.get("cost_center") is not False:
doc.cost_center = "Main - _TC" if company == "_Test Company" else "Main - _TC1"
for row in kwargs.get("dunning_letter_text", [{"language": "en", "body_text": "Test body"}]):
doc.append("dunning_letter_text", row)
return doc
class TestDunningType(FrappeTestCase):
def test_income_account_must_belong_to_company(self):
doc = make_dunning_type("_Test Dunning Wrong Company Account", income_account="Sales - _TC1")
self.assertRaisesRegex(frappe.ValidationError, "doesn't belong to Company", doc.insert)
def test_income_account_must_not_be_disabled(self):
disabled_account = frappe.get_doc(
{
"doctype": "Account",
"account_name": "_Test Disabled Income Account",
"parent_account": "Direct Income - _TC",
"company": "_Test Company",
"account_type": "Income Account",
"disabled": 1,
}
).insert()
doc = make_dunning_type("_Test Dunning Disabled Account", income_account=disabled_account.name)
self.assertRaisesRegex(frappe.ValidationError, "is disabled", doc.insert)
def test_income_account_must_be_income_type(self):
doc = make_dunning_type("_Test Dunning Non Income Account", income_account="Debtors - _TC")
self.assertRaisesRegex(frappe.ValidationError, "is not an Income Account", doc.insert)
def test_income_account_must_not_be_group(self):
doc = make_dunning_type("_Test Dunning Group Account", income_account="Income - _TC")
self.assertRaisesRegex(frappe.ValidationError, "is a group account", doc.insert)
def test_income_account_is_optional(self):
doc = make_dunning_type("_Test Dunning No Income Account", income_account=False)
doc.insert()
self.assertFalse(doc.income_account)
def test_valid_income_account_passes(self):
doc = make_dunning_type("_Test Dunning Valid Income Account", income_account="Sales - _TC")
doc.insert()
self.assertEqual(doc.income_account, "Sales - _TC")
def test_cost_center_must_belong_to_company(self):
doc = make_dunning_type("_Test Dunning Wrong Company CC", cost_center="Main - _TC1")
self.assertRaisesRegex(frappe.ValidationError, "doesn't belong to Company", doc.insert)
def test_cost_center_must_not_be_disabled(self):
disabled_cc = frappe.get_doc(
{
"doctype": "Cost Center",
"cost_center_name": "_Test Disabled Cost Center",
"parent_cost_center": "_Test Company - _TC",
"company": "_Test Company",
"disabled": 1,
}
).insert()
doc = make_dunning_type("_Test Dunning Disabled CC", cost_center=disabled_cc.name)
self.assertRaisesRegex(frappe.ValidationError, "is disabled", doc.insert)
def test_cost_center_must_not_be_group(self):
doc = make_dunning_type("_Test Dunning Group CC", cost_center="_Test Company - _TC")
self.assertRaisesRegex(frappe.ValidationError, "is a group Cost Center", doc.insert)
def test_cost_center_is_optional(self):
doc = make_dunning_type("_Test Dunning No CC", cost_center=False)
doc.insert()
self.assertFalse(doc.cost_center)
def test_valid_cost_center_passes(self):
doc = make_dunning_type("_Test Dunning Valid CC", cost_center="Main - _TC")
doc.insert()
self.assertEqual(doc.cost_center, "Main - _TC")
def test_duplicate_languages_not_allowed(self):
doc = make_dunning_type(
"_Test Dunning Duplicate Language",
dunning_letter_text=[
{"language": "en", "body_text": "Body one"},
{"language": "en", "body_text": "Body two"},
],
)
self.assertRaisesRegex(frappe.ValidationError, "Duplicate languages found", doc.insert)
def test_unique_languages_allowed(self):
doc = make_dunning_type(
"_Test Dunning Unique Languages",
dunning_letter_text=[
{"language": "en", "body_text": "Body one"},
{"language": "de", "body_text": "Body two"},
],
)
doc.insert()
self.assertEqual(len(doc.dunning_letter_text), 2)
def test_only_one_default_language_allowed(self):
doc = make_dunning_type(
"_Test Dunning Multiple Default Language",
dunning_letter_text=[
{"language": "en", "body_text": "Body one", "is_default_language": 1},
{"language": "de", "body_text": "Body two", "is_default_language": 1},
],
)
self.assertRaisesRegex(
frappe.ValidationError, "languages are marked as default languages", doc.insert
)
def test_single_default_language_allowed(self):
doc = make_dunning_type(
"_Test Dunning Single Default Language",
dunning_letter_text=[
{"language": "en", "body_text": "Body one", "is_default_language": 1},
{"language": "de", "body_text": "Body two", "is_default_language": 0},
],
)
doc.insert()
self.assertEqual(doc.dunning_letter_text[0].is_default_language, 1)
def test_invalid_jinja_template_in_body_text_raises(self):
doc = make_dunning_type(
"_Test Dunning Invalid Body Template",
dunning_letter_text=[{"language": "en", "body_text": "{{ unclosed"}],
)
self.assertRaisesRegex(frappe.ValidationError, "Syntax error in template", doc.insert)
def test_invalid_jinja_template_in_closing_text_raises(self):
doc = make_dunning_type(
"_Test Dunning Invalid Closing Template",
dunning_letter_text=[
{"language": "en", "body_text": "Valid body", "closing_text": "{{ unclosed"}
],
)
self.assertRaisesRegex(frappe.ValidationError, "Syntax error in template", doc.insert)
def test_valid_jinja_template_passes(self):
doc = make_dunning_type(
"_Test Dunning Valid Template",
dunning_letter_text=[
{
"language": "en",
"body_text": "Outstanding amount is {{ outstanding_amount }}",
"closing_text": "Regards, {{ company }}",
}
],
)
doc.insert()
self.assertTrue(doc.name)
def test_set_default_dunning_type_unsets_previous_default(self):
first = make_dunning_type("_Test Dunning Default One", is_default=1)
first.insert()
self.assertEqual(frappe.db.get_value("Dunning Type", first.name, "is_default"), 1)
second = make_dunning_type("_Test Dunning Default Two", is_default=1)
second.insert()
self.assertEqual(frappe.db.get_value("Dunning Type", first.name, "is_default"), 0)
self.assertEqual(frappe.db.get_value("Dunning Type", second.name, "is_default"), 1)
def test_set_default_dunning_type_scoped_per_company(self):
company_1 = make_dunning_type("_Test Dunning Default Co1", is_default=1)
company_1.insert()
company_2 = make_dunning_type(
"_Test Dunning Default Co2",
company="_Test Company 1",
is_default=1,
)
company_2.insert()
self.assertEqual(frappe.db.get_value("Dunning Type", company_1.name, "is_default"), 1)
self.assertEqual(frappe.db.get_value("Dunning Type", company_2.name, "is_default"), 1)

View File

@@ -97,9 +97,9 @@ class ProcessStatementOfAccounts(Document):
if not self.pdf_name:
self.pdf_name = "{{ customer.customer_name }}"
validate_template(self.subject)
validate_template(self.body)
validate_template(self.pdf_name)
validate_template(self.subject, restrict_globals=True)
validate_template(self.body, restrict_globals=True)
validate_template(self.pdf_name, restrict_globals=True)
if not self.customers:
frappe.throw(_("Customers not selected."))
@@ -501,15 +501,15 @@ def send_emails(document_name, from_scheduler=False, posting_date=None):
if report:
for customer, report_pdf in report.items():
context = get_context(customer, doc)
filename = frappe.render_template(doc.pdf_name, context)
filename = frappe.render_template(doc.pdf_name, context, restrict_globals=True)
attachments = [{"fname": filename + ".pdf", "fcontent": report_pdf}]
recipients, cc = get_recipients_and_cc(customer, doc)
if not recipients:
continue
subject = frappe.render_template(doc.subject, context)
message = frappe.render_template(doc.body, context)
subject = frappe.render_template(doc.subject, context, restrict_globals=True)
message = frappe.render_template(doc.body, context, restrict_globals=True)
if doc.sender:
sender_email = frappe.db.get_value("Email Account", doc.sender, "email_id")

View File

@@ -2884,8 +2884,6 @@ def create_dunning(source_name, target_doc=None, ignore_permissions=False):
from frappe.model.mapper import get_mapped_doc
def postprocess_dunning(source, target):
from erpnext.accounts.doctype.dunning.dunning import get_dunning_letter_text
dunning_type = frappe.db.exists("Dunning Type", {"is_default": 1, "company": source.company})
if dunning_type:
dunning_type = frappe.get_doc("Dunning Type", dunning_type)
@@ -2894,14 +2892,8 @@ def create_dunning(source_name, target_doc=None, ignore_permissions=False):
target.dunning_fee = dunning_type.dunning_fee
target.income_account = dunning_type.income_account
target.cost_center = dunning_type.cost_center
letter_text = get_dunning_letter_text(
dunning_type=dunning_type.name, doc=target.as_dict(), language=source.language
)
if letter_text:
target.body_text = letter_text.get("body_text")
target.closing_text = letter_text.get("closing_text")
target.language = letter_text.get("language")
target.language = source.language
target.get_dunning_letter_text()
# update outstanding from doc
if source.payment_schedule and len(source.payment_schedule) == 1:

View File

@@ -28,16 +28,16 @@
<br>{%= __("Clearance Date") %}: {%= frappe.datetime.str_to_user(data[i]["clearance_date"]) %}
{% } %}
</td>
<td style="text-align: right">{%= format_currency(data[i]["debit"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["credit"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["debit"], data[i]["account_currency"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["credit"], data[i]["account_currency"]) %}</td>
</tr>
{% } else { %}
<tr>
<td></td>
<td></td>
<td>{%= data[i]["payment_entry"] %}</td>
<td style="text-align: right">{%= format_currency(data[i]["debit"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["credit"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["debit"], data[i]["account_currency"]) %}</td>
<td style="text-align: right">{%= format_currency(data[i]["credit"], data[i]["account_currency"]) %}</td>
</tr>
{% } %}
{% } %}

View File

@@ -16,7 +16,10 @@ from erpnext.accounts.party import (
validate_party_accounts,
validate_party_currency_before_merging,
)
from erpnext.controllers.website_list_for_contact import add_role_for_portal_user
from erpnext.controllers.website_list_for_contact import (
add_role_for_portal_user,
link_portal_users_to_contacts,
)
from erpnext.utilities.transaction_base import TransactionBase
@@ -103,6 +106,7 @@ class Supplier(TransactionBase):
def on_update(self):
self.create_primary_contact()
self.create_primary_address()
link_portal_users_to_contacts(self)
def add_role_for_user(self):
for portal_user in self.portal_users:

View File

@@ -204,7 +204,28 @@ class TestSupplierPortal(FrappeTestCase):
supplier.append("portal_users", {"user": user})
supplier.save()
frappe.set_user(user)
_, suppliers = get_customers_suppliers("Purchase Order", user)
with self.set_user(user):
_, suppliers = get_customers_suppliers("Purchase Order", user)
self.assertIn(supplier.name, suppliers)
def test_portal_user_contact_link(self):
user_email = frappe.generate_hash() + "@example.com"
user = frappe.new_doc("User")
user.email = user_email
user.first_name = "Test Portal Contact User"
user.send_welcome_email = False
user.insert(ignore_permissions=True)
contact = frappe.new_doc("Contact")
contact.first_name = "Test Portal Contact User"
contact.add_email(user_email, is_primary=1)
contact.links = []
contact.insert(ignore_permissions=True)
supplier = create_supplier()
supplier.append("portal_users", {"user": user.name})
supplier.save()
contact.reload()
self.assertTrue(contact.has_link("Supplier", supplier.name))

View File

@@ -307,6 +307,7 @@
"fieldname": "net_rate",
"fieldtype": "Currency",
"label": "Net Rate",
"options": "currency",
"print_hide": 1,
"read_only": 1
},
@@ -613,7 +614,7 @@
"index_web_pages_for_search": 1,
"istable": 1,
"links": [],
"modified": "2025-06-17 12:05:52.441645",
"modified": "2026-07-15 10:33:24.855979",
"modified_by": "Administrator",
"module": "Buying",
"name": "Supplier Quotation Item",

View File

@@ -1513,9 +1513,10 @@ class StockController(AccountsController):
@frappe.whitelist()
def show_accounting_ledger_preview(company, doctype, docname):
def show_accounting_ledger_preview(company: str, doctype: str, docname: str):
filters = frappe._dict(company=company, include_dimensions=1)
doc = frappe.get_doc(doctype, docname)
doc.check_permission("read")
doc.run_method("before_gl_preview")
gl_columns, gl_data = get_accounting_ledger_preview(doc, filters)
@@ -1526,9 +1527,10 @@ def show_accounting_ledger_preview(company, doctype, docname):
@frappe.whitelist()
def show_stock_ledger_preview(company, doctype, docname):
def show_stock_ledger_preview(company: str, doctype: str, docname: str):
filters = frappe._dict(company=company)
doc = frappe.get_doc(doctype, docname)
doc.check_permission("read")
doc.run_method("before_sl_preview")
sl_columns, sl_data = get_stock_ledger_preview(doc, filters)

View File

@@ -7,6 +7,8 @@ import json
import frappe
from frappe import _
from frappe.modules.utils import get_module_app
from frappe.query_builder import Criterion
from frappe.query_builder.functions import Lower
from frappe.utils import cint, flt, has_common
from frappe.utils.user import is_website_user
@@ -306,3 +308,63 @@ def add_role_for_portal_user(portal_user, role):
user_doc.add_roles(role)
frappe.msgprint(_("Added {1} Role to User {0}.").format(frappe.bold(user_doc.name), role), alert=True)
def link_portal_users_to_contacts(doc):
"""When portal users are added to Supplier/Customer, link them to the Contact profile."""
# a User's name is its (lowercased) email, so portal_users are already the emails
portal_users = {p.user for p in doc.get("portal_users") or [] if p.user}
if not portal_users:
return
before = doc.get_doc_before_save()
if before:
previous_users = {p.user for p in before.get("portal_users") or [] if p.user}
if portal_users == previous_users:
return
portal_users = list(portal_users)
contact = frappe.qb.DocType("Contact")
contact_email = frappe.qb.DocType("Contact Email")
query = (
frappe.qb.from_(contact)
.left_join(contact_email)
.on(contact_email.parent == contact.name)
.select(contact.name)
.distinct()
)
conditions = [
contact.user.isin(portal_users),
Lower(contact.email_id).isin(portal_users),
Lower(contact_email.email_id).isin(portal_users),
]
query = query.where(Criterion.any(conditions))
contacts = query.run(pluck=True)
if not contacts:
return
dynamic_link = frappe.qb.DocType("Dynamic Link")
existing_links = (
frappe.qb.from_(dynamic_link)
.select(dynamic_link.parent)
.where(
(dynamic_link.parenttype == "Contact")
& (dynamic_link.parent.isin(contacts))
& (dynamic_link.link_doctype == doc.doctype)
& (dynamic_link.link_name == doc.name)
)
.run(pluck=True)
)
contacts_to_link = [name for name in contacts if name not in existing_links]
for name in contacts_to_link:
contact_doc = frappe.get_doc("Contact", name)
if not contact_doc.has_link(doc.doctype, doc.name):
contact_doc.append("links", {"link_doctype": doc.doctype, "link_name": doc.name})
contact_doc.save(ignore_permissions=True)

View File

@@ -7,7 +7,11 @@
"engine": "InnoDB",
"field_order": [
"scheduled_time",
"column_break_xaox",
"status",
"created_through_portal",
"email_verified",
"verification_token",
"customer_details_section",
"customer_name",
"customer_phone_number",
@@ -54,7 +58,8 @@
"fieldtype": "Datetime",
"in_list_view": 1,
"label": "Scheduled Time",
"reqd": 1
"reqd": 1,
"search_index": 1
},
{
"fieldname": "status",
@@ -77,6 +82,7 @@
"fieldname": "customer_email",
"fieldtype": "Data",
"label": "Email",
"options": "Email",
"reqd": 1
},
{
@@ -99,14 +105,43 @@
"fieldtype": "Dynamic Link",
"label": "Party",
"options": "appointment_with"
},
{
"default": "0",
"fieldname": "created_through_portal",
"fieldtype": "Check",
"label": "Created through Portal",
"read_only": 1,
"set_only_once": 1
},
{
"fieldname": "column_break_xaox",
"fieldtype": "Column Break"
},
{
"default": "0",
"depends_on": "eval:doc.created_through_portal === 1;",
"fieldname": "email_verified",
"fieldtype": "Check",
"label": "Email Verified",
"read_only": 1
},
{
"fieldname": "verification_token",
"fieldtype": "Data",
"label": "Verification Token",
"hidden": 1,
"read_only": 1,
"no_copy": 1,
"search_index": 1
}
],
"links": [],
"modified": "2022-12-15 11:11:02.131986",
"modified": "2026-07-20 02:00:00.000000",
"modified_by": "Administrator",
"module": "CRM",
"name": "Appointment",
"name_case": "UPPER CASE",
"naming_rule": "Expression (old style)",
"owner": "Administrator",
"permissions": [
{
@@ -158,8 +193,9 @@
}
],
"quick_entry": 1,
"sort_field": "modified",
"row_format": "Dynamic",
"sort_field": "creation",
"sort_order": "DESC",
"states": [],
"track_changes": 1
}
}

View File

@@ -3,14 +3,20 @@
from collections import Counter
from datetime import timedelta
from urllib.parse import urlencode
import frappe
from frappe import _
from frappe.desk.form.assign_to import add as add_assignment
from frappe.model.document import Document
from frappe.share import add_docshare
from frappe.utils import get_url, getdate, now
from frappe.utils.verified_command import get_signed_params
from frappe.utils import add_to_date, cint, date_diff, get_datetime, get_url, getdate, now, now_datetime
from frappe.utils.data import sha256_hash
from erpnext.setup.doctype.holiday_list.holiday_list import is_holiday
WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
class Appointment(Document):
@@ -24,103 +30,227 @@ class Appointment(Document):
appointment_with: DF.Link | None
calendar_event: DF.Link | None
created_through_portal: DF.Check
customer_details: DF.LongText | None
customer_email: DF.Data
customer_name: DF.Data
customer_phone_number: DF.Data | None
customer_skype: DF.Data | None
email_verified: DF.Check
party: DF.DynamicLink | None
scheduled_time: DF.Datetime
status: DF.Literal["Open", "Unverified", "Closed"]
verification_token: DF.Data | None
# end: auto-generated types
def find_lead_by_email(self):
lead_list = frappe.get_list(
"Lead", filters={"email_id": self.customer_email}, ignore_permissions=True
)
if lead_list:
return lead_list[0].name
return None
def validate(self):
self.validate_status_update()
if not self.has_value_changed("scheduled_time"):
return
def find_customer_by_email(self):
customer_list = frappe.get_list(
"Customer", filters={"email_id": self.customer_email}, ignore_permissions=True
self.validate_backdated_booking()
if is_appointment_scheduling_enabled():
self.validate_advanced_booking()
self.validate_holiday()
self.validate_slot_timing()
self.validate_available_time_slot()
def validate_status_update(self):
if not self.has_value_changed("status"):
return
if not self.created_through_portal:
if self.status == "Unverified":
frappe.throw(_("Appointments created manually cannot have 'Unverified' status."))
return
if self.status == "Unverified" and self.email_verified:
frappe.throw(_("A verified appointment cannot be moved back to 'Unverified' status."))
if self.status == "Open" and not self.email_verified:
frappe.throw(
_("An appointment booked through the portal can only be opened via email verification.")
)
def validate_backdated_booking(self):
if get_datetime(self.scheduled_time) < now_datetime():
frappe.throw(_("Appointment cannot be scheduled for a past time."))
def validate_advanced_booking(self):
advance_booking_days = cint(get_booking_settings().advance_booking_days)
if advance_booking_days and date_diff(self.scheduled_time, now_datetime()) > advance_booking_days:
frappe.throw(
_("Appointment can only be scheduled up to {0} day(s) in advance.").format(
advance_booking_days
)
)
def validate_holiday(self):
holiday_list = get_booking_settings().holiday_list
if not holiday_list:
frappe.throw(_("Please add a valid Holiday List on Appointment Booking Settings."))
if is_holiday(holiday_list, getdate(self.scheduled_time)):
frappe.throw(_("Appointment cannot be scheduled on a holiday."))
def validate_slot_timing(self):
settings = get_booking_settings()
if not settings.availability_of_slots:
frappe.throw(_("No availability of slots are found. Please add on Appointment Booking Settings."))
scheduled_time = get_datetime(self.scheduled_time)
day_of_week = WEEKDAYS[scheduled_time.weekday()]
slot_start = timedelta(
hours=scheduled_time.hour, minutes=scheduled_time.minute, seconds=scheduled_time.second
)
if customer_list:
return customer_list[0].name
return None
slot_end = slot_start + timedelta(minutes=cint(settings.appointment_duration))
for slot in settings.availability_of_slots:
if slot.day_of_week == day_of_week and slot.from_time <= slot_start and slot_end <= slot.to_time:
return
frappe.throw(_("Appointment must be scheduled within the available slot timings."))
def validate_available_time_slot(self):
settings = get_booking_settings()
if not cint(settings.number_of_agents):
return
# the locking read serializes concurrent bookings for the same window,
# so two simultaneous requests cannot both pass the capacity check
booked = count_overlapping_appointments(
self.scheduled_time,
cint(settings.appointment_duration),
exclude_appointment=self.name,
for_update=True,
)
if booked >= cint(settings.number_of_agents):
frappe.throw(_("Time slot is not available"))
def before_insert(self):
number_of_appointments_in_same_slot = frappe.db.count(
"Appointment", filters={"scheduled_time": self.scheduled_time}
)
number_of_agents = frappe.db.get_single_value("Appointment Booking Settings", "number_of_agents")
if not number_of_agents == 0:
if number_of_appointments_in_same_slot >= number_of_agents:
frappe.throw(_("Time slot is not available"))
# Link lead
if not self.party:
lead = self.find_lead_by_email()
customer = self.find_customer_by_email()
if customer:
self.appointment_with = "Customer"
self.party = customer
else:
self.appointment_with = "Lead"
self.party = lead
# Set status to "Unverified" for new Appointments.
if self.created_through_portal:
self.status = "Unverified"
return
self.link_customer_lead()
def after_insert(self):
if self.party:
# Create Calendar event
if not self.created_through_portal and self.party:
self.auto_assign()
self.create_calendar_event()
else:
# Set status to unverified
self.status = "Unverified"
# Send email to confirm
self.send_confirmation_email()
return
# Send email to confirm
self.send_confirmation_email()
def on_update(self):
# capture transitions before nested saves during materialization
# refresh the before-save snapshot
status_changed = self.has_value_changed("status")
email_just_verified = bool(
self.created_through_portal and self.email_verified
) and self.has_value_changed("email_verified")
self.link_auto_assign_and_create_calendar_event()
if email_just_verified:
self.send_appointment_confirmed_email()
if status_changed:
self.update_event_and_assignments_status()
def on_trash(self):
# the Event only references the party, not the appointment,
# so it must be cleaned up explicitly
if not self.calendar_event:
return
event = self.calendar_event
self.db_set("calendar_event", None, update_modified=False)
frappe.delete_doc("Event", event, ignore_permissions=True)
def send_confirmation_email(self):
verify_url = self._get_verify_url()
template = "confirm_appointment"
args = {
"link": verify_url,
"site_url": frappe.utils.get_url(),
"full_name": self.customer_name,
}
self.send_email_to_customer(
template="confirm_appointment",
subject=_("Appointment Confirmation"),
args={"link": self._get_verify_url(), "expiry_minutes": get_verification_link_expiry()},
)
frappe.msgprint(_("Please check your email to confirm the appointment."))
def send_appointment_confirmed_email(self):
self.send_email_to_customer(
template="appointment_confirmed",
subject=_("Appointment Confirmed"),
args={"scheduled_time": frappe.utils.format_datetime(self.scheduled_time)},
reference_doctype="Appointment",
reference_name=self.name,
)
def send_email_to_customer(self, template, subject, args, **kwargs):
frappe.sendmail(
recipients=[self.customer_email],
template=template,
args=args,
subject=_("Appointment Confirmation"),
args={"full_name": self.customer_name, "site_url": frappe.utils.get_url(), **args},
subject=subject,
**kwargs,
)
if frappe.session.user == "Guest":
frappe.msgprint(_("Please check your email to confirm the appointment"))
else:
frappe.msgprint(
_("Appointment was created. But no lead was found. Please check the email to confirm")
)
def on_change(self):
# Sync Calendar
if not self.calendar_event:
def link_auto_assign_and_create_calendar_event(self):
if self.is_new() or (self.created_through_portal and not self.email_verified):
return
if not self.calendar_event:
# first materialization: link the party, assign an agent, create the event
self.link_customer_lead()
self.auto_assign()
self.create_calendar_event()
self.sync_calendar_event()
def sync_calendar_event(self):
if not self.calendar_event or not self.has_value_changed("scheduled_time"):
return
cal_event = frappe.get_doc("Event", self.calendar_event)
cal_event.starts_on = self.scheduled_time
cal_event.save(ignore_permissions=True)
def set_verified(self, email):
if not email == self.customer_email:
frappe.throw(_("Email verification failed."))
# Create new lead
def update_event_and_assignments_status(self):
"""Close or reopen the calendar event and assignments along with the appointment."""
if self.status == "Unverified":
return
is_closed = self.status == "Closed"
new_status = "Closed" if is_closed else "Open"
if self.calendar_event:
frappe.db.set_value("Event", self.calendar_event, "status", new_status)
# only move ToDos between Open and Closed - never touch Cancelled ones
todo_filters = {
"reference_type": "Appointment",
"reference_name": self.name,
"status": "Open" if is_closed else "Closed",
}
frappe.db.set_value("ToDo", todo_filters, "status", new_status)
def link_customer_lead(self):
if not self.party:
customer = self.find_party_by_email("Customer")
self.appointment_with = "Customer" if customer else "Lead"
self.party = customer or self.find_party_by_email("Lead")
self.create_lead_and_link()
# Remove unverified status
self.status = "Open"
# Create calender event
self.auto_assign()
self.create_calendar_event()
self.save(ignore_permissions=True)
frappe.db.commit()
def find_party_by_email(self, doctype):
party = frappe.get_all(doctype, filters={"email_id": self.customer_email}, limit=1, pluck="name")
return party[0] if party else None
def create_lead_and_link(self):
# Return if already linked
@@ -139,86 +269,39 @@ class Appointment(Document):
if self.customer_details:
lead.append(
"notes",
{
"note": self.customer_details,
"added_by": frappe.session.user,
"added_on": now(),
},
{"note": self.customer_details, "added_by": frappe.session.user, "added_on": now()},
)
lead.insert(ignore_permissions=True)
# Link lead
self.party = lead.name
self.party = lead.insert(ignore_permissions=True).name
def auto_assign(self):
existing_assignee = self.get_assignee_from_latest_opportunity()
if existing_assignee:
# If the latest opportunity is assigned to someone
# Assign the appointment to the same
self.assign_agent(existing_assignee)
return
if self._assign:
return
available_agents = _get_agents_sorted_by_asc_workload(getdate(self.scheduled_time))
for agent in available_agents:
if _check_agent_availability(agent, self.scheduled_time):
self.assign_agent(agent[0])
break
if existing_assignee := self.get_assignee_from_latest_opportunity():
# assign to whoever handles the party's latest opportunity
self.assign_agent(existing_assignee)
return
busy_agents = get_busy_agents(self.scheduled_time)
for agent in _get_agents_sorted_by_asc_workload(getdate(self.scheduled_time)):
if agent not in busy_agents:
self.assign_agent(agent)
break
def get_assignee_from_latest_opportunity(self):
if not self.party:
if not self.party or not frappe.db.exists("Lead", self.party):
return None
if not frappe.db.exists("Lead", self.party):
return None
opporutnities = frappe.get_list(
opportunities = frappe.get_all(
"Opportunity",
filters={
"party_name": self.party,
},
ignore_permissions=True,
filters={"party_name": self.party},
fields=["_assign"],
order_by="creation desc",
limit=1,
)
if not opporutnities:
return None
latest_opportunity = frappe.get_doc("Opportunity", opporutnities[0].name)
assignee = latest_opportunity._assign
if not assignee:
return None
assignee = frappe.parse_json(assignee)[0]
return assignee
def create_calendar_event(self):
if self.calendar_event:
return
appointment_event = frappe.get_doc(
{
"doctype": "Event",
"subject": " ".join(["Appointment with", self.customer_name]),
"starts_on": self.scheduled_time,
"status": "Open",
"type": "Public",
"send_reminder": frappe.db.get_single_value(
"Appointment Booking Settings", "email_reminders"
),
"event_participants": [
dict(reference_doctype=self.appointment_with, reference_docname=self.party)
],
}
)
employee = _get_employee_from_user(self._assign)
if employee:
appointment_event.append(
"event_participants", dict(reference_doctype="Employee", reference_docname=employee.name)
)
appointment_event.insert(ignore_permissions=True)
self.calendar_event = appointment_event.name
self.save(ignore_permissions=True)
def _get_verify_url(self):
verify_route = "/book_appointment/verify"
params = {"email": self.customer_email, "appointment": self.name}
return get_url(verify_route + "?" + get_signed_params(params))
assignees = opportunities and frappe.parse_json(opportunities[0]._assign or "[]")
return assignees[0] if assignees else None
def assign_agent(self, agent):
if not frappe.has_permission(doc=self, user=agent):
@@ -226,45 +309,157 @@ class Appointment(Document):
add_assignment({"doctype": self.doctype, "name": self.name, "assign_to": [agent]})
def create_calendar_event(self):
if self.calendar_event:
return
event = frappe.get_doc(
{
"doctype": "Event",
"subject": f"Appointment with {self.customer_name}",
"starts_on": self.scheduled_time,
"status": "Open",
"type": "Public",
"send_reminder": cint(get_booking_settings().email_reminders),
"event_participants": self.get_event_participants(),
}
).insert(ignore_permissions=True)
self.calendar_event = event.name
self.save(ignore_permissions=True)
def get_event_participants(self):
participants = [dict(reference_doctype=self.appointment_with, reference_docname=self.party)]
if employee := _get_employee_from_user(self._assign):
participants.append(dict(reference_doctype="Employee", reference_docname=employee.name))
return participants
def _get_verify_url(self):
key = self.generate_verification_key()
return get_url("/book_appointment/verify?" + urlencode({"key": key}))
def generate_verification_key(self):
# store only the hash; the raw key lives solely in the emailed link
key = frappe.generate_hash()
self.db_set("verification_token", sha256_hash(key), update_modified=False)
return key
def get_booking_settings():
return frappe.get_cached_doc("Appointment Booking Settings")
def is_appointment_scheduling_enabled():
return bool(cint(get_booking_settings().enable_scheduling))
def get_verification_link_expiry():
"""Verification link expiry window in minutes."""
return cint(get_booking_settings().verification_link_expiry_duration)
def count_overlapping_appointments(
scheduled_time, appointment_duration, exclude_appointment=None, for_update=False
):
"""Count non-Closed appointments whose duration window overlaps `scheduled_time`.
With `for_update`, the range stays locked until commit, serializing concurrent bookings."""
# select the rows (not COUNT) so `for_update` stays valid: PostgreSQL
# rejects `FOR UPDATE` combined with an aggregate function
appointment = frappe.qb.DocType("Appointment")
query = (
frappe.qb.from_(appointment)
.select(appointment.name)
.where(appointment.scheduled_time > add_to_date(scheduled_time, minutes=-appointment_duration))
.where(appointment.scheduled_time < add_to_date(scheduled_time, minutes=appointment_duration))
.where(appointment.status != "Closed")
)
if exclude_appointment:
query = query.where(appointment.name != exclude_appointment)
if for_update:
query = query.for_update()
return len(query.run())
def handle_expired_unverified_appointments():
"""Close or delete Unverified appointments whose verification link has expired."""
expiry = get_verification_link_expiry()
if not expiry:
return
cutoff = add_to_date(now_datetime(), minutes=-expiry)
filters = {"status": "Unverified", "creation": ("<", cutoff)}
action = get_booking_settings().action_for_expired_unverified_appointments or "Mark as Closed"
if action == "Mark as Closed":
frappe.db.set_value("Appointment", filters, "status", "Closed")
elif action == "Delete Permanently":
for name in frappe.get_all("Appointment", filters=filters, pluck="name"):
frappe.delete_doc("Appointment", name, ignore_permissions=True)
def _get_agents_sorted_by_asc_workload(date):
appointments = frappe.get_all("Appointment", fields="*")
agent_list = _get_agent_list_as_strings()
if not appointments:
return agent_list
appointment_counter = Counter(agent_list)
for appointment in appointments:
assign_data = appointment._assign
if isinstance(assign_data, str):
assign_data = assign_data.strip()
if not assign_data:
continue
assigned_to = frappe.parse_json(assign_data)
if assigned_to and (assigned_to[0] in agent_list) and getdate(appointment.scheduled_time) == date:
appointment_counter[assigned_to[0]] += 1
sorted_agent_list = appointment_counter.most_common()
sorted_agent_list.reverse()
return sorted_agent_list
# count only the given day's assignments; scheduled_time is indexed so the
# date range is resolved in SQL instead of scanning every appointment ever
workload = Counter(agent.user for agent in get_booking_settings().agent_list)
assigns = frappe.get_all(
"Appointment",
filters=[
["_assign", "is", "set"],
["scheduled_time", ">=", getdate(date)],
["scheduled_time", "<", add_to_date(getdate(date), days=1)],
],
pluck="_assign",
)
for assign in assigns:
assignees = frappe.parse_json((assign or "").strip() or "[]")
if assignees and assignees[0] in workload:
workload[assignees[0]] += 1
return [agent for agent, _workload in reversed(workload.most_common())]
def _get_agent_list_as_strings():
agent_list_as_strings = []
agent_list = frappe.get_doc("Appointment Booking Settings").agent_list
for agent in agent_list:
agent_list_as_strings.append(agent.user)
return agent_list_as_strings
def get_busy_agents(scheduled_time):
"""Agents already assigned to a non-Closed appointment overlapping `scheduled_time`."""
duration = _get_appointment_duration()
assigns = frappe.get_all(
"Appointment",
filters=[
["scheduled_time", ">", add_to_date(scheduled_time, minutes=-duration)],
["scheduled_time", "<", add_to_date(scheduled_time, minutes=duration)],
["status", "!=", "Closed"],
],
pluck="_assign",
)
return {assignee for assign in assigns for assignee in frappe.parse_json(assign or "[]")}
def _check_agent_availability(agent_email, scheduled_time):
appointemnts_at_scheduled_time = frappe.get_all("Appointment", filters={"scheduled_time": scheduled_time})
for appointment in appointemnts_at_scheduled_time:
if appointment._assign == agent_email:
return False
return True
return agent_email not in get_busy_agents(scheduled_time)
def get_booked_slot_times(from_time, to_time):
"""scheduled_times of non-Closed appointments within (from_time, to_time), for slot availability."""
return frappe.get_all(
"Appointment",
filters=[
["scheduled_time", ">", from_time],
["scheduled_time", "<", to_time],
["status", "!=", "Closed"],
],
pluck="scheduled_time",
)
def _get_appointment_duration():
return cint(get_booking_settings().appointment_duration)
def _get_employee_from_user(user):
employee_docname = frappe.db.get_value("Employee", {"user_id": user})
if employee_docname:
return frappe.get_doc("Employee", employee_docname)
return None
return frappe.get_doc("Employee", employee_docname) if employee_docname else None

View File

@@ -2,37 +2,175 @@
# See license.txt
import datetime
import unittest
from unittest.mock import patch
from urllib.parse import parse_qs, urlparse
import frappe
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_to_date, getdate, now_datetime, set_request
from frappe.utils.data import get_system_timezone, sha256_hash
from erpnext.crm.doctype.appointment.appointment import (
Appointment,
_check_agent_availability,
handle_expired_unverified_appointments,
)
from erpnext.setup.doctype.holiday_list.test_holiday_list import make_holiday_list
from erpnext.www.book_appointment.index import create_appointment, get_appointment_slots
from erpnext.www.book_appointment.verify import index as verify_index
LEAD_EMAIL = "test_appointment_lead@example.com"
VERIFICATION_EXPIRY_MINUTES = 30
ALL_WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
def create_test_appointment():
test_appointment = frappe.get_doc(
{
"doctype": "Appointment",
"status": "Open",
"customer_name": "Test Lead",
"customer_phone_number": "666",
"customer_skype": "test",
"customer_email": LEAD_EMAIL,
"scheduled_time": datetime.datetime.now(),
"customer_details": "Hello, Friend!",
}
)
def create_test_appointment(**kwargs):
args = {
"doctype": "Appointment",
"status": "Open",
"customer_name": "Test Lead",
"customer_phone_number": "666",
"customer_skype": "test",
"customer_email": LEAD_EMAIL,
"scheduled_time": add_to_date(now_datetime(), hours=2),
"customer_details": "Hello, Friend!",
}
args.update(kwargs)
test_appointment = frappe.get_doc(args)
test_appointment.insert()
return test_appointment
class TestAppointment(unittest.TestCase):
def setUpClass():
frappe.db.delete("Lead", {"email_id": LEAD_EMAIL})
def create_lead(email, name="Existing Lead"):
frappe.db.delete("Lead", {"email_id": email})
return frappe.get_doc({"doctype": "Lead", "lead_name": name, "email_id": email}).insert(
ignore_permissions=True
)
def set_booking_setting(field, value):
frappe.db.set_single_value("Appointment Booking Settings", field, value)
def slot_on(days_from_now, hour, minute=0):
day = datetime.date.today() + datetime.timedelta(days=days_from_now)
return datetime.datetime.combine(day, datetime.time(hour, minute))
def backdate_creation(appointment_name, minutes):
frappe.db.set_value(
"Appointment",
appointment_name,
"creation",
add_to_date(now_datetime(), minutes=-minutes),
update_modified=False,
)
def get_status(appointment_name):
return frappe.db.get_value("Appointment", appointment_name, "status")
def get_assignees(appointment_name):
return frappe.parse_json(frappe.db.get_value("Appointment", appointment_name, "_assign") or "[]")
def get_todo_statuses(appointment_name):
return frappe.get_all(
"ToDo",
filters={"reference_type": "Appointment", "reference_name": appointment_name},
pluck="status",
)
def parse_verify_url(verify_url):
parsed = urlparse(verify_url)
return parsed, {key: value[0] for key, value in parse_qs(parsed.query).items()}
class TestAppointment(FrappeTestCase):
def setUp(self):
# sending an email commits the transaction (EmailQueue sets its status
# with commit=True), which would break the per-test rollback below
frappe.flags.mute_emails = 1
set_booking_setting("verification_link_expiry_duration", VERIFICATION_EXPIRY_MINUTES)
frappe.db.delete("Lead", {"email_id": LEAD_EMAIL})
self.test_appointment = create_test_appointment()
self.test_appointment.set_verified(self.test_appointment.customer_email)
def tearDown(self):
frappe.db.rollback()
frappe.clear_document_cache("Appointment Booking Settings", "Appointment Booking Settings")
frappe.flags.mute_emails = 0
def _configure_booking_settings(self, holiday_dates=None, agents=None):
holiday_list = make_holiday_list(
"_Test Appointment Holiday List",
from_date=getdate(),
to_date=add_to_date(getdate(), days=60),
holiday_dates=holiday_dates or [],
)
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 1
settings.enable_appointment_portal = 1
settings.appointment_duration = 30
settings.advance_booking_days = 30
settings.verification_link_expiry_duration = VERIFICATION_EXPIRY_MINUTES
settings.holiday_list = holiday_list.name
settings.set("agent_list", [])
for agent in agents or ["Administrator"]:
settings.append("agent_list", {"user": agent})
settings.set("availability_of_slots", [])
for day in ALL_WEEKDAYS:
settings.append(
"availability_of_slots", {"day_of_week": day, "from_time": "09:00:00", "to_time": "17:00:00"}
)
settings.save()
def _create_portal_appointment(self, email, days_from_now=7, time="10:00:00"):
"""Book as Guest. The verification email is mocked and kept on
``self._verification_email_mock`` for assertions."""
if not getattr(self, "_booking_settings_configured", False):
self._configure_booking_settings()
self._booking_settings_configured = True
with self.set_user("Guest"), patch.object(Appointment, "send_confirmation_email") as mock_send:
appointment = create_appointment(
date=str(datetime.date.today() + datetime.timedelta(days=days_from_now)),
time=time,
tz=get_system_timezone(),
contact={"name": "Portal Visitor", "email": email, "number": "123", "skype": "", "notes": ""},
)
self._verification_email_mock = mock_send
return appointment
def _request_verification(self, appointment, verify_url=None):
"""Simulate the GET request made by clicking the emailed verification link.
The confirmation email sent on successful verification is mocked and kept
on ``self._confirmed_email_mock`` for assertions.
"""
parsed, params = parse_verify_url(verify_url or appointment._get_verify_url())
old_request = getattr(frappe.local, "request", None)
old_form_dict = frappe.local.form_dict
old_user = frappe.session.user
try:
# the real link is clicked by an anonymous visitor; set_user resets
# form_dict, so switch the user before populating the request
frappe.set_user("Guest")
set_request(method="GET", path=f"{parsed.path}?{parsed.query}")
frappe.local.form_dict = frappe._dict(params)
context = frappe._dict()
with patch.object(Appointment, "send_appointment_confirmed_email") as mock_confirmed:
verify_index.get_context(context)
self._confirmed_email_mock = mock_confirmed
return context
finally:
frappe.set_user(old_user)
frappe.local.request = old_request
frappe.local.form_dict = old_form_dict
frappe.local.flags.commit = False
def test_calendar_event_created(self):
cal_event = frappe.get_doc("Event", self.test_appointment.calendar_event)
@@ -40,3 +178,369 @@ class TestAppointment(unittest.TestCase):
def test_lead_linked(self):
self.assertTrue(self.test_appointment.party)
def test_desk_created_appointment_skips_email_verification(self):
"""Appointments created from the desk (created_through_portal unset) must be
linked and confirmed immediately - no verification email should be sent."""
with patch.object(Appointment, "send_confirmation_email") as mock_send:
appointment = create_test_appointment(customer_email="another_desk_lead@example.com")
mock_send.assert_not_called()
self.assertEqual(appointment.status, "Open")
self.assertTrue(appointment.party)
frappe.db.delete("Lead", {"email_id": "another_desk_lead@example.com"})
def test_portal_booking_stays_unverified_for_existing_lead(self):
"""A portal booking whose email matches an existing Lead/Customer must NOT
be auto-linked - it must stay Unverified until the email is confirmed."""
create_lead("existing_lead@example.com")
appointment = self._create_portal_appointment("existing_lead@example.com", days_from_now=5)
self._verification_email_mock.assert_called_once()
self.assertTrue(appointment.created_through_portal)
self.assertEqual(appointment.status, "Unverified")
self.assertFalse(appointment.email_verified)
self.assertFalse(appointment.party)
def test_verify_url_uses_opaque_token(self):
appointment = self._create_portal_appointment("portal_visitor@example.com")
parsed, params = parse_verify_url(appointment._get_verify_url())
# the link carries only an opaque key - no email, name or signed params
self.assertEqual(set(params), {"key"})
self.assertNotIn("email", parsed.query)
# only the hash of that key is stored on the appointment
stored = frappe.db.get_value("Appointment", appointment.name, "verification_token")
self.assertEqual(stored, sha256_hash(params["key"]))
def test_email_verification_within_expiry_window(self):
# Link used within the validity window - verification succeeds and the
# appointment gets linked, assigned and added to the calendar
on_time = self._create_portal_appointment("portal_visitor_on_time@example.com")
context = self._request_verification(on_time)
self.assertTrue(context.success)
self._confirmed_email_mock.assert_called_once()
on_time.reload()
self.assertEqual(on_time.status, "Open")
self.assertTrue(on_time.email_verified)
self.assertTrue(on_time.party)
self.assertTrue(on_time.calendar_event)
# Link used after the validity window - verification fails
late = self._create_portal_appointment("portal_visitor_late@example.com", days_from_now=10)
after_expiry = add_to_date(now_datetime(), minutes=VERIFICATION_EXPIRY_MINUTES + 1)
with patch.object(verify_index, "now_datetime", return_value=after_expiry):
context = self._request_verification(late)
self.assertFalse(context.success)
self._confirmed_email_mock.assert_not_called()
late.reload()
self.assertEqual(late.status, "Unverified")
self.assertFalse(late.email_verified)
self.assertFalse(late.party)
def test_verification_link_reused_after_success(self):
appointment = self._create_portal_appointment("portal_visitor_twice@example.com")
verify_url = appointment._get_verify_url()
context = self._request_verification(appointment, verify_url=verify_url)
self.assertTrue(context.success)
self._confirmed_email_mock.assert_called_once()
# re-clicking the link is idempotent and does not send another email
context = self._request_verification(appointment, verify_url=verify_url)
self.assertTrue(context.success)
self.assertIn("already verified", context.message)
self._confirmed_email_mock.assert_not_called()
def test_verification_link_for_deleted_appointment(self):
"""A verification link can outlive its appointment - clicking it must
render a friendly message, not crash."""
appointment = self._create_portal_appointment("portal_visitor_gone@example.com")
verify_url = appointment._get_verify_url()
frappe.delete_doc("Appointment", appointment.name, ignore_permissions=True)
context = self._request_verification(appointment, verify_url=verify_url)
self.assertFalse(context.success)
self.assertIn("book the appointment again", context.message)
def test_reschedule_syncs_calendar_event(self):
new_time = add_to_date(self.test_appointment.scheduled_time, hours=1)
self.test_appointment.scheduled_time = new_time
self.test_appointment.save()
starts_on = frappe.db.get_value("Event", self.test_appointment.calendar_event, "starts_on")
self.assertEqual(starts_on, new_time)
def test_portal_endpoint_disabled(self):
self._configure_booking_settings()
set_booking_setting("enable_appointment_portal", 0)
with self.set_user("Guest"), self.assertRaises(frappe.Redirect):
create_appointment(
date=str(datetime.date.today() + datetime.timedelta(days=3)),
time="10:00:00",
tz="UTC",
contact={
"name": "Blocked",
"email": "blocked@example.com",
"number": "1",
"skype": "",
"notes": "",
},
)
def test_booked_slot_unavailable_on_portal(self):
self._configure_booking_settings()
tz = get_system_timezone()
day = datetime.date.today() + datetime.timedelta(days=2)
def get_availability():
with self.set_user("Guest"):
slots = get_appointment_slots(str(day), tz)
return {slot["time"].strftime("%H:%M"): slot["availability"] for slot in slots}
booked = create_test_appointment(
customer_email="slot_taken@example.com", scheduled_time=slot_on(2, 10)
)
availability = get_availability()
self.assertFalse(availability["10:00"])
self.assertTrue(availability["13:00"])
# closing the appointment frees its slot on the portal
booked.status = "Closed"
booked.save()
self.assertTrue(get_availability()["10:00"])
# an off-grid desk appointment blocks every portal slot it overlaps
create_test_appointment(customer_email="off_grid@example.com", scheduled_time=slot_on(2, 13, 15))
availability = get_availability()
self.assertFalse(availability["13:00"])
self.assertFalse(availability["13:30"])
self.assertTrue(availability["14:00"])
def test_expired_unverified_appointments_are_closed(self):
stale = self._create_portal_appointment("portal_visitor_stale@example.com", days_from_now=8)
fresh = self._create_portal_appointment("portal_visitor_fresh@example.com", days_from_now=9)
verify_url = stale._get_verify_url()
backdate_creation(stale.name, VERIFICATION_EXPIRY_MINUTES + 15)
set_booking_setting("action_for_expired_unverified_appointments", "Mark as Closed")
handle_expired_unverified_appointments()
self.assertEqual(get_status(stale.name), "Closed")
self.assertEqual(get_status(fresh.name), "Unverified")
# Open appointments are never touched, regardless of age
self.assertEqual(get_status(self.test_appointment.name), "Open")
# clicking the link of a closed appointment renders a friendly message
context = self._request_verification(stale, verify_url=verify_url)
self.assertFalse(context.success)
self.assertIn("closed", context.message)
def test_expired_unverified_appointments_are_deleted(self):
stale = self._create_portal_appointment("portal_visitor_purged@example.com", days_from_now=8)
fresh = self._create_portal_appointment("portal_visitor_kept@example.com", days_from_now=9)
backdate_creation(stale.name, VERIFICATION_EXPIRY_MINUTES + 15)
set_booking_setting("action_for_expired_unverified_appointments", "Delete Permanently")
handle_expired_unverified_appointments()
self.assertFalse(frappe.db.exists("Appointment", stale.name))
self.assertTrue(frappe.db.exists("Appointment", fresh.name))
self.assertTrue(frappe.db.exists("Appointment", self.test_appointment.name))
def test_cleanup_skipped_when_expiry_not_configured(self):
appointment = self._create_portal_appointment("portal_visitor_no_expiry@example.com")
backdate_creation(appointment.name, 5)
set_booking_setting("verification_link_expiry_duration", 0)
handle_expired_unverified_appointments()
self.assertEqual(get_status(appointment.name), "Unverified")
def test_status_transition_rules(self):
# desk appointments can never be Unverified
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="desk_unverified@example.com", status="Unverified")
# portal appointments cannot be opened manually before verification
unverified = self._create_portal_appointment("manual_open@example.com")
unverified.status = "Open"
with self.assertRaises(frappe.ValidationError):
unverified.save(ignore_permissions=True)
# verified appointments cannot be reverted to Unverified
verified = self._create_portal_appointment("revert_unverified@example.com", days_from_now=8)
self._request_verification(verified)
verified.reload()
verified.status = "Unverified"
with self.assertRaises(frappe.ValidationError):
verified.save(ignore_permissions=True)
# both desk and verified portal appointments can be closed and reopened
for appointment in (self.test_appointment, verified):
appointment.reload()
appointment.status = "Closed"
appointment.save(ignore_permissions=True)
appointment.status = "Open"
appointment.save(ignore_permissions=True)
self.assertEqual(appointment.status, "Open")
def test_agent_auto_assignment(self):
agent_email = "appointment_agent@example.com"
if not frappe.db.exists("User", agent_email):
frappe.get_doc(
{"doctype": "User", "email": agent_email, "first_name": "Appointment Agent"}
).insert(ignore_permissions=True)
self._configure_booking_settings(agents=["Administrator", agent_email])
first = create_test_appointment(
customer_email="assigned_one@example.com", scheduled_time=slot_on(2, 11)
)
second = create_test_appointment(
customer_email="assigned_two@example.com", scheduled_time=slot_on(2, 11)
)
# both appointments in the same slot get an agent, and never the same one
self.assertTrue(get_assignees(first.name))
self.assertTrue(get_assignees(second.name))
self.assertNotEqual(get_assignees(first.name), get_assignees(second.name))
# closing an assigned appointment closes its ToDo without re-assigning
first.reload()
first.status = "Closed"
first.save()
self.assertTrue(get_todo_statuses(first.name))
self.assertTrue(all(status == "Closed" for status in get_todo_statuses(first.name)))
# reopening brings the ToDos back
first.status = "Open"
first.save()
self.assertTrue(all(status == "Open" for status in get_todo_statuses(first.name)))
def test_agent_busy_for_the_whole_appointment_duration(self):
self._configure_booking_settings()
slot = slot_on(3, 11)
appointment = create_test_appointment(customer_email="busy_agent@example.com", scheduled_time=slot)
assignee = get_assignees(appointment.name)[0]
# busy anywhere inside the 30-minute appointment window, free right after it
self.assertFalse(_check_agent_availability(assignee, slot))
self.assertFalse(_check_agent_availability(assignee, slot + datetime.timedelta(minutes=15)))
self.assertTrue(_check_agent_availability(assignee, slot + datetime.timedelta(minutes=30)))
def test_closed_appointment_closes_calendar_event(self):
self.test_appointment.status = "Closed"
self.test_appointment.save()
event_status = frappe.db.get_value("Event", self.test_appointment.calendar_event, "status")
self.assertEqual(event_status, "Closed")
# reopening the appointment reopens the calendar event
self.test_appointment.status = "Open"
self.test_appointment.save()
event_status = frappe.db.get_value("Event", self.test_appointment.calendar_event, "status")
self.assertEqual(event_status, "Open")
def test_deleting_appointment_deletes_calendar_event(self):
event = self.test_appointment.calendar_event
self.assertTrue(frappe.db.exists("Event", event))
frappe.delete_doc("Appointment", self.test_appointment.name)
self.assertFalse(frappe.db.exists("Event", event))
def test_backdated_appointment_is_rejected(self):
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="backdated@example.com",
scheduled_time=add_to_date(now_datetime(), hours=-1),
)
def test_booking_beyond_advance_window_is_rejected(self):
self._configure_booking_settings()
set_booking_setting("advance_booking_days", 7)
# within the advance booking window - allowed
within = create_test_appointment(
customer_email="advance_within@example.com", scheduled_time=slot_on(5, 10)
)
self.assertTrue(frappe.db.exists("Appointment", within.name))
# beyond the advance booking window - rejected
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="advance_beyond@example.com", scheduled_time=slot_on(8, 10)
)
def test_appointment_on_holiday_is_rejected(self):
holiday = add_to_date(getdate(), days=3)
self._configure_booking_settings(
holiday_dates=[{"holiday_date": holiday, "description": "Test Holiday"}]
)
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="on_holiday@example.com", scheduled_time=slot_on(3, 10))
# the day after the holiday is bookable
after_holiday = create_test_appointment(
customer_email="after_holiday@example.com", scheduled_time=slot_on(4, 10)
)
self.assertTrue(frappe.db.exists("Appointment", after_holiday.name))
def test_appointment_outside_slot_timing_is_rejected(self):
self._configure_booking_settings()
# before the slot opens
with self.assertRaises(frappe.ValidationError):
create_test_appointment(customer_email="before_opening@example.com", scheduled_time=slot_on(2, 8))
# starts within the slot but would end after it closes
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="past_closing@example.com", scheduled_time=slot_on(2, 16, 45)
)
# within the slot timings
within = create_test_appointment(
customer_email="within_slot@example.com", scheduled_time=slot_on(2, 10)
)
self.assertTrue(frappe.db.exists("Appointment", within.name))
def test_overlapping_time_slot_capacity(self):
set_booking_setting("number_of_agents", 1)
set_booking_setting("appointment_duration", 30)
slot = slot_on(1, 10)
first = create_test_appointment(customer_email="slot_first@example.com", scheduled_time=slot)
# a booking starting inside the first appointment's duration is rejected
with self.assertRaises(frappe.ValidationError):
create_test_appointment(
customer_email="slot_overlap@example.com",
scheduled_time=slot + datetime.timedelta(minutes=15),
)
# rescheduling must not count the appointment's own booked slot
first.scheduled_time = slot + datetime.timedelta(minutes=10)
first.save()
# a booking starting exactly when the rescheduled one ends is allowed
adjacent = create_test_appointment(
customer_email="slot_adjacent@example.com",
scheduled_time=slot + datetime.timedelta(minutes=40),
)
self.assertTrue(frappe.db.exists("Appointment", adjacent.name))
# a closed (cancelled) appointment frees its slot
first.status = "Closed"
first.save()
after_cancellation = create_test_appointment(
customer_email="after_cancellation@example.com", scheduled_time=slot
)
self.assertTrue(frappe.db.exists("Appointment", after_cancellation.name))

View File

@@ -1,48 +1,56 @@
{
"actions": [],
"allow_bulk_edit": 1,
"creation": "2019-08-27 10:56:48.309824",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"enable_scheduling",
"agent_detail_section",
"availability_of_slots",
"number_of_agents",
"agent_list",
"holiday_list",
"appointment_details_section",
"appointment_duration",
"email_reminders",
"column_break_ehiq",
"agent_list",
"number_of_agents",
"agent_detail_section",
"enable_scheduling",
"availability_of_slots",
"section_break_bkln",
"column_break_alwa",
"advance_booking_days",
"column_break_bspp",
"holiday_list",
"success_details",
"success_redirect_url"
"enable_appointment_portal",
"verification_link_expiry_duration",
"column_break_fovk",
"success_redirect_url",
"action_for_expired_unverified_appointments"
],
"fields": [
{
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "availability_of_slots",
"fieldtype": "Table",
"label": "Availability Of Slots",
"options": "Appointment Booking Slots",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Appointment Booking Slots"
},
{
"default": "1",
"fieldname": "number_of_agents",
"fieldtype": "Int",
"hidden": 1,
"in_list_view": 1,
"label": "Number of Concurrent Appointments",
"read_only": 1,
"reqd": 1
"read_only": 1
},
{
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "holiday_list",
"fieldtype": "Link",
"in_list_view": 1,
"label": "Holiday List",
"options": "Holiday List",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Holiday List"
},
{
"default": "60",
@@ -60,29 +68,31 @@
},
{
"default": "7",
"depends_on": "eval:doc.enable_scheduling === 1;",
"fieldname": "advance_booking_days",
"fieldtype": "Int",
"label": "Number of days appointments can be booked in advance",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;"
},
{
"fieldname": "agent_list",
"fieldtype": "Table MultiSelect",
"label": "Agents",
"options": "Assignment Rule User",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_scheduling === 1;",
"options": "Assignment Rule User"
},
{
"default": "0",
"fieldname": "enable_scheduling",
"fieldtype": "Check",
"label": "Enable Appointment Scheduling",
"reqd": 1
"mandatory_depends_on": "eval:doc.enable_appointment_portal === 1;"
},
{
"fieldname": "agent_detail_section",
"fieldtype": "Section Break",
"label": "Agent Details"
"hide_border": 1,
"label": "Appointment Scheduling"
},
{
"fieldname": "appointment_details_section",
@@ -92,18 +102,68 @@
{
"fieldname": "success_details",
"fieldtype": "Section Break",
"label": "Success Settings"
"label": "Appointment Booking Portal Settings"
},
{
"description": "Leave blank for home.\nThis is relative to site URL, for example \"about\" will redirect to \"https://yoursitename.com/about\"",
"fieldname": "success_redirect_url",
"fieldtype": "Data",
"label": "Success Redirect URL"
"label": "Success Redirect URL",
"permlevel": 1
},
{
"default": "30",
"depends_on": "eval: doc.enable_scheduling === 1;",
"description": "In Minutes (min: 15 mins, max: 60 mins)",
"fieldname": "verification_link_expiry_duration",
"fieldtype": "Int",
"label": "Verification Link Expiry Duration",
"mandatory_depends_on": "eval:doc.enable_appointment_portal === 1;",
"max_value": 60.0,
"min_value": 15.0,
"non_negative": 1,
"permlevel": 1
},
{
"fieldname": "column_break_ehiq",
"fieldtype": "Column Break"
},
{
"default": "0",
"fieldname": "enable_appointment_portal",
"fieldtype": "Check",
"label": "Enable Appointment Booking Through Portal",
"permlevel": 1
},
{
"fieldname": "column_break_fovk",
"fieldtype": "Column Break"
},
{
"default": "Mark as Closed",
"fieldname": "action_for_expired_unverified_appointments",
"fieldtype": "Select",
"label": "Action for Expired Unverified Appointments",
"options": "Mark as Closed\nDelete Permanently",
"permlevel": 1
},
{
"fieldname": "section_break_bkln",
"fieldtype": "Section Break"
},
{
"fieldname": "column_break_alwa",
"fieldtype": "Column Break"
},
{
"fieldname": "column_break_bspp",
"fieldtype": "Column Break"
}
],
"grid_page_length": 50,
"issingle": 1,
"links": [],
"modified": "2022-12-15 11:10:13.517742",
"modified": "2026-07-20 00:11:18.996384",
"modified_by": "Administrator",
"module": "CRM",
"name": "Appointment Booking Settings",
@@ -137,6 +197,15 @@
"role": "Sales Manager",
"share": 1,
"write": 1
},
{
"email": 1,
"permlevel": 1,
"print": 1,
"read": 1,
"role": "System Manager",
"share": 1,
"write": 1
}
],
"quick_entry": 1,
@@ -144,4 +213,4 @@
"sort_order": "DESC",
"states": [],
"track_changes": 1
}
}

View File

@@ -3,11 +3,11 @@
import datetime
import typing
import frappe
from frappe import _
from frappe.model.document import Document
from frappe.utils import getdate
class AppointmentBookingSettings(Document):
@@ -26,33 +26,43 @@ class AppointmentBookingSettings(Document):
AppointmentBookingSlots,
)
action_for_expired_unverified_appointments: DF.Literal["Mark as Closed", "Delete Permanently"]
advance_booking_days: DF.Int
agent_list: DF.TableMultiSelect[AssignmentRuleUser]
appointment_duration: DF.Int
availability_of_slots: DF.Table[AppointmentBookingSlots]
email_reminders: DF.Check
enable_appointment_portal: DF.Check
enable_scheduling: DF.Check
holiday_list: DF.Link
holiday_list: DF.Link | None
number_of_agents: DF.Int
success_redirect_url: DF.Data | None
verification_link_expiry_duration: DF.Int
# end: auto-generated types
agent_list: typing.ClassVar[list] = [] # Hack
min_date = "01/01/1970 "
format_string = "%d/%m/%Y %H:%M:%S"
def validate(self):
self.validate_availability_of_slots()
def save(self):
self.number_of_agents = len(self.agent_list)
super().save()
self.validate_appointment_scheduling()
self.validate_portal_booking()
def validate_appointment_scheduling(self):
if not self.enable_scheduling:
return
self.validate_availability_of_slots()
self.validate_holiday_list()
self.validate_advance_booking_days()
def validate_availability_of_slots(self):
if not self.availability_of_slots:
frappe.throw(
_("Please fill up the Availability of Slots table to enable Appointment Scheduling.")
)
format_string = "%Y-%m-%d %H:%M:%S"
for record in self.availability_of_slots:
from_time = datetime.datetime.strptime(self.min_date + record.from_time, self.format_string)
to_time = datetime.datetime.strptime(self.min_date + record.to_time, self.format_string)
to_time - from_time
from_time = datetime.datetime.strptime(f"1970-01-01 {record.from_time}", format_string)
to_time = datetime.datetime.strptime(f"1970-01-01 {record.to_time}", format_string)
self.validate_from_and_to_time(from_time, to_time, record)
self.duration_is_divisible(from_time, to_time)
@@ -67,3 +77,38 @@ class AppointmentBookingSettings(Document):
timedelta = to_time - from_time
if timedelta.total_seconds() % (self.appointment_duration * 60):
frappe.throw(_("The difference between from time and To Time must be a multiple of Appointment"))
def validate_holiday_list(self):
if not self.holiday_list:
frappe.throw(_("Please select a Holiday List to enable Appointment Scheduling."))
hl_from_date, hl_to_date = frappe.get_cached_value(
"Holiday List", self.holiday_list, ["from_date", "to_date"]
)
now = getdate()
if not (now >= hl_from_date and now <= hl_to_date):
frappe.throw(_("Holiday List - {0} is not valid for current date.").format(self.holiday_list))
def validate_advance_booking_days(self):
if not self.advance_booking_days:
frappe.throw(_("Advance Booking Days is mandatory for Appointment Scheduling."))
def validate_portal_booking(self):
if not self.enable_appointment_portal:
return
if not self.enable_scheduling:
frappe.throw(
_("Appointment Scheduling needs to be enabled for Appointment Booking through portal.")
)
self.validate_link_expiry_duration()
def validate_link_expiry_duration(self):
if (
not self.verification_link_expiry_duration
or self.verification_link_expiry_duration > 60
or self.verification_link_expiry_duration < 15
):
frappe.throw(_("'Verification Link Expiry Duration' must be between 15 to 60 minutes."))

View File

@@ -1,9 +1,125 @@
# Copyright (c) 2019, Frappe Technologies Pvt. Ltd. and Contributors
# See license.txt
# import frappe
import unittest
import datetime
import frappe
from frappe.tests.utils import FrappeTestCase
from frappe.utils import add_to_date, getdate
from erpnext.setup.doctype.holiday_list.test_holiday_list import make_holiday_list
class TestAppointmentBookingSettings(unittest.TestCase):
pass
class TestAppointmentBookingSettings(FrappeTestCase):
def assert_invalid(self, settings):
with self.assertRaises(frappe.ValidationError):
settings.save()
def make_settings(self, appointment_duration=30):
doc = frappe.new_doc("Appointment Booking Settings")
doc.appointment_duration = appointment_duration
return doc
def dt(self, hms):
# the controller parses times against a fixed epoch date
return datetime.datetime.strptime("1970-01-01 " + hms, "%Y-%m-%d %H:%M:%S")
def get_valid_scheduling_settings(self):
holiday_list = make_holiday_list(
"_Test Booking Settings Holiday List",
from_date=getdate(),
to_date=add_to_date(getdate(), days=30),
holiday_dates=[],
)
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 1
settings.appointment_duration = 30
settings.advance_booking_days = 7
settings.verification_link_expiry_duration = 30
settings.holiday_list = holiday_list.name
settings.set("agent_list", [])
settings.append("agent_list", {"user": "Administrator"})
settings.set("availability_of_slots", [])
settings.append(
"availability_of_slots",
{"day_of_week": "Monday", "from_time": "09:00:00", "to_time": "17:00:00"},
)
return settings
def test_from_time_must_precede_to_time(self):
doc = self.make_settings()
record = frappe._dict(day_of_week="Monday")
self.assertRaises(
frappe.ValidationError,
doc.validate_from_and_to_time,
self.dt("18:00:00"),
self.dt("09:00:00"),
record,
)
doc.validate_from_and_to_time(self.dt("09:00:00"), self.dt("18:00:00"), record) # valid order
def test_slot_length_must_be_a_multiple_of_the_duration(self):
doc = self.make_settings(appointment_duration=30)
# 60 minutes is two 30-minute appointments -> fine
doc.duration_is_divisible(self.dt("09:00:00"), self.dt("10:00:00"))
# 45 minutes leaves a partial appointment -> rejected
self.assertRaises(
frappe.ValidationError, doc.duration_is_divisible, self.dt("09:00:00"), self.dt("09:45:00")
)
def test_scheduling_requires_slots(self):
settings = self.get_valid_scheduling_settings()
settings.set("availability_of_slots", [])
self.assert_invalid(settings)
def test_validate_checks_every_slot(self):
settings = self.get_valid_scheduling_settings()
settings.append(
"availability_of_slots",
{"day_of_week": "Tuesday", "from_time": "09:00:00", "to_time": "09:45:00"},
)
self.assert_invalid(settings)
def test_scheduling_requires_holiday_list_covering_today(self):
settings = self.get_valid_scheduling_settings()
settings.holiday_list = None
self.assert_invalid(settings)
expired_list = make_holiday_list(
"_Test Booking Settings Expired Holiday List",
from_date=add_to_date(getdate(), days=-60),
to_date=add_to_date(getdate(), days=-30),
holiday_dates=[],
)
settings.holiday_list = expired_list.name
self.assert_invalid(settings)
def test_scheduling_requires_advance_booking_days(self):
settings = self.get_valid_scheduling_settings()
settings.advance_booking_days = 0
self.assert_invalid(settings)
def test_portal_requires_scheduling(self):
settings = frappe.get_doc("Appointment Booking Settings")
settings.enable_scheduling = 0
settings.enable_appointment_portal = 1
self.assert_invalid(settings)
def test_portal_expiry_duration_bounds(self):
settings = self.get_valid_scheduling_settings()
settings.enable_appointment_portal = 1
settings.verification_link_expiry_duration = 5
self.assert_invalid(settings)
def test_number_of_agents_derived_from_agent_list(self):
settings = self.get_valid_scheduling_settings()
settings.number_of_agents = 99
settings.save()
self.assertEqual(frappe.db.get_single_value("Appointment Booking Settings", "number_of_agents"), 1)

View File

@@ -431,6 +431,7 @@ scheduler_events = {
],
"hourly_long": [],
"hourly_maintenance": [
"erpnext.crm.doctype.appointment.appointment.handle_expired_unverified_appointments",
"erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.repost_entries",
"erpnext.utilities.bulk_transaction.retry",
"erpnext.projects.doctype.project.project.collect_project_status",

View File

@@ -365,8 +365,6 @@ frappe.ui.form.on("Production Plan", {
frappe.throw(__("Select the Warehouse"));
}
frm.set_value("consider_minimum_order_qty", 0);
if (frm.doc.ignore_existing_ordered_qty) {
frm.events.get_items_for_material_requests(frm);
} else {

View File

@@ -1708,7 +1708,13 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d
if (not ignore_existing_ordered_qty or get_parent_warehouse_data) and warehouses:
new_mr_items = []
for item in mr_items:
get_materials_from_other_locations(item, warehouses, new_mr_items, company)
get_materials_from_other_locations(
item,
warehouses,
new_mr_items,
company,
consider_minimum_order_qty=doc.get("consider_minimum_order_qty"),
)
mr_items = new_mr_items
@@ -1728,7 +1734,9 @@ def get_items_for_material_requests(doc, warehouses=None, get_parent_warehouse_d
return mr_items
def get_materials_from_other_locations(item, warehouses, new_mr_items, company):
def get_materials_from_other_locations(
item, warehouses, new_mr_items, company, consider_minimum_order_qty=False
):
from erpnext.stock.doctype.pick_list.pick_list import get_available_item_locations
stock_uom, purchase_uom = frappe.db.get_value(
@@ -1773,7 +1781,8 @@ def get_materials_from_other_locations(item, warehouses, new_mr_items, company):
precision = frappe.get_precision("Material Request Plan Item", "quantity")
if flt(required_qty, precision) > 0:
required_qty = required_qty
if consider_minimum_order_qty:
required_qty = max(required_qty, flt(item.get("min_order_qty")))
if frappe.db.get_value("UOM", purchase_uom, "must_be_whole_number"):
required_qty = ceil(required_qty)

View File

@@ -1832,6 +1832,17 @@ class TestProductionPlan(FrappeTestCase):
for d in mr_items:
self.assertEqual(d.get("quantity"), 1000.0)
source_warehouse = create_warehouse("MOQ Source Warehouse", company="_Test Company")
make_stock_entry(item_code=rm_item, qty=7, rate=100, target=source_warehouse)
mr_items = get_items_for_material_requests(
pln.as_dict(), warehouses=[{"warehouse": source_warehouse}]
)
self.assertEqual(len(mr_items), 2)
items_by_type = {d.get("material_request_type"): d for d in mr_items}
self.assertEqual(items_by_type["Material Transfer"].get("quantity"), 7.0)
self.assertEqual(items_by_type["Purchase"].get("quantity"), 1000.0)
def test_fg_item_quantity(self):
fg_item = make_item(properties={"is_stock_item": 1}).name
rm_item = make_item(properties={"is_stock_item": 1}).name

View File

@@ -82,6 +82,7 @@
"bold": 0,
"collapsible": 0,
"columns": 0,
"fetch_from": "material_request.transaction_date",
"fieldname": "material_request_date",
"fieldtype": "Date",
"hidden": 0,
@@ -122,7 +123,7 @@
"issingle": 0,
"istable": 1,
"max_attachments": 0,
"modified": "2017-10-29 12:31:57.986869",
"modified": "2026-07-17 22:06:35.428875",
"modified_by": "Administrator",
"module": "Manufacturing",
"name": "Production Plan Material Request",

View File

@@ -442,3 +442,4 @@ erpnext.patches.v15_0.backfill_sla_link_filters_on_custom_field
erpnext.patches.v15_0.backfill_sla_link_filters_on_docfield
erpnext.patches.v16_0.crm_settings_handle_allowed_users_for_frappe_crm
erpnext.patches.v16_0.backfill_pick_list_transferred_qty
erpnext.patches.v16_0.access_control_for_project_users

View File

@@ -0,0 +1,33 @@
import frappe
def execute():
Project = frappe.qb.DocType("Project")
ProjectUser = frappe.qb.DocType("Project User")
query = (
frappe.qb.from_(Project)
.join(ProjectUser)
.on(Project.name == ProjectUser.parent)
.select(Project.name, ProjectUser.user)
)
proj_users = query.run(as_dict=1)
project_mapped_users = get_project_mapped_users(proj_users)
for d in proj_users:
if d.user in project_mapped_users[d.name]:
continue
frappe.share.add_docshare("Project", d.name, user=d.user)
def get_project_mapped_users(proj_users):
projects = set([d.name for d in proj_users])
project_mapped_users = {}
for d in projects:
project_mapped_users[d] = [d.user for d in frappe.share.get_users("Project", d)]
return project_mapped_users

View File

@@ -83,7 +83,7 @@
"no_copy": 1,
"oldfieldname": "status",
"oldfieldtype": "Select",
"options": "Open\nCompleted\nCancelled",
"options": "Open\nOn hold\nCompleted\nCancelled",
"search_index": 1
},
{
@@ -117,7 +117,7 @@
"fieldtype": "Percent",
"label": "% Completed",
"no_copy": 1,
"read_only": 1
"read_only_depends_on": "eval:doc.percent_complete_method != 'Manual'"
},
{
"fieldname": "column_break_5",
@@ -205,13 +205,15 @@
"fieldname": "users",
"fieldtype": "Table",
"label": "Users",
"options": "Project User"
"options": "Project User",
"permlevel": 1
},
{
"fieldname": "copied_from",
"fieldtype": "Data",
"hidden": 1,
"label": "Copied From",
"permlevel": 1,
"read_only": 1
},
{
@@ -464,13 +466,25 @@
"index_web_pages_for_search": 1,
"links": [],
"max_attachments": 4,
"modified": "2026-05-22 16:45:50.762759",
"modified": "2026-07-21 11:23:22.000000",
"modified_by": "Administrator",
"module": "Projects",
"name": "Project",
"naming_rule": "By \"Naming Series\" field",
"owner": "Administrator",
"permissions": [
{
"delete": 1,
"email": 1,
"export": 1,
"permlevel": 1,
"print": 1,
"read": 1,
"report": 1,
"role": "Projects Manager",
"share": 1,
"write": 1
},
{
"create": 1,
"delete": 1,

View File

@@ -61,7 +61,7 @@ class Project(Document):
project_type: DF.Link | None
sales_order: DF.Link | None
second_email: DF.Time | None
status: DF.Literal["Open", "Completed", "Cancelled"]
status: DF.Literal["Open", "On hold", "Completed", "Cancelled"]
subject: DF.Data | None
to_time: DF.Time | None
total_billable_amount: DF.Currency
@@ -93,6 +93,7 @@ class Project(Document):
def validate(self):
if not self.is_new():
self.copy_from_template() # nosemgrep
self.control_access_for_project_users()
self.send_welcome_email()
self.update_costing()
self.update_percent_complete()
@@ -207,6 +208,7 @@ class Project(Document):
self.copy_from_template() # nosemgrep
if self.sales_order:
frappe.db.set_value("Sales Order", self.sales_order, "project", self.name)
self.control_access_for_project_users()
def on_trash(self):
frappe.db.set_value("Sales Order", {"project": self.name}, "project", "")
@@ -222,6 +224,8 @@ class Project(Document):
if self.percent_complete_method == "Manual":
if self.status == "Completed":
self.percent_complete = 100
elif flt(self.percent_complete) < 0 or flt(self.percent_complete) > 100:
frappe.throw(_("% Complete must be between 0 and 100"))
return
total = frappe.db.count("Task", dict(project=self.name))
@@ -264,8 +268,8 @@ class Project(Document):
pct_complete += row["progress"] * frappe.utils.safe_div(row["task_weight"], weight_sum)
self.percent_complete = flt(flt(pct_complete), 2)
# don't update status if it is cancelled
if self.status == "Cancelled":
# don't update status if it is manually set to cancelled or on hold
if self.status in ("Cancelled", "On hold"):
return
self.status = "Completed" if self.percent_complete == 100 else "Open"
@@ -377,6 +381,34 @@ class Project(Document):
)
user.welcome_email_sent = 1
def control_access_for_project_users(self):
def revoke_access_for_project_users(removed_users):
users = set([d.user for d in frappe.share.get_users(self.doctype, self.name)])
for user in removed_users:
if user not in users:
continue
frappe.share.remove(self.doctype, self.name, user)
def grant_access_for_project_users(new_users):
for user in new_users:
frappe.share.add_docshare(self.doctype, self.name, user=user)
current_users = set([d.user for d in self.users])
old_doc = self.get_doc_before_save()
if not old_doc:
grant_access_for_project_users(current_users)
return
previous_users = set([d.user for d in old_doc.users])
new_users = current_users - previous_users
removed_users = previous_users - current_users
revoke_access_for_project_users(removed_users)
grant_access_for_project_users(new_users)
def get_timeline_data(doctype: str, name: str) -> dict[int, int]:
"""Return timeline for attendance"""

View File

@@ -4,6 +4,8 @@ frappe.listview_settings["Project"] = {
get_indicator: function (doc) {
if (doc.status == "Open" && doc.percent_complete) {
return [__("{0}%", [cint(doc.percent_complete)]), "orange", "percent_complete,>,0|status,=,Open"];
} else if (doc.status == "On hold") {
return [__("On hold"), "blue", "status,=,On hold"];
} else {
return [__(doc.status), frappe.utils.guess_colour(doc.status), "status,=," + doc.status];
}

View File

@@ -227,6 +227,116 @@ class TestProject(FrappeTestCase):
project.save()
self.assertEqual(project.status, "Completed")
def _project_with_tasks(self, method, count):
name = f"_Test PercentComplete {frappe.generate_hash(length=8)}"
project = frappe.get_doc(
{
"doctype": "Project",
"project_name": name,
"status": "Open",
"percent_complete_method": method,
"company": "_Test Company",
"expected_start_date": nowdate(),
}
).insert()
task_names = []
for i in range(count):
task = frappe.get_doc(
{
"doctype": "Task",
"subject": f"{name} Task {i}",
"project": project.name,
"status": "Open",
"exp_start_date": nowdate(),
"exp_end_date": nowdate(),
}
).insert()
task_names.append(task.name)
return project, task_names
def test_percent_complete_manual(self):
project, tasks = self._project_with_tasks("Manual", 2)
# manual value is preserved on save, even with linked tasks
project.percent_complete = 42
project.save()
self.assertEqual(project.percent_complete, 42)
# task updates do not overwrite the manual value
frappe.db.set_value("Task", tasks[0], "status", "Completed")
project.update_percent_complete()
self.assertEqual(project.percent_complete, 42)
# out-of-range values are rejected
project.percent_complete = 150
self.assertRaises(frappe.ValidationError, project.save)
project.reload()
project.percent_complete = -10
self.assertRaises(frappe.ValidationError, project.save)
project.reload()
# Completed status forces 100 regardless of the manual value
project.percent_complete = 42
project.status = "Completed"
project.save()
self.assertEqual(project.percent_complete, 100)
def _create_portal_user(self, email):
"""A user with no Project-related role, so read access can only come from
control_access_for_project_users() sharing the doc with them."""
if not frappe.db.exists("User", email):
frappe.get_doc(
{
"doctype": "User",
"email": email,
"first_name": "Portal",
"send_welcome_email": 0,
}
).insert(ignore_permissions=True)
return email
def test_new_project_grants_access_to_its_users(self):
member = self._create_portal_user(f"new_proj_member_{frappe.generate_hash(length=6)}@example.com")
project = frappe.get_doc(
doctype="Project",
project_name=f"_Test New Project Access {frappe.generate_hash(length=6)}",
status="Open",
company="_Test Company",
)
project.append("users", {"user": member, "welcome_email_sent": 1})
project.insert() # must not raise
self.assertTrue(project.has_permission(user=member))
shared_with = [d.user for d in frappe.share.get_users("Project", project.name)]
self.assertIn(member, shared_with)
def test_adding_and_removing_project_user_updates_access(self):
stays = self._create_portal_user(f"stays_{frappe.generate_hash(length=6)}@example.com")
leaves = self._create_portal_user(f"leaves_{frappe.generate_hash(length=6)}@example.com")
project = frappe.get_doc(
doctype="Project",
project_name=f"_Test Project User Membership {frappe.generate_hash(length=6)}",
status="Open",
company="_Test Company",
)
project.append("users", {"user": stays, "welcome_email_sent": 1})
project.insert()
self.assertTrue(project.has_permission(user=stays))
# adding a user on update (not insert) must also grant them access
project.append("users", {"user": leaves, "welcome_email_sent": 1})
project.save()
self.assertTrue(project.has_permission(user=leaves))
# removing a user must revoke the share that was granted for membership
project.users = [d for d in project.users if d.user != leaves]
project.save()
self.assertFalse(project.has_permission(user=leaves))
self.assertTrue(project.has_permission(user=stays))
def get_project(name, template):
project = frappe.get_doc(

View File

@@ -22,6 +22,7 @@ erpnext.accounts.dimensions = {
});
me.default_dimensions = r.message[1];
me.setup_filters(frm, doctype);
me.update_dimension(frm, doctype);
},
});
},

View File

@@ -23,7 +23,10 @@ from erpnext.accounts.party import (
validate_party_accounts,
validate_party_currency_before_merging,
)
from erpnext.controllers.website_list_for_contact import add_role_for_portal_user
from erpnext.controllers.website_list_for_contact import (
add_role_for_portal_user,
link_portal_users_to_contacts,
)
from erpnext.utilities.transaction_base import TransactionBase
@@ -243,6 +246,8 @@ class Customer(TransactionBase):
self.update_customer_groups()
link_portal_users_to_contacts(self)
def add_role_for_user(self):
for portal_user in self.portal_users:
add_role_for_portal_user(portal_user, "Customer")

View File

@@ -29,7 +29,8 @@ class TestCustomer(FrappeTestCase):
make_test_records("Item")
def tearDown(self):
set_credit_limit("_Test Customer", "_Test Company", 0)
if frappe.db.exists("Customer", "_Test Customer"):
set_credit_limit("_Test Customer", "_Test Company", 0)
def test_get_customer_group_details(self):
doc = frappe.new_doc("Customer Group")
@@ -386,6 +387,34 @@ class TestCustomer(FrappeTestCase):
self.assertEqual(middle, "Michael")
self.assertEqual(last, "Doe")
def test_portal_user_contact_link(self):
user_email = frappe.generate_hash() + "@example.com"
user = frappe.new_doc("User")
user.email = user_email
user.first_name = "Test Portal Customer User"
user.send_welcome_email = False
user.insert(ignore_permissions=True)
contact = frappe.new_doc("Contact")
contact.first_name = "Test Portal Customer User"
contact.add_email(user_email, is_primary=1)
contact.links = []
contact.insert(ignore_permissions=True)
customer = frappe.get_doc(
{
"doctype": "Customer",
"customer_name": "Test Portal Contact Customer",
"customer_type": "Individual",
"customer_group": "_Test Customer Group",
}
)
customer.append("portal_users", {"user": user.name})
customer.insert()
contact.reload()
self.assertTrue(contact.has_link("Customer", customer.name))
def get_customer_dict(customer_name):
return {

View File

@@ -589,6 +589,11 @@ erpnext.selling.SalesOrderController = class SalesOrderController extends erpnex
super.onload(doc, dt, dn);
}
company() {
super.company();
erpnext.accounts.dimensions.update_dimension(this.frm, this.frm.doctype);
}
refresh(doc, dt, dn) {
var me = this;
super.refresh();

View File

@@ -4,7 +4,9 @@
import frappe
from frappe import _, msgprint, qb
from frappe.query_builder import Criterion
from frappe.desk.reportview import build_match_conditions
from frappe.query_builder import Case, Criterion
from pypika.terms import LiteralValue
from erpnext import get_company_currency
@@ -155,84 +157,94 @@ def get_columns(filters):
def get_entries(filters):
date_field = filters["doc_type"] == "Sales Order" and "transaction_date" or "posting_date"
if filters["doc_type"] == "Sales Order":
qty_field = "delivered_qty"
else:
qty_field = "qty"
conditions, values = get_conditions(filters, date_field)
doc_type = filters["doc_type"]
entries = frappe.db.sql(
"""
SELECT
dt.name, dt.customer, dt.territory, dt.{} as posting_date, dt_item.item_code,
st.sales_person, st.allocated_percentage, dt_item.warehouse,
CASE
WHEN dt.status = "Closed" THEN dt_item.{} * dt_item.conversion_factor
ELSE dt_item.stock_qty
END as stock_qty,
CASE
WHEN dt.status = "Closed" THEN (dt_item.base_net_rate * dt_item.{} * dt_item.conversion_factor)
ELSE dt_item.base_net_amount
END as base_net_amount,
CASE
WHEN dt.status = "Closed" THEN ((dt_item.base_net_rate * dt_item.{} * dt_item.conversion_factor) * st.allocated_percentage/100)
ELSE dt_item.base_net_amount * st.allocated_percentage/100
END as contribution_amt
FROM
`tab{}` dt, `tab{} Item` dt_item, `tabSales Team` st
WHERE
st.parent = dt.name and dt.name = dt_item.parent and st.parenttype = {}
and dt.docstatus = 1 {} order by st.sales_person, dt.name desc
""".format(
date_field,
qty_field,
qty_field,
qty_field,
filters["doc_type"],
filters["doc_type"],
"%s",
conditions,
),
tuple([filters["doc_type"], *values]),
as_dict=1,
date_field = "transaction_date" if doc_type == "Sales Order" else "posting_date"
qty_field = "delivered_qty" if doc_type == "Sales Order" else "qty"
dt = frappe.qb.DocType(doc_type)
dt_item = frappe.qb.DocType(f"{doc_type} Item")
st = frappe.qb.DocType("Sales Team")
calc_qty = dt_item[qty_field] * dt_item.conversion_factor
calc_net_amount = dt_item.base_net_rate * calc_qty
stock_qty_case = Case().when(dt.status == "Closed", calc_qty).else_(dt_item.stock_qty).as_("stock_qty")
base_net_amount_case = (
Case()
.when(dt.status == "Closed", calc_net_amount)
.else_(dt_item.base_net_amount)
.as_("base_net_amount")
)
return entries
def get_conditions(filters, date_field):
conditions = [""]
values = []
contribution_amt_case = (
Case()
.when(dt.status == "Closed", (calc_net_amount * st.allocated_percentage / 100))
.else_(dt_item.base_net_amount * st.allocated_percentage / 100)
.as_("contribution_amt")
)
# Only pass valid document-field filters to get_query; report-specific keys such as
# doc_type / sales_person / item_group are handled separately below.
doc_filters = {"docstatus": 1}
for field in ["company", "customer", "territory"]:
if filters.get(field):
conditions.append(f"dt.{field}=%s")
values.append(filters[field])
doc_filters[field] = filters.get(field)
if filters.get("from_date") and filters.get("to_date"):
doc_filters[date_field] = ["between", [filters.get("from_date"), filters.get("to_date")]]
elif filters.get("from_date"):
doc_filters[date_field] = [">=", filters.get("from_date")]
elif filters.get("to_date"):
doc_filters[date_field] = ["<=", filters.get("to_date")]
query = (
frappe.get_query(dt, filters=doc_filters)
.join(dt_item)
.on(dt.name == dt_item.parent)
.join(st)
.on(dt.name == st.parent)
.select(
dt.name,
dt.customer,
dt.territory,
dt[date_field].as_("posting_date"),
dt_item.item_code,
st.sales_person,
st.allocated_percentage,
dt_item.warehouse,
stock_qty_case,
base_net_amount_case,
contribution_amt_case,
)
.where(st.parenttype == doc_type)
)
if filters.get("sales_person"):
lft, rgt = frappe.get_value("Sales Person", filters.get("sales_person"), ["lft", "rgt"])
conditions.append(
f"exists(select name from `tabSales Person` where lft >= {lft} and rgt <= {rgt} and name=st.sales_person)"
lft, rgt = frappe.db.get_value("Sales Person", filters.get("sales_person"), ["lft", "rgt"])
sp = frappe.qb.DocType("Sales Person")
query = query.where(
st.sales_person.isin(frappe.qb.from_(sp).select(sp.name).where((sp.lft >= lft) & (sp.rgt <= rgt)))
)
if filters.get("from_date"):
conditions.append(f"dt.{date_field}>=%s")
values.append(filters["from_date"])
# only resolve items when an item_group/brand filter is set; otherwise get_items
# would return every item in the system and add a huge IN() clause on each run
if filters.get("item_group") or filters.get("brand"):
items = get_items(filters)
if not items:
# the item_group/brand filter matched nothing -> no rows
return []
query = query.where(dt_item.item_code.isin([d[0] for d in items]))
if filters.get("to_date"):
conditions.append(f"dt.{date_field}<=%s")
values.append(filters["to_date"])
query = query.orderby(st.sales_person).orderby(dt.name, order=frappe.qb.desc)
items = get_items(filters)
if items:
conditions.append("dt_item.item_code in (%s)" % ", ".join(["%s"] * len(items)))
values += items
else:
# return empty result, if no items are fetched after filtering on 'item group' and 'brand'
conditions.append("dt_item.item_code = Null")
# Apply user permissions (v15: ignore_permissions is not available)
match_conditions = build_match_conditions(doc_type)
if match_conditions:
query = query.where(LiteralValue(match_conditions))
return " and ".join(conditions), values
return query.run(as_dict=True)
def get_items(filters):
@@ -259,8 +271,5 @@ def get_items(filters):
def get_item_details():
item_details = {}
for d in frappe.db.sql("""SELECT `name`, `item_group`, `brand` FROM `tabItem`""", as_dict=1):
item_details.setdefault(d.name, d)
return item_details
items = frappe.get_all("Item", fields=["name", "item_group", "brand"])
return {d.name: d for d in items}

View File

@@ -53,6 +53,11 @@ def get_data(
"reserved_qty": ["!=", 0],
"reserved_qty_for_production": ["!=", 0],
"reserved_qty_for_sub_contract": ["!=", 0],
"reserved_qty_for_production_plan": ["!=", 0],
"reserved_stock": ["!=", 0],
"ordered_qty": ["!=", 0],
"indented_qty": ["!=", 0],
"planned_qty": ["!=", 0],
"actual_qty": ["!=", 0],
},
filters=filters,

View File

@@ -92,7 +92,10 @@ frappe.ui.form.on("Material Request", {
erpnext.accounts.dimensions.setup_dimension_filters(frm, frm.doctype);
if (!frm.doc.buying_price_list) {
frm.doc.buying_price_list = frappe.defaults.get_default("buying_price_list");
const buying_price_list = frappe.defaults.get_default("buying_price_list");
if (frappe.has_permission("Price List", "read", buying_price_list)) {
frm.set_value("buying_price_list", buying_price_list);
}
}
},
@@ -271,9 +274,7 @@ frappe.ui.form.on("Material Request", {
from_warehouse: item.from_warehouse,
warehouse: item.warehouse,
doctype: frm.doc.doctype,
buying_price_list: frm.doc.buying_price_list
? frm.doc.buying_price_list
: frappe.defaults.get_default("buying_price_list"),
buying_price_list: frm.doc.buying_price_list,
currency: frappe.defaults.get_default("Currency"),
name: frm.doc.name,
qty: item.qty || 1,

View File

@@ -19,6 +19,7 @@ from erpnext.buying.utils import check_on_hold_or_closed_status, validate_for_it
from erpnext.controllers.buying_controller import BuyingController
from erpnext.manufacturing.doctype.work_order.work_order import get_item_details
from erpnext.stock.doctype.item.item import get_item_defaults
from erpnext.stock.get_item_details import get_price_list_rate_for
from erpnext.stock.stock_balance import get_indented_qty, update_bin_qty
form_grid_templates = {"items": "templates/form_grid/material_request_grid.html"}
@@ -169,8 +170,46 @@ class MaterialRequest(BuyingController):
self.reset_default_field_value("set_warehouse", "items", "warehouse")
self.reset_default_field_value("set_from_warehouse", "items", "from_warehouse")
if self.buying_price_list and not frappe.get_value("Price List", self.buying_price_list, "buying"):
self.buying_price_list = None
if not self.buying_price_list:
self.buying_price_list = frappe.defaults.get_defaults().buying_price_list
buying_price_list = frappe.defaults.get_defaults().buying_price_list
if frappe.has_permission("Price List", "read", buying_price_list):
self.buying_price_list = buying_price_list
def on_update(self):
if not self.is_new() and self.buying_price_list and self.has_value_changed("buying_price_list"):
self.update_item_rates()
def update_item_rates(self):
price_not_uom_dependent = frappe.get_value(
"Price List", self.buying_price_list, "price_not_uom_dependent"
)
for item in self.items:
rate = get_price_list_rate_for(
frappe._dict(
{
"price_list": self.buying_price_list,
"uom": item.uom,
"transaction_date": self.transaction_date,
"qty": item.qty,
"stock_uom": item.stock_uom,
"conversion_factor": item.conversion_factor,
"price_list_uom_dependant": price_not_uom_dependent,
}
),
item.item_code,
)
if rate is not None:
item.db_set({"rate": rate, "amount": flt(rate * item.qty, item.precision("amount"))})
frappe.msgprint(
_("Item rates have been updated based on the selected Buying Price List {0}").format(
self.buying_price_list
),
alert=True,
)
def before_update_after_submit(self):
self.validate_schedule_date()

View File

@@ -2648,7 +2648,7 @@ class StockEntry(StockController):
self.process_loss_qty = flt(
(flt(self.fg_completed_qty) * flt(self.process_loss_percentage)) / 100
)
elif self.process_loss_qty and not self.process_loss_percentage:
elif self.process_loss_qty and self.fg_completed_qty:
self.process_loss_percentage = flt(
(flt(self.process_loss_qty) / flt(self.fg_completed_qty)) * 100
)

View File

@@ -2729,6 +2729,28 @@ class TestStockEntry(FrappeTestCase):
# delete naming rule
frappe.delete_doc("Document Naming Rule", qc_naming_rule.name)
def test_process_loss_percentage_resyncs_from_qty(self):
# changing fg qty recomputes process_loss_qty
se = frappe.new_doc("Stock Entry")
se.purpose = "Manufacture"
se.fg_completed_qty = 200
se.process_loss_qty = 100
se.process_loss_percentage = 80
se.set_process_loss_qty()
self.assertEqual(se.process_loss_percentage, 50)
def test_process_loss_qty_derived_from_percentage_when_qty_blank(self):
se = frappe.new_doc("Stock Entry")
se.purpose = "Manufacture"
se.fg_completed_qty = 200
se.process_loss_percentage = 25
se.set_process_loss_qty()
self.assertEqual(se.process_loss_qty, 50)
def make_serialized_item(**args):
args = frappe._dict(args)

View File

@@ -48,11 +48,19 @@ frappe.pages["stock-balance"].on_page_load = function (wrapper) {
sort_by: "projected_qty",
sort_order: "asc",
options: [
{ fieldname: "projected_qty", label: __("Projected qty") },
{ fieldname: "reserved_qty", label: __("Reserved for sale") },
{ fieldname: "reserved_qty_for_production", label: __("Reserved for manufacturing") },
{ fieldname: "reserved_qty_for_sub_contract", label: __("Reserved for sub contracting") },
{ fieldname: "actual_qty", label: __("Actual qty in stock") },
{ fieldname: "projected_qty", label: __("Projected Qty") },
{ fieldname: "reserved_qty", label: __("Reserved Qty") },
{ fieldname: "reserved_qty_for_production", label: __("Reserved Qty for Production") },
{ fieldname: "reserved_qty_for_sub_contract", label: __("Reserved Qty for Subcontract") },
{
fieldname: "reserved_qty_for_production_plan",
label: __("Reserved Qty for Production Plan"),
},
{ fieldname: "reserved_stock", label: __("Reserved Stock") },
{ fieldname: "ordered_qty", label: __("Ordered Qty") },
{ fieldname: "indented_qty", label: __("Requested Qty") },
{ fieldname: "planned_qty", label: __("Planned Qty") },
{ fieldname: "actual_qty", label: __("Actual Qty") },
],
},
change: function (sort_by, sort_order) {

View File

@@ -306,6 +306,7 @@ class FIFOSlots:
# prepare single sle voucher detail lookup
self.prepare_stock_reco_voucher_wise_count()
self.float_precision = get_float_precision()
if stock_ledger_entries is None:
# streaming path: nested queries invalidate the streaming cursor below,
@@ -370,6 +371,7 @@ class FIFOSlots:
row, fifo_queue, transferred_item_key, serial_nos, batch_nos, from_end
)
self._revalue_stock_reconciliation_slots(row, fifo_queue, batch_nos)
self._update_balances(row, key)
self._trim_serial_fifo_queue(row, key, fifo_queue)
@@ -393,6 +395,36 @@ class FIFOSlots:
# Stock reconciliation stores the final balance; FIFO needs the movement delta.
row.actual_qty = flt(row.qty_after_transaction) - flt(prev_balance_qty)
def _revalue_stock_reconciliation_slots(self, row: dict, fifo_queue: list, batch_nos: list) -> None:
if row.voucher_type != "Stock Reconciliation" or row.has_serial_no:
return
if row.has_batch_no:
if flt(row.actual_qty) > 0:
self._revalue_reconciled_batch_slots(fifo_queue, batch_nos)
return
for slot in fifo_queue:
if is_qty_slot(slot):
slot[FIFO_VALUE_INDEX] = flt(slot[FIFO_QTY_INDEX] * flt(row.valuation_rate))
def _revalue_reconciled_batch_slots(self, fifo_queue: list, batch_nos: list) -> None:
for batch_no, _use_batchwise_valuation, qty, stock_value_difference in batch_nos:
if not flt(qty):
continue
slots = [
slot
for slot in fifo_queue
if is_batch_slot(slot) and slot[BATCH_SLOT_BATCH_INDEX] == batch_no
]
if flt(sum(flt(slot[BATCH_SLOT_QTY_INDEX]) for slot in slots) - flt(qty), self.float_precision):
continue
rate = flt(stock_value_difference) / flt(qty)
for slot in slots:
slot[BATCH_SLOT_VALUE_INDEX] = flt(slot[BATCH_SLOT_QTY_INDEX] * rate)
def _get_serial_and_batch_nos(
self, row: dict, bundle_wise_serial_nos: dict, bundle_wise_batch_nos: dict
) -> tuple[list, list]:

View File

@@ -383,6 +383,243 @@ class TestStockAgeing(FrappeTestCase):
self.assertEqual(queue, [[60.0, "2025-11-30", 60.0], [30.0, "2026-01-31", 30.0]])
self.assertEqual(report_data[0][7:15], [30.0, 30.0, 0.0, 0.0, 60.0, 60.0, 0.0, 0.0])
def test_stock_reco_revaluation_rescales_queue_values(self):
"Ledger (same wh): [+15 @ 100, reco reset >> 20 @ 50]"
sle = [
frappe._dict(
name="Flask Item",
actual_qty=15,
qty_after_transaction=15,
stock_value_difference=1500,
valuation_rate=100,
warehouse="WH 1",
posting_date="2021-12-01",
voucher_type="Stock Entry",
voucher_no="001",
has_serial_no=False,
serial_no=None,
),
frappe._dict(
name="Flask Item",
actual_qty=0,
qty_after_transaction=20,
stock_value_difference=(-500),
valuation_rate=50,
warehouse="WH 1",
posting_date="2021-12-02",
voucher_type="Stock Reconciliation",
voucher_no="002",
has_serial_no=False,
serial_no=None,
),
]
slots = FIFOSlots(self.filters, sle).generate()
queue = slots["Flask Item"]["fifo_queue"]
self.assertEqual(queue, [[15.0, "2021-12-01", 750.0], [5.0, "2021-12-02", 250.0]])
def test_stock_reco_with_split_out_and_in_sles_revalues_queue(self):
"Ledger (same wh): [+10 @ 100, reco out >> 0, reco in >> 12 @ 2]"
sle = [
frappe._dict(
name="Flask Item",
actual_qty=10,
qty_after_transaction=10,
stock_value_difference=1000,
valuation_rate=100,
warehouse="WH 1",
posting_date="2021-12-01",
voucher_type="Stock Entry",
voucher_no="001",
has_serial_no=False,
serial_no=None,
),
frappe._dict(
name="Flask Item",
actual_qty=(-10),
qty_after_transaction=0,
stock_value_difference=(-1000),
valuation_rate=100,
warehouse="WH 1",
posting_date="2021-12-02",
voucher_type="Stock Reconciliation",
voucher_no="002",
has_serial_no=False,
serial_no=None,
),
frappe._dict(
name="Flask Item",
actual_qty=12,
qty_after_transaction=12,
stock_value_difference=24,
valuation_rate=2,
warehouse="WH 1",
posting_date="2021-12-02",
voucher_type="Stock Reconciliation",
voucher_no="002",
has_serial_no=False,
serial_no=None,
),
]
slots = FIFOSlots(self.filters, sle).generate()
queue = slots["Flask Item"]["fifo_queue"]
self.assertEqual(queue, [[10.0, "2021-12-01", 20.0], [2.0, "2021-12-02", 4.0]])
def test_stock_reco_decrease_rescales_slots_at_reco_rate(self):
"""Ledger (same wh): [+10 @ 100, +20 @ 250, reco reset >> 25 @ 220]
The valuation engine collapses the FIFO stack to qty_after * valuation_rate
on a reco, so remaining slot values follow the reco rate, not the lot rates."""
sle = [
frappe._dict(
name="Flask Item",
actual_qty=10,
qty_after_transaction=10,
stock_value_difference=1000,
valuation_rate=100,
warehouse="WH 1",
posting_date="2021-12-01",
voucher_type="Stock Entry",
voucher_no="001",
has_serial_no=False,
serial_no=None,
),
frappe._dict(
name="Flask Item",
actual_qty=20,
qty_after_transaction=30,
stock_value_difference=5000,
valuation_rate=200,
warehouse="WH 1",
posting_date="2021-12-02",
voucher_type="Stock Entry",
voucher_no="002",
has_serial_no=False,
serial_no=None,
),
frappe._dict(
name="Flask Item",
actual_qty=0,
qty_after_transaction=25,
stock_value_difference=(-500),
valuation_rate=220,
warehouse="WH 1",
posting_date="2021-12-03",
voucher_type="Stock Reconciliation",
voucher_no="003",
has_serial_no=False,
serial_no=None,
),
]
slots = FIFOSlots(self.filters, sle).generate()
queue = slots["Flask Item"]["fifo_queue"]
self.assertEqual(queue, [[5.0, "2021-12-01", 1100.0], [20.0, "2021-12-02", 4400.0]])
def test_batch_stock_reco_revaluation_rescales_slot_values(self):
"Ledger (same wh, batch B): [+10 @ 100, reco out >> 0, reco in >> 12 @ 2]"
from erpnext.stock.doctype.item.test_item import make_item
item_code = make_item(
"Test Stock Ageing Batch Reco Revaluation",
{"is_stock_item": 1, "has_batch_no": 1, "valuation_method": "FIFO"},
).name
batch_no = "SA-RECO-REVALUE-BATCH"
if not frappe.db.exists("Batch", batch_no):
frappe.get_doc({"doctype": "Batch", "batch_id": batch_no, "item": item_code}).insert(
ignore_permissions=True
)
frappe.db.set_value("Batch", batch_no, "use_batchwise_valuation", 1)
def make_sle(posting_date, voucher_type, voucher_no, actual_qty, qty_after, stock_value_difference):
return frappe._dict(
name=item_code,
actual_qty=actual_qty,
qty_after_transaction=qty_after,
stock_value_difference=stock_value_difference,
valuation_rate=abs(stock_value_difference / actual_qty) if actual_qty else 0,
warehouse="WH 1",
posting_date=posting_date,
voucher_type=voucher_type,
voucher_no=voucher_no,
has_serial_no=False,
has_batch_no=True,
serial_no=None,
batch_no=batch_no,
)
sle = [
make_sle("2021-12-01", "Stock Entry", "001", 10, 10, 1000),
make_sle("2021-12-02", "Stock Reconciliation", "002", -10, 0, -1000),
make_sle("2021-12-02", "Stock Reconciliation", "002", 12, 12, 24),
]
slots = FIFOSlots(self.filters, sle).generate()
queue = slots[item_code]["fifo_queue"]
self.assertEqual(
queue,
[
[batch_no, 1, 10.0, "2021-12-01", 20.0],
[batch_no, 1, 2.0, "2021-12-02", 4.0],
],
)
def test_partial_batch_reco_keeps_existing_slot_values(self):
"""Ledger (same wh, batch B): [+10 @ 100, single-SLE reco >> 12]
The reco entry qty (delta 2) does not cover the whole batch, so
stock_value_difference / qty is not the batch rate: skip the rescale."""
from erpnext.stock.doctype.item.test_item import make_item
item_code = make_item(
"Test Stock Ageing Partial Batch Reco",
{"is_stock_item": 1, "has_batch_no": 1, "valuation_method": "FIFO"},
).name
batch_no = "SA-PARTIAL-RECO-BATCH"
if not frappe.db.exists("Batch", batch_no):
frappe.get_doc({"doctype": "Batch", "batch_id": batch_no, "item": item_code}).insert(
ignore_permissions=True
)
frappe.db.set_value("Batch", batch_no, "use_batchwise_valuation", 1)
def make_sle(posting_date, voucher_type, voucher_no, actual_qty, qty_after, stock_value_difference):
return frappe._dict(
name=item_code,
actual_qty=actual_qty,
qty_after_transaction=qty_after,
stock_value_difference=stock_value_difference,
valuation_rate=abs(stock_value_difference / actual_qty) if actual_qty else 0,
warehouse="WH 1",
posting_date=posting_date,
voucher_type=voucher_type,
voucher_no=voucher_no,
has_serial_no=False,
has_batch_no=True,
serial_no=None,
batch_no=batch_no,
)
sle = [
make_sle("2021-12-01", "Stock Entry", "001", 10, 10, 1000),
make_sle("2021-12-02", "Stock Reconciliation", "002", 0, 12, -400),
]
slots = FIFOSlots(self.filters, sle).generate()
queue = slots[item_code]["fifo_queue"]
self.assertEqual(
queue,
[
[batch_no, 1, 10.0, "2021-12-01", 1000.0],
[batch_no, 1, 2.0, "2021-12-01", 400.0],
],
)
def test_sequential_stock_reco_same_warehouse(self):
"""
Test back to back stock recos (same warehouse).

View File

@@ -206,7 +206,7 @@ def create_reposting_entries(rows: str | list, company: str):
"posting_date": sle.posting_date,
"posting_time": sle.posting_time,
"company": company,
"allow_nagative_stock": 1,
"allow_negative_stock": 1,
}
).submit()
@@ -247,7 +247,7 @@ def repost_based_on_transaction(rows, company=None, entries=None):
"posting_date": row.get("posting_date"),
"posting_time": row.get("posting_time"),
"company": company,
"allow_nagative_stock": 1,
"allow_negative_stock": 1,
"recalculate_valuation_rate": 1,
}
).submit()

View File

@@ -384,8 +384,9 @@ class StockBalanceReport:
sle.batch_no,
sle.serial_no,
sle.serial_and_batch_bundle,
sle.has_serial_no,
sle.voucher_detail_no,
item_table.has_serial_no,
item_table.has_batch_no,
item_table.item_group,
item_table.stock_uom,
item_table.item_name,

View File

@@ -325,7 +325,7 @@ def create_reposting_entries(rows, item_code=None, warehouse=None):
"warehouse": warehouse or row.warehouse,
"posting_date": row.posting_date,
"posting_time": row.posting_time,
"allow_nagative_stock": 1,
"allow_negative_stock": 1,
}
).submit()

View File

@@ -84,6 +84,7 @@ def execute(filters=None):
bin.reserved_qty_for_production_plan,
bin.reserved_qty_for_sub_contract,
reserved_qty_for_pos,
bin.reserved_stock,
bin.projected_qty,
re_order_level,
re_order_qty,
@@ -200,6 +201,13 @@ def get_columns():
"width": 100,
"convertible": "qty",
},
{
"label": _("Reserved Stock"),
"fieldname": "reserved_stock",
"fieldtype": "Float",
"width": 100,
"convertible": "qty",
},
{
"label": _("Projected Qty"),
"fieldname": "projected_qty",
@@ -246,6 +254,7 @@ def get_bin_list(filters):
bin.reserved_qty_for_production,
bin.reserved_qty_for_sub_contract,
bin.reserved_qty_for_production_plan,
bin.reserved_stock,
bin.projected_qty,
)
.orderby(bin.item_code, bin.warehouse)

View File

@@ -0,0 +1,6 @@
<p>{{_("Dear")}} {{ full_name }},</p>
<p>{{_("Your email has been verified and your appointment has been confirmed for {0}").format(scheduled_time)}}.</p>
<p>{{_("We look forward to meeting you")}}.</p>
<br>
<p style="font-size: 85%;">{{_("This email was sent from {0}").format(site_url)}}</p>

View File

@@ -1,6 +1,7 @@
<p>{{_("Dear")}} {{ full_name }}{% if last_name %} {{ last_name}}{% endif %},</p>
<p>{{_("A new appointment has been created for you with {0}").format(site_url)}}.</p>
<p>{{_("Click on the link below to verify your email and confirm the appointment")}}.</p>
<p>{{_("This link is valid for {0} minutes").format(expiry_minutes)}}.</p>
<p style="margin: 30px 0px;">
<a href="{{ link }}" rel="nofollow" style="padding: 8px 20px; background-color: #7575ff; color: #fff; border-radius: 4px; text-decoration: none; line-height: 1; border-bottom: 3px solid rgba(0, 0, 0, 0.2); font-size: 14px; font-weight: 200;">{{ _("Verify Email") }}</a>

View File

@@ -6,21 +6,12 @@ import frappe
def get_context(context):
project_user = frappe.db.get_value(
"Project User",
{"parent": frappe.form_dict.project, "user": frappe.session.user},
["user", "view_attachments", "hide_timesheets"],
as_dict=True,
)
if frappe.session.user != "Administrator" and (not project_user or frappe.session.user == "Guest"):
raise frappe.PermissionError
project_user = validate_and_get_project_user(project=frappe.form_dict.project)
context.no_cache = 1
context.show_sidebar = True
project = frappe.get_doc("Project", frappe.form_dict.project)
project.has_permission("read")
project.tasks = get_tasks(
project.name, start=0, item_status="open", search=frappe.form_dict.get("search")
)
@@ -64,6 +55,22 @@ def get_tasks(project, start=0, search=None, item_status=None):
return list(filter(lambda x: not x.parent_task, tasks))
@frappe.whitelist()
def get_task_html(project: str, start: int = 0, item_status: str | None = None):
validate_and_get_project_user(project=project)
return frappe.render_template(
"erpnext/templates/includes/projects/project_tasks.html",
{
"doc": {
"name": project,
"project_name": project,
"tasks": get_tasks(project, start, item_status=item_status),
}
},
is_path=True,
)
def get_timesheets(project, start=0, search=None):
filters = {"project": project}
if search:
@@ -89,9 +96,28 @@ def get_timesheets(project, start=0, search=None):
return timesheets
@frappe.whitelist()
def get_timesheet_html(project: str, start: int = 0):
validate_and_get_project_user(project=project)
return frappe.render_template(
"erpnext/templates/includes/projects/project_timesheets.html",
{"doc": {"timesheets": get_timesheets(project, start)}},
is_path=True,
)
def get_attachments(project):
return frappe.get_all(
"File",
filters={"attached_to_name": project, "attached_to_doctype": "Project", "is_private": 0},
fields=["file_name", "file_url", "file_size"],
)
def validate_and_get_project_user(project: str):
project_doc = frappe.get_doc("Project", project)
project_doc.check_permission()
project_user = next((d for d in project_doc.users if d.user == frappe.session.user), None)
return project_user

View File

@@ -0,0 +1,72 @@
# Copyright (c) 2026, Frappe Technologies Pvt. Ltd. and Contributors
# License: GNU General Public License v3. See license.txt
import frappe
from frappe.tests.utils import FrappeTestCase
from erpnext.projects.doctype.project.test_project import make_project
from erpnext.templates.pages.projects import validate_and_get_project_user
class TestProjectsPage(FrappeTestCase):
"""validate_and_get_project_user() gates the /projects portal page. It must raise
frappe.PermissionError for a user who can't read the Project, and otherwise return
that user's Project User row (or None if they're permitted but not listed as one --
e.g. an internal Projects Manager browsing the portal)."""
def _create_user(self, email):
if not frappe.db.exists("User", email):
frappe.get_doc(
{
"doctype": "User",
"email": email,
"first_name": "Portal",
"send_welcome_email": 0,
}
).insert(ignore_permissions=True)
return email
def test_raises_permission_error_for_user_without_access(self):
project = make_project({"project_name": f"_Test Portal Access {frappe.generate_hash(length=6)}"})
outsider = self._create_user(f"outsider_{frappe.generate_hash(length=6)}@example.com")
with self.set_user(outsider):
self.assertRaises(frappe.PermissionError, validate_and_get_project_user, project.name)
def test_allows_user_listed_as_project_user_and_returns_their_row(self):
# Being a Project User shares the Project with that user (see
# Project.control_access_for_project_users), which is what lets them past
# check_permission() here.
member = self._create_user(f"member_{frappe.generate_hash(length=6)}@example.com")
project = frappe.get_doc(
doctype="Project",
project_name=f"_Test Portal Access {frappe.generate_hash(length=6)}",
status="Open",
company="_Test Company",
)
project.append(
"users", {"user": member, "view_attachments": 1, "hide_timesheets": 1, "welcome_email_sent": 1}
)
project.insert()
with self.set_user(member):
project_user = validate_and_get_project_user(project.name)
self.assertIsNotNone(project_user)
self.assertEqual(project_user.user, member)
self.assertEqual(project_user.view_attachments, 1)
self.assertEqual(project_user.hide_timesheets, 1)
def test_allows_internally_permitted_user_not_listed_as_project_user(self):
# The permission gate must be the real permission system (check_permission()),
# not "is this user in the Project's users child table" -- a Projects Manager
# can open any project's portal page without ever being added as its user.
project = make_project({"project_name": f"_Test Portal Access {frappe.generate_hash(length=6)}"})
manager = self._create_user(f"manager_{frappe.generate_hash(length=6)}@example.com")
frappe.get_doc("User", manager).add_roles("Projects Manager")
with self.set_user(manager):
project_user = validate_and_get_project_user(project.name)
self.assertIsNone(project_user)

View File

@@ -47,7 +47,11 @@ def payment_app_import_guard():
msg = _("payments app is not installed. Please install it from {} or {}").format(
marketplace_link, github_link
)
if "payments" not in frappe.get_installed_apps():
frappe.throw(msg, title=_("Missing Payments App"), exc=frappe.AppNotInstalledError)
try:
yield
except ImportError:
frappe.throw(msg, title=_("Missing Payments App"))
frappe.throw(msg, title=_("Missing Payments App"), exc=frappe.AppNotInstalledError)

View File

@@ -3,9 +3,9 @@
import frappe
from apiclient.discovery import build
from frappe import _
from frappe.model.document import Document
from pyyoutube import Api, PyYouTubeException
class VideoSettings(Document):
@@ -28,7 +28,7 @@ class VideoSettings(Document):
def validate_youtube_api_key(self):
if self.enable_youtube_tracking and self.api_key:
try:
build("youtube", "v3", developerKey=self.api_key)
Api(api_key=self.api_key).get_i18n_languages(parts="snippet")
except Exception:
title = _("Failed to Authenticate the API key.")
self.log_error("Failed to authenticate API key")

View File

@@ -237,9 +237,9 @@ async function submit() {
frappe.show_alert(__("Appointment Created Successfully"));
}
setTimeout(() => {
let redirect_url = "/";
let redirect_url = "/book_appointment";
if (window.appointment_settings.success_redirect_url) {
redirect_url += window.appointment_settings.success_redirect_url;
redirect_url = `/${window.appointment_settings.success_redirect_url}`;
}
window.location.href = redirect_url;
}, 5000);

View File

@@ -4,6 +4,7 @@ import json
import frappe
import pytz
from frappe import _
from frappe.rate_limiter import rate_limit
from frappe.utils.data import get_system_timezone
WEEKDAYS = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
@@ -18,7 +19,7 @@ def get_context(context):
def handle_appointment_booking_disabled():
if not frappe.get_single_value("Appointment Booking Settings", "enable_scheduling"):
if not frappe.get_single_value("Appointment Booking Settings", "enable_appointment_portal"):
frappe.redirect_to_message(
_("Appointment Scheduling Disabled"),
_("Appointment Scheduling has been disabled for this site"),
@@ -66,6 +67,8 @@ def get_appointment_slots(date, timezone):
)
holiday_list = frappe.get_doc("Holiday List", settings.holiday_list)
timeslots = get_available_slots_between(query_start_time, query_end_time, settings)
# fetch the day's booked slots once instead of querying per timeslot
booked_times = get_booked_slot_times_for(timeslots, settings.appointment_duration)
# Filter and convert timeslots
converted_timeslots = []
@@ -76,7 +79,7 @@ def get_appointment_slots(date, timezone):
converted_timeslots.append(dict(time=converted_timeslot, availability=False))
continue
# Check availability
if check_availabilty(timeslot, settings) and converted_timeslot >= now:
if is_slot_available(timeslot, booked_times, settings) and converted_timeslot >= now:
converted_timeslots.append(dict(time=converted_timeslot, availability=True))
else:
converted_timeslots.append(dict(time=converted_timeslot, availability=False))
@@ -102,7 +105,8 @@ def get_available_slots_between(query_start_time, query_end_time, settings):
return timeslots
@frappe.whitelist(allow_guest=True)
@frappe.whitelist(allow_guest=True, methods=["POST"])
@rate_limit(limit=5, seconds=300)
def create_appointment(date, time, tz, contact):
handle_appointment_booking_disabled()
format_string = "%Y-%m-%d %H:%M:%S"
@@ -114,13 +118,13 @@ def create_appointment(date, time, tz, contact):
# Create a appointment document from form
appointment = frappe.new_doc("Appointment")
appointment.scheduled_time = scheduled_time
contact = json.loads(contact)
contact = frappe.parse_json(contact)
appointment.customer_name = contact.get("name", None)
appointment.customer_phone_number = contact.get("number", None)
appointment.customer_skype = contact.get("skype", None)
appointment.customer_details = contact.get("notes", None)
appointment.customer_email = contact.get("email", None)
appointment.status = "Open"
appointment.created_through_portal = 1
appointment.insert(ignore_permissions=True)
return appointment
@@ -150,8 +154,23 @@ def convert_to_system_timezone(guest_tz, datetimeobject):
return datetimeobject
def check_availabilty(timeslot, settings):
return frappe.db.count("Appointment", {"scheduled_time": timeslot}) < settings.number_of_agents
def get_booked_slot_times_for(timeslots, appointment_duration):
if not timeslots:
return []
from erpnext.crm.doctype.appointment.appointment import get_booked_slot_times
duration = datetime.timedelta(minutes=appointment_duration)
return get_booked_slot_times(min(timeslots) - duration, max(timeslots) + duration)
def is_slot_available(timeslot, booked_times, settings):
# mirror the server capacity check: count non-Closed appointments whose
# duration window overlaps this slot, without a per-slot query
duration = datetime.timedelta(minutes=settings.appointment_duration)
lower, upper = timeslot - duration, timeslot + duration
overlapping = sum(1 for booked in booked_times if lower < booked < upper)
return overlapping < settings.number_of_agents
def _is_holiday(date, holiday_list):

View File

@@ -12,7 +12,7 @@
</div>
{% else %}
<div class="alert alert-danger">
{{ _("Verification failed please check the link") }}
{{ message or _("Verification failed please check the link") }}
</div>
{% endif %}
{% endblock%}

View File

@@ -1,20 +1,58 @@
import frappe
from frappe.utils.verified_command import verify_request
from frappe import _
from frappe.utils import add_to_date, now_datetime
from frappe.utils.data import sha256_hash
from erpnext.crm.doctype.appointment.appointment import get_verification_link_expiry
def get_context(context):
if not verify_request():
key = frappe.form_dict.get("key")
if not key:
context.success = False
return context
email = frappe.form_dict["email"]
appointment_name = frappe.form_dict["appointment"]
appointment_name = frappe.db.get_value("Appointment", {"verification_token": sha256_hash(key)}, "name")
if not appointment_name:
context.success = False
context.message = _("This verification link is invalid. Please book the appointment again.")
return context
if email and appointment_name:
appointment = frappe.get_doc("Appointment", appointment_name)
appointment.set_verified(email)
appointment = frappe.get_doc("Appointment", appointment_name)
# report a settled status before expiry: a closed/verified appointment is
# more informative than a generic "expired" (and creation-based expiry would
# otherwise mask a sweeper-closed appointment)
if appointment.status == "Closed":
context.success = False
context.message = _("Appointment has been closed. Please book the appointment again.")
return context
if appointment.status == "Open":
context.success = True
context.message = _("Appointment is already verified.")
return context
else:
if now_datetime() > add_to_date(appointment.creation, minutes=get_verification_link_expiry()):
context.success = False
context.message = _("Verification link has expired.")
return context
verify_appointment(appointment)
# GET requests are rolled back at the end of the request unless this flag is set
frappe.local.flags.commit = True
context.success = True
return context
def verify_appointment(appointment):
# the signed link is the authorization; materializing the appointment
# (agent assignment) needs system privileges the Guest visitor lacks
visitor = frappe.session.user
try:
frappe.set_user("Administrator")
appointment.email_verified = True
appointment.status = "Open"
appointment.save(ignore_permissions=True)
finally:
frappe.set_user(visitor)

View File

@@ -18,7 +18,7 @@ dependencies = [
# integration dependencies
"googlemaps",
"plaid-python~=7.2.1",
"python-youtube~=0.8.0",
"python-youtube~=0.9.9",
# Not used directly - required by PyQRCode for PNG generation
"pypng~=0.20220715.0",
@@ -29,7 +29,7 @@ requires = ["flit_core >=3.4,<4"]
build-backend = "flit_core.buildapi"
[tool.bench.frappe-dependencies]
frappe = ">=15.40.4,<16.0.0"
frappe = ">=15.111.0,<16.0.0"
[tool.ruff]
line-length = 110