nsinnovations/erpnext
2
0
Fork 0
mirror of https://github.com/frappe/erpnext.git synced 2026-09-17 02:26:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

style: bulk format code with black

Browse Source 
v13 port because otherwise backports will result in conflicts always
This commit is contained in:
Ankush Menat
2022-03-29 17:29:34 +05:30
parent 7cc84dcbb4
commit c07713b860
1555 changed files with 96709 additions and 66138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
erpnext/support/__init__.py
View File

@@ -1,6 +1,5 @@
install_docs = [
{'doctype':'Role', 'role_name':'Support Team', 'name':'Support Team'},
{'doctype':'Role', 'role_name':'Maintenance User', 'name':'Maintenance User'},
{'doctype':'Role', 'role_name':'Maintenance Manager', 'name':'Maintenance Manager'}
{"doctype": "Role", "role_name": "Support Team", "name": "Support Team"},
{"doctype": "Role", "role_name": "Maintenance User", "name": "Maintenance User"},
{"doctype": "Role", "role_name": "Maintenance Manager", "name": "Maintenance Manager"},
]

344
erpnext/support/doctype/issue/issue.py
View File

@@ -67,8 +67,9 @@ class Issue(Document):
self.customer = contact.get_link_for("Customer")
if not self.company:
self.company = frappe.db.get_value("Lead", self.lead, "company") or \
frappe.db.get_default("Company")
self.company = frappe.db.get_value("Lead", self.lead, "company") or frappe.db.get_default(
"Company"
)
def reset_sla_fields(self):
self.agreement_status = ""
@@ -103,19 +104,20 @@ class Issue(Document):
def handle_hold_time(self, status):
if self.service_level_agreement:
# set response and resolution variance as None as the issue is on Hold
pause_sla_on = frappe.db.get_all("Pause SLA On Status", fields=["status"],
filters={"parent": self.service_level_agreement})
pause_sla_on = frappe.db.get_all(
"Pause SLA On Status", fields=["status"], filters={"parent": self.service_level_agreement}
)
hold_statuses = [entry.status for entry in pause_sla_on]
update_values = {}
if hold_statuses:
if self.status in hold_statuses and status not in hold_statuses:
update_values['on_hold_since'] = frappe.flags.current_time or now_datetime()
update_values["on_hold_since"] = frappe.flags.current_time or now_datetime()
if not self.first_responded_on:
update_values['response_by'] = None
update_values['response_by_variance'] = 0
update_values['resolution_by'] = None
update_values['resolution_by_variance'] = 0
update_values["response_by"] = None
update_values["response_by_variance"] = 0
update_values["resolution_by"] = None
update_values["resolution_by_variance"] = 0
# calculate hold time when status is changed from any hold status to any non-hold status
if self.status not in hold_statuses and status in hold_statuses:
@@ -125,7 +127,7 @@ class Issue(Document):
if self.on_hold_since:
# last_hold_time will be added to the sla variables
last_hold_time = time_diff_in_seconds(now_time, self.on_hold_since)
update_values['total_hold_time'] = hold_time + last_hold_time
update_values["total_hold_time"] = hold_time + last_hold_time
# re-calculate SLA variables after issue changes from any hold status to any non-hold status
# add hold time to SLA variables
@@ -134,25 +136,31 @@ class Issue(Document):
now_time = frappe.flags.current_time or now_datetime()
if not self.first_responded_on:
response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
response_by = get_expected_time_for(
parameter="response", service_level=priority, start_date_time=start_date_time
)
response_by = add_to_date(response_by, seconds=round(last_hold_time))
response_by_variance = round(time_diff_in_seconds(response_by, now_time))
update_values['response_by'] = response_by
update_values['response_by_variance'] = response_by_variance + last_hold_time
update_values["response_by"] = response_by
update_values["response_by_variance"] = response_by_variance + last_hold_time
resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
resolution_by = get_expected_time_for(
parameter="resolution", service_level=priority, start_date_time=start_date_time
)
resolution_by = add_to_date(resolution_by, seconds=round(last_hold_time))
resolution_by_variance = round(time_diff_in_seconds(resolution_by, now_time))
update_values['resolution_by'] = resolution_by
update_values['resolution_by_variance'] = resolution_by_variance + last_hold_time
update_values['on_hold_since'] = None
update_values["resolution_by"] = resolution_by
update_values["resolution_by_variance"] = resolution_by_variance + last_hold_time
update_values["on_hold_since"] = None
self.db_set(update_values)
def update_agreement_status(self):
if self.service_level_agreement and self.agreement_status == "Ongoing":
if cint(frappe.db.get_value("Issue", self.name, "response_by_variance")) < 0 or \
cint(frappe.db.get_value("Issue", self.name, "resolution_by_variance")) < 0:
if (
cint(frappe.db.get_value("Issue", self.name, "response_by_variance")) < 0
or cint(frappe.db.get_value("Issue", self.name, "resolution_by_variance")) < 0
):
self.agreement_status = "Failed"
else:
@@ -160,30 +168,34 @@ class Issue(Document):
def update_agreement_status_on_custom_status(self):
"""
Update Agreement Fulfilled status using Custom Scripts for Custom Issue Status
Update Agreement Fulfilled status using Custom Scripts for Custom Issue Status
"""
if not self.first_responded_on: # first_responded_on set when first reply is sent to customer
if not self.first_responded_on: # first_responded_on set when first reply is sent to customer
self.response_by_variance = round(time_diff_in_seconds(self.response_by, now_datetime()), 2)
if not self.resolution_date: # resolution_date set when issue has been closed
if not self.resolution_date: # resolution_date set when issue has been closed
self.resolution_by_variance = round(time_diff_in_seconds(self.resolution_by, now_datetime()), 2)
self.agreement_status = "Fulfilled" if self.response_by_variance > 0 and self.resolution_by_variance > 0 else "Failed"
self.agreement_status = (
"Fulfilled" if self.response_by_variance > 0 and self.resolution_by_variance > 0 else "Failed"
)
def create_communication(self):
communication = frappe.new_doc("Communication")
communication.update({
"communication_type": "Communication",
"communication_medium": "Email",
"sent_or_received": "Received",
"email_status": "Open",
"subject": self.subject,
"sender": self.raised_by,
"content": self.description,
"status": "Linked",
"reference_doctype": "Issue",
"reference_name": self.name
})
communication.update(
{
"communication_type": "Communication",
"communication_medium": "Email",
"sent_or_received": "Received",
"email_status": "Open",
"subject": self.subject,
"sender": self.raised_by,
"content": self.description,
"status": "Linked",
"reference_doctype": "Issue",
"reference_name": self.name,
}
)
communication.ignore_permissions = True
communication.ignore_mandatory = True
communication.save()
@@ -216,23 +228,31 @@ class Issue(Document):
# Replicate linked Communications
# TODO: get all communications in timeline before this, and modify them to append them to new doc
comm_to_split_from = frappe.get_doc("Communication", communication_id)
communications = frappe.get_all("Communication",
filters={"reference_doctype": "Issue",
communications = frappe.get_all(
"Communication",
filters={
"reference_doctype": "Issue",
"reference_name": comm_to_split_from.reference_name,
"creation": (">=", comm_to_split_from.creation)})
"creation": (">=", comm_to_split_from.creation),
},
)
for communication in communications:
doc = frappe.get_doc("Communication", communication.name)
doc.reference_name = replicated_issue.name
doc.save(ignore_permissions=True)
frappe.get_doc({
"doctype": "Comment",
"comment_type": "Info",
"reference_doctype": "Issue",
"reference_name": replicated_issue.name,
"content": " - Split the Issue from <a href='/app/Form/Issue/{0}'>{1}</a>".format(self.name, frappe.bold(self.name)),
}).insert(ignore_permissions=True)
frappe.get_doc(
{
"doctype": "Comment",
"comment_type": "Info",
"reference_doctype": "Issue",
"reference_name": replicated_issue.name,
"content": " - Split the Issue from <a href='/app/Form/Issue/{0}'>{1}</a>".format(
self.name, frappe.bold(self.name)
),
}
).insert(ignore_permissions=True)
return replicated_issue.name
@@ -243,7 +263,9 @@ class Issue(Document):
def before_insert(self):
if frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
if frappe.flags.in_test:
self.set_response_and_resolution_time(priority=self.priority, service_level_agreement=self.service_level_agreement)
self.set_response_and_resolution_time(
priority=self.priority, service_level_agreement=self.service_level_agreement
)
else:
self.set_response_and_resolution_time()
@@ -252,11 +274,19 @@ class Issue(Document):
if not service_level_agreement:
if frappe.db.get_value("Issue", self.name, "service_level_agreement"):
frappe.throw(_("Couldn't Set Service Level Agreement {0}.").format(self.service_level_agreement))
frappe.throw(
_("Couldn't Set Service Level Agreement {0}.").format(self.service_level_agreement)
)
return
if (service_level_agreement.customer and self.customer) and not (service_level_agreement.customer == self.customer):
frappe.throw(_("This Service Level Agreement is specific to Customer {0}").format(service_level_agreement.customer))
if (service_level_agreement.customer and self.customer) and not (
service_level_agreement.customer == self.customer
):
frappe.throw(
_("This Service Level Agreement is specific to Customer {0}").format(
service_level_agreement.customer
)
)
self.service_level_agreement = service_level_agreement.name
if not self.priority:
@@ -269,40 +299,59 @@ class Issue(Document):
self.service_level_agreement_creation = now_datetime()
start_date_time = get_datetime(self.service_level_agreement_creation)
self.response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
self.resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
self.response_by = get_expected_time_for(
parameter="response", service_level=priority, start_date_time=start_date_time
)
self.resolution_by = get_expected_time_for(
parameter="resolution", service_level=priority, start_date_time=start_date_time
)
self.response_by_variance = round(time_diff_in_seconds(self.response_by, now_datetime()))
self.resolution_by_variance = round(time_diff_in_seconds(self.resolution_by, now_datetime()))
def change_service_level_agreement_and_priority(self):
if self.service_level_agreement and frappe.db.exists("Issue", self.name) and \
frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
if (
self.service_level_agreement
and frappe.db.exists("Issue", self.name)
and frappe.db.get_single_value("Support Settings", "track_service_level_agreement")
):
if not self.priority == frappe.db.get_value("Issue", self.name, "priority"):
self.set_response_and_resolution_time(priority=self.priority, service_level_agreement=self.service_level_agreement)
self.set_response_and_resolution_time(
priority=self.priority, service_level_agreement=self.service_level_agreement
)
frappe.msgprint(_("Priority has been changed to {0}.").format(self.priority))
if not self.service_level_agreement == frappe.db.get_value("Issue", self.name, "service_level_agreement"):
self.set_response_and_resolution_time(priority=self.priority, service_level_agreement=self.service_level_agreement)
frappe.msgprint(_("Service Level Agreement has been changed to {0}.").format(self.service_level_agreement))
if not self.service_level_agreement == frappe.db.get_value(
"Issue", self.name, "service_level_agreement"
):
self.set_response_and_resolution_time(
priority=self.priority, service_level_agreement=self.service_level_agreement
)
frappe.msgprint(
_("Service Level Agreement has been changed to {0}.").format(self.service_level_agreement)
)
@frappe.whitelist()
def reset_service_level_agreement(self, reason, user):
if not frappe.db.get_single_value("Support Settings", "allow_resetting_service_level_agreement"):
frappe.throw(_("Allow Resetting Service Level Agreement from Support Settings."))
frappe.get_doc({
"doctype": "Comment",
"comment_type": "Info",
"reference_doctype": self.doctype,
"reference_name": self.name,
"comment_email": user,
"content": " resetted Service Level Agreement - {0}".format(_(reason)),
}).insert(ignore_permissions=True)
frappe.get_doc(
{
"doctype": "Comment",
"comment_type": "Info",
"reference_doctype": self.doctype,
"reference_name": self.name,
"comment_email": user,
"content": " resetted Service Level Agreement - {0}".format(_(reason)),
}
).insert(ignore_permissions=True)
self.service_level_agreement_creation = now_datetime()
self.set_response_and_resolution_time(priority=self.priority, service_level_agreement=self.service_level_agreement)
self.set_response_and_resolution_time(
priority=self.priority, service_level_agreement=self.service_level_agreement
)
self.agreement_status = "Ongoing"
self.save()
@@ -310,10 +359,12 @@ class Issue(Document):
def get_priority(issue):
service_level_agreement = frappe.get_doc("Service Level Agreement", issue.service_level_agreement)
priority = service_level_agreement.get_service_level_agreement_priority(issue.priority)
priority.update({
"support_and_resolution": service_level_agreement.support_and_resolution,
"holiday_list": service_level_agreement.holiday_list
})
priority.update(
{
"support_and_resolution": service_level_agreement.support_and_resolution,
"holiday_list": service_level_agreement.holiday_list,
}
)
return priority
@@ -334,10 +385,12 @@ def get_expected_time_for(parameter, service_level, start_date_time):
support_days = {}
for service in service_level.get("support_and_resolution"):
support_days[service.workday] = frappe._dict({
"start_time": service.start_time,
"end_time": service.end_time,
})
support_days[service.workday] = frappe._dict(
{
"start_time": service.start_time,
"end_time": service.end_time,
}
)
holidays = get_holidays(service_level.get("holiday_list"))
weekdays = get_weekdays()
@@ -346,14 +399,19 @@ def get_expected_time_for(parameter, service_level, start_date_time):
current_weekday = weekdays[current_date_time.weekday()]
if not is_holiday(current_date_time, holidays) and current_weekday in support_days:
start_time = current_date_time - datetime(current_date_time.year, current_date_time.month, current_date_time.day) \
if getdate(current_date_time) == getdate(start_date_time) and get_time_in_timedelta(current_date_time.time()) > support_days[current_weekday].start_time \
start_time = (
current_date_time
- datetime(current_date_time.year, current_date_time.month, current_date_time.day)
if getdate(current_date_time) == getdate(start_date_time)
and get_time_in_timedelta(current_date_time.time()) > support_days[current_weekday].start_time
else support_days[current_weekday].start_time
)
end_time = support_days[current_weekday].end_time
time_left_today = time_diff_in_seconds(end_time, start_time)
# no time left for support today
if time_left_today <= 0: pass
if time_left_today <= 0:
pass
elif allotted_seconds:
if time_left_today >= allotted_seconds:
expected_time = datetime.combine(getdate(current_date_time), get_time(start_time))
@@ -372,6 +430,7 @@ def get_expected_time_for(parameter, service_level, start_date_time):
return current_date_time
def set_service_level_agreement_variance(issue=None):
current_time = frappe.flags.current_time or now_datetime()
@@ -382,17 +441,25 @@ def set_service_level_agreement_variance(issue=None):
for issue in frappe.get_list("Issue", filters=filters):
doc = frappe.get_doc("Issue", issue.name)
if not doc.first_responded_on: # first_responded_on set when first reply is sent to customer
if not doc.first_responded_on: # first_responded_on set when first reply is sent to customer
variance = round(time_diff_in_seconds(doc.response_by, current_time), 2)
frappe.db.set_value(dt="Issue", dn=doc.name, field="response_by_variance", val=variance, update_modified=False)
frappe.db.set_value(
dt="Issue", dn=doc.name, field="response_by_variance", val=variance, update_modified=False
)
if variance < 0:
frappe.db.set_value(dt="Issue", dn=doc.name, field="agreement_status", val="Failed", update_modified=False)
frappe.db.set_value(
dt="Issue", dn=doc.name, field="agreement_status", val="Failed", update_modified=False
)
if not doc.resolution_date: # resolution_date set when issue has been closed
if not doc.resolution_date: # resolution_date set when issue has been closed
variance = round(time_diff_in_seconds(doc.resolution_by, current_time), 2)
frappe.db.set_value(dt="Issue", dn=doc.name, field="resolution_by_variance", val=variance, update_modified=False)
frappe.db.set_value(
dt="Issue", dn=doc.name, field="resolution_by_variance", val=variance, update_modified=False
)
if variance < 0:
frappe.db.set_value(dt="Issue", dn=doc.name, field="agreement_status", val="Failed", update_modified=False)
frappe.db.set_value(
dt="Issue", dn=doc.name, field="agreement_status", val="Failed", update_modified=False
)
def set_resolution_time(issue):
@@ -403,18 +470,20 @@ def set_resolution_time(issue):
def set_user_resolution_time(issue):
# total time taken by a user to close the issue apart from wait_time
communications = frappe.get_list("Communication", filters={
"reference_doctype": issue.doctype,
"reference_name": issue.name
},
communications = frappe.get_list(
"Communication",
filters={"reference_doctype": issue.doctype, "reference_name": issue.name},
fields=["sent_or_received", "name", "creation"],
order_by="creation"
order_by="creation",
)
pending_time = []
for i in range(len(communications)):
if communications[i].sent_or_received == "Received" and communications[i-1].sent_or_received == "Sent":
wait_time = time_diff_in_seconds(communications[i].creation, communications[i-1].creation)
if (
communications[i].sent_or_received == "Received"
and communications[i - 1].sent_or_received == "Sent"
):
wait_time = time_diff_in_seconds(communications[i].creation, communications[i - 1].creation)
if wait_time > 0:
pending_time.append(wait_time)
@@ -431,7 +500,7 @@ def get_list_context(context=None):
"row_template": "templates/includes/issue_row.html",
"show_sidebar": True,
"show_search": True,
"no_breadcrumbs": True
"no_breadcrumbs": True,
}
@@ -448,7 +517,8 @@ def get_issue_list(doctype, txt, filters, limit_start, limit_page_length=20, ord
ignore_permissions = False
if is_website_user():
if not filters: filters = {}
if not filters:
filters = {}
if customer:
filters["customer"] = customer
@@ -457,7 +527,9 @@ def get_issue_list(doctype, txt, filters, limit_start, limit_page_length=20, ord
ignore_permissions = True
return get_list(doctype, txt, filters, limit_start, limit_page_length, ignore_permissions=ignore_permissions)
return get_list(
doctype, txt, filters, limit_start, limit_page_length, ignore_permissions=ignore_permissions
)
@frappe.whitelist()
@@ -466,18 +538,26 @@ def set_multiple_status(names, status):
for name in names:
set_status(name, status)
@frappe.whitelist()
def set_status(name, status):
st = frappe.get_doc("Issue", name)
st.status = status
st.save()
def auto_close_tickets():
"""Auto-close replied support tickets after 7 days"""
auto_close_after_days = frappe.db.get_value("Support Settings", "Support Settings", "close_issue_after_days") or 7
auto_close_after_days = (
frappe.db.get_value("Support Settings", "Support Settings", "close_issue_after_days") or 7
)
issues = frappe.db.sql(""" select name from tabIssue where status='Replied' and
modified<DATE_SUB(CURDATE(), INTERVAL %s DAY) """, (auto_close_after_days), as_dict=True)
issues = frappe.db.sql(
""" select name from tabIssue where status='Replied' and
modified<DATE_SUB(CURDATE(), INTERVAL %s DAY) """,
(auto_close_after_days),
as_dict=True,
)
for issue in issues:
doc = frappe.get_doc("Issue", issue.get("name"))
@@ -486,80 +566,97 @@ def auto_close_tickets():
doc.flags.ignore_mandatory = True
doc.save()
def has_website_permission(doc, ptype, user, verbose=False):
from erpnext.controllers.website_list_for_contact import has_website_permission
permission_based_on_customer = has_website_permission(doc, ptype, user, verbose)
return permission_based_on_customer or doc.raised_by==user
return permission_based_on_customer or doc.raised_by == user
def update_issue(contact, method):
"""Called when Contact is deleted"""
frappe.db.sql("""UPDATE `tabIssue` set contact='' where contact=%s""", contact.name)
def get_holidays(holiday_list_name):
holiday_list = frappe.get_cached_doc("Holiday List", holiday_list_name)
holidays = [holiday.holiday_date for holiday in holiday_list.holidays]
return holidays
def is_holiday(date, holidays):
return getdate(date) in holidays
@frappe.whitelist()
def make_task(source_name, target_doc=None):
return get_mapped_doc("Issue", source_name, {
"Issue": {
"doctype": "Task"
}
}, target_doc)
return get_mapped_doc("Issue", source_name, {"Issue": {"doctype": "Task"}}, target_doc)
@frappe.whitelist()
def make_issue_from_communication(communication, ignore_communication_links=False):
""" raise a issue from email """
"""raise a issue from email"""
doc = frappe.get_doc("Communication", communication)
issue = frappe.get_doc({
"doctype": "Issue",
"subject": doc.subject,
"communication_medium": doc.communication_medium,
"raised_by": doc.sender or "",
"raised_by_phone": doc.phone_no or ""
}).insert(ignore_permissions=True)
issue = frappe.get_doc(
{
"doctype": "Issue",
"subject": doc.subject,
"communication_medium": doc.communication_medium,
"raised_by": doc.sender or "",
"raised_by_phone": doc.phone_no or "",
}
).insert(ignore_permissions=True)
link_communication_to_document(doc, "Issue", issue.name, ignore_communication_links)
return issue.name
def get_time_in_timedelta(time):
"""
Converts datetime.time(10, 36, 55, 961454) to datetime.timedelta(seconds=38215)
Converts datetime.time(10, 36, 55, 961454) to datetime.timedelta(seconds=38215)
"""
return timedelta(hours=time.hour, minutes=time.minute, seconds=time.second)
def set_first_response_time(communication, method):
if communication.get('reference_doctype') == "Issue":
if communication.get("reference_doctype") == "Issue":
issue = get_parent_doc(communication)
if is_first_response(issue) and issue.service_level_agreement:
first_response_time = calculate_first_response_time(issue, get_datetime(issue.first_responded_on))
first_response_time = calculate_first_response_time(
issue, get_datetime(issue.first_responded_on)
)
issue.db_set("first_response_time", first_response_time)
def is_first_response(issue):
responses = frappe.get_all('Communication', filters = {'reference_name': issue.name, 'sent_or_received': 'Sent'})
responses = frappe.get_all(
"Communication", filters={"reference_name": issue.name, "sent_or_received": "Sent"}
)
if len(responses) == 1:
return True
return False
def calculate_first_response_time(issue, first_responded_on):
issue_creation_date = issue.creation
issue_creation_time = get_time_in_seconds(issue_creation_date)
first_responded_on_in_seconds = get_time_in_seconds(first_responded_on)
support_hours = frappe.get_cached_doc("Service Level Agreement", issue.service_level_agreement).support_and_resolution
support_hours = frappe.get_cached_doc(
"Service Level Agreement", issue.service_level_agreement
).support_and_resolution
if issue_creation_date.day == first_responded_on.day:
if is_work_day(issue_creation_date, support_hours):
start_time, end_time = get_working_hours(issue_creation_date, support_hours)
# issue creation and response on the same day during working hours
if is_during_working_hours(issue_creation_date, support_hours) and is_during_working_hours(first_responded_on, support_hours):
if is_during_working_hours(issue_creation_date, support_hours) and is_during_working_hours(
first_responded_on, support_hours
):
return get_elapsed_time(issue_creation_date, first_responded_on)
# issue creation is during working hours, but first response was after working hours
@@ -572,7 +669,7 @@ def calculate_first_response_time(issue, first_responded_on):
# both issue creation and first response were after working hours
else:
return 1.0 # this should ideally be zero, but it gets reset when the next response is sent if the value is zero
return 1.0 # this should ideally be zero, but it gets reset when the next response is sent if the value is zero
else:
return 1.0
@@ -582,7 +679,9 @@ def calculate_first_response_time(issue, first_responded_on):
if date_diff(first_responded_on, issue_creation_date) == 1:
first_response_time = 0
else:
first_response_time = calculate_initial_frt(issue_creation_date, date_diff(first_responded_on, issue_creation_date)- 1, support_hours)
first_response_time = calculate_initial_frt(
issue_creation_date, date_diff(first_responded_on, issue_creation_date) - 1, support_hours
)
# time taken on day of issue creation
if is_work_day(issue_creation_date, support_hours):
@@ -607,9 +706,11 @@ def calculate_first_response_time(issue, first_responded_on):
else:
return 1.0
def get_time_in_seconds(date):
return timedelta(hours=date.hour, minutes=date.minute, seconds=date.second)
def get_working_hours(date, support_hours):
if is_work_day(date, support_hours):
weekday = frappe.utils.get_weekday(date)
@@ -617,6 +718,7 @@ def get_working_hours(date, support_hours):
if day.workday == weekday:
return day.start_time, day.end_time
def is_work_day(date, support_hours):
weekday = frappe.utils.get_weekday(date)
for day in support_hours:
@@ -624,6 +726,7 @@ def is_work_day(date, support_hours):
return True
return False
def is_during_working_hours(date, support_hours):
start_time, end_time = get_working_hours(date, support_hours)
time = get_time_in_seconds(date)
@@ -631,19 +734,22 @@ def is_during_working_hours(date, support_hours):
return True
return False
def get_elapsed_time(start_time, end_time):
return round(time_diff_in_seconds(end_time, start_time), 2)
def calculate_initial_frt(issue_creation_date, days_in_between, support_hours):
initial_frt = 0
for i in range(days_in_between):
date = issue_creation_date + timedelta(days = (i+1))
date = issue_creation_date + timedelta(days=(i + 1))
if is_work_day(date, support_hours):
start_time, end_time = get_working_hours(date, support_hours)
initial_frt += get_elapsed_time(start_time, end_time)
return initial_frt
def is_before_working_hours(date, support_hours):
start_time, end_time = get_working_hours(date, support_hours)
time = get_time_in_seconds(date)

11
erpnext/support/doctype/issue/issue_dashboard.py
View File

@@ -1,14 +1,5 @@
from frappe import _
def get_data():
return {
'fieldname': 'issue',
'transactions': [
{
'label': _('Activity'),
'items': ['Task']
}
]
}
return {"fieldname": "issue", "transactions": [{"label": _("Activity"), "items": ["Task"]}]}

284
erpnext/support/doctype/issue/test_issue.py
View File

@@ -18,6 +18,7 @@ class TestSetUp(unittest.TestCase):
frappe.db.set_value("Support Settings", None, "track_service_level_agreement", 1)
create_service_level_agreements_for_issues()
class TestIssue(TestSetUp):
def test_response_time_and_resolution_time_based_on_different_sla(self):
creation = get_datetime("2019-03-04 12:00")
@@ -36,9 +37,10 @@ class TestIssue(TestSetUp):
self.assertEqual(issue.response_by, get_datetime("2019-03-04 14:00"))
self.assertEqual(issue.resolution_by, get_datetime("2019-03-04 15:00"))
# make issue with territory specific SLA
customer = create_customer("___Test Customer", "__Test SLA Customer Group", "_Test SLA Territory")
customer = create_customer(
"___Test Customer", "__Test SLA Customer Group", "_Test SLA Territory"
)
issue = make_issue(creation, "___Test Customer", 3)
self.assertEqual(issue.response_by, get_datetime("2019-03-04 14:00"))
@@ -73,10 +75,10 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2019-03-04 15:00")
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
self.assertEqual(issue.agreement_status, 'Fulfilled')
self.assertEqual(issue.agreement_status, "Fulfilled")
def test_issue_metrics(self):
creation = get_datetime("2020-03-04 4:00")
@@ -95,7 +97,7 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2020-03-04 5:05")
issue.reload()
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
self.assertEqual(issue.avg_response_time, 600)
@@ -113,7 +115,7 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2020-03-04 4:15")
issue.reload()
issue.status = 'Replied'
issue.status = "Replied"
issue.save()
self.assertEqual(issue.on_hold_since, frappe.flags.current_time)
@@ -131,12 +133,13 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2020-03-04 5:05")
issue.reload()
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
issue.reload()
self.assertEqual(flt(issue.total_hold_time, 2), 2700)
class TestFirstResponseTime(TestSetUp):
# working hours used in all cases: Mon-Fri, 10am to 6pm
# all dates are in the mm-dd-yyyy format
@@ -144,209 +147,268 @@ class TestFirstResponseTime(TestSetUp):
# issue creation and first response are on the same day
def test_first_response_time_case1(self):
"""
Test frt when issue creation and first response are during working hours on the same day.
Test frt when issue creation and first response are during working hours on the same day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 11:00"), get_datetime("06-28-2021 12:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 11:00"), get_datetime("06-28-2021 12:00")
)
self.assertEqual(issue.first_response_time, 3600.0)
def test_first_response_time_case2(self):
"""
Test frt when issue creation was during working hours, but first response is sent after working hours on the same day.
Test frt when issue creation was during working hours, but first response is sent after working hours on the same day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("06-28-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("06-28-2021 20:00")
)
self.assertEqual(issue.first_response_time, 21600.0)
def test_first_response_time_case3(self):
"""
Test frt when issue creation was before working hours but first response is sent during working hours on the same day.
Test frt when issue creation was before working hours but first response is sent during working hours on the same day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("06-28-2021 12:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("06-28-2021 12:00")
)
self.assertEqual(issue.first_response_time, 7200.0)
def test_first_response_time_case4(self):
"""
Test frt when both issue creation and first response were after working hours on the same day.
Test frt when both issue creation and first response were after working hours on the same day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 19:00"), get_datetime("06-28-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 19:00"), get_datetime("06-28-2021 20:00")
)
self.assertEqual(issue.first_response_time, 1.0)
def test_first_response_time_case5(self):
"""
Test frt when both issue creation and first response are on the same day, but it's not a work day.
Test frt when both issue creation and first response are on the same day, but it's not a work day.
"""
issue = create_issue_and_communication(get_datetime("06-27-2021 10:00"), get_datetime("06-27-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-27-2021 10:00"), get_datetime("06-27-2021 11:00")
)
self.assertEqual(issue.first_response_time, 1.0)
# issue creation and first response are on consecutive days
def test_first_response_time_case6(self):
"""
Test frt when the issue was created before working hours and the first response is also sent before working hours, but on the next day.
Test frt when the issue was created before working hours and the first response is also sent before working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 6:00")
)
self.assertEqual(issue.first_response_time, 28800.0)
def test_first_response_time_case7(self):
"""
Test frt when the issue was created before working hours and the first response is sent during working hours, but on the next day.
Test frt when the issue was created before working hours and the first response is sent during working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 11:00")
)
self.assertEqual(issue.first_response_time, 32400.0)
def test_first_response_time_case8(self):
"""
Test frt when the issue was created before working hours and the first response is sent after working hours, but on the next day.
Test frt when the issue was created before working hours and the first response is sent after working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("06-29-2021 20:00")
)
self.assertEqual(issue.first_response_time, 57600.0)
def test_first_response_time_case9(self):
"""
Test frt when the issue was created before working hours and the first response is sent on the next day, which is not a work day.
Test frt when the issue was created before working hours and the first response is sent on the next day, which is not a work day.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 6:00"), get_datetime("06-26-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 6:00"), get_datetime("06-26-2021 11:00")
)
self.assertEqual(issue.first_response_time, 28800.0)
def test_first_response_time_case10(self):
"""
Test frt when the issue was created during working hours and the first response is sent before working hours, but on the next day.
Test frt when the issue was created during working hours and the first response is sent before working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 6:00")
)
self.assertEqual(issue.first_response_time, 21600.0)
def test_first_response_time_case11(self):
"""
Test frt when the issue was created during working hours and the first response is also sent during working hours, but on the next day.
Test frt when the issue was created during working hours and the first response is also sent during working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 11:00")
)
self.assertEqual(issue.first_response_time, 25200.0)
def test_first_response_time_case12(self):
"""
Test frt when the issue was created during working hours and the first response is sent after working hours, but on the next day.
Test frt when the issue was created during working hours and the first response is sent after working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("06-29-2021 20:00")
)
self.assertEqual(issue.first_response_time, 50400.0)
def test_first_response_time_case13(self):
"""
Test frt when the issue was created during working hours and the first response is sent on the next day, which is not a work day.
Test frt when the issue was created during working hours and the first response is sent on the next day, which is not a work day.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 12:00"), get_datetime("06-26-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 12:00"), get_datetime("06-26-2021 11:00")
)
self.assertEqual(issue.first_response_time, 21600.0)
def test_first_response_time_case14(self):
"""
Test frt when the issue was created after working hours and the first response is sent before working hours, but on the next day.
Test frt when the issue was created after working hours and the first response is sent before working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 6:00")
)
self.assertEqual(issue.first_response_time, 1.0)
def test_first_response_time_case15(self):
"""
Test frt when the issue was created after working hours and the first response is sent during working hours, but on the next day.
Test frt when the issue was created after working hours and the first response is sent during working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 11:00")
)
self.assertEqual(issue.first_response_time, 3600.0)
def test_first_response_time_case16(self):
"""
Test frt when the issue was created after working hours and the first response is also sent after working hours, but on the next day.
Test frt when the issue was created after working hours and the first response is also sent after working hours, but on the next day.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("06-29-2021 20:00")
)
self.assertEqual(issue.first_response_time, 28800.0)
def test_first_response_time_case17(self):
"""
Test frt when the issue was created after working hours and the first response is sent on the next day, which is not a work day.
Test frt when the issue was created after working hours and the first response is sent on the next day, which is not a work day.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 20:00"), get_datetime("06-26-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 20:00"), get_datetime("06-26-2021 11:00")
)
self.assertEqual(issue.first_response_time, 1.0)
# issue creation and first response are a few days apart
def test_first_response_time_case18(self):
"""
Test frt when the issue was created before working hours and the first response is also sent before working hours, but after a few days.
Test frt when the issue was created before working hours and the first response is also sent before working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 6:00")
)
self.assertEqual(issue.first_response_time, 86400.0)
def test_first_response_time_case19(self):
"""
Test frt when the issue was created before working hours and the first response is sent during working hours, but after a few days.
Test frt when the issue was created before working hours and the first response is sent during working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 11:00")
)
self.assertEqual(issue.first_response_time, 90000.0)
def test_first_response_time_case20(self):
"""
Test frt when the issue was created before working hours and the first response is sent after working hours, but after a few days.
Test frt when the issue was created before working hours and the first response is sent after working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 6:00"), get_datetime("07-01-2021 20:00")
)
self.assertEqual(issue.first_response_time, 115200.0)
def test_first_response_time_case21(self):
"""
Test frt when the issue was created before working hours and the first response is sent after a few days, on a holiday.
Test frt when the issue was created before working hours and the first response is sent after a few days, on a holiday.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 6:00"), get_datetime("06-27-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 6:00"), get_datetime("06-27-2021 11:00")
)
self.assertEqual(issue.first_response_time, 28800.0)
def test_first_response_time_case22(self):
"""
Test frt when the issue was created during working hours and the first response is sent before working hours, but after a few days.
Test frt when the issue was created during working hours and the first response is sent before working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 6:00")
)
self.assertEqual(issue.first_response_time, 79200.0)
def test_first_response_time_case23(self):
"""
Test frt when the issue was created during working hours and the first response is also sent during working hours, but after a few days.
Test frt when the issue was created during working hours and the first response is also sent during working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 11:00")
)
self.assertEqual(issue.first_response_time, 82800.0)
def test_first_response_time_case24(self):
"""
Test frt when the issue was created during working hours and the first response is sent after working hours, but after a few days.
Test frt when the issue was created during working hours and the first response is sent after working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 12:00"), get_datetime("07-01-2021 20:00")
)
self.assertEqual(issue.first_response_time, 108000.0)
def test_first_response_time_case25(self):
"""
Test frt when the issue was created during working hours and the first response is sent after a few days, on a holiday.
Test frt when the issue was created during working hours and the first response is sent after a few days, on a holiday.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 12:00"), get_datetime("06-27-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 12:00"), get_datetime("06-27-2021 11:00")
)
self.assertEqual(issue.first_response_time, 21600.0)
def test_first_response_time_case26(self):
"""
Test frt when the issue was created after working hours and the first response is sent before working hours, but after a few days.
Test frt when the issue was created after working hours and the first response is sent before working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 6:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 6:00")
)
self.assertEqual(issue.first_response_time, 57600.0)
def test_first_response_time_case27(self):
"""
Test frt when the issue was created after working hours and the first response is sent during working hours, but after a few days.
Test frt when the issue was created after working hours and the first response is sent during working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 11:00")
)
self.assertEqual(issue.first_response_time, 61200.0)
def test_first_response_time_case28(self):
"""
Test frt when the issue was created after working hours and the first response is also sent after working hours, but after a few days.
Test frt when the issue was created after working hours and the first response is also sent after working hours, but after a few days.
"""
issue = create_issue_and_communication(get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 20:00"))
issue = create_issue_and_communication(
get_datetime("06-28-2021 20:00"), get_datetime("07-01-2021 20:00")
)
self.assertEqual(issue.first_response_time, 86400.0)
def test_first_response_time_case29(self):
"""
Test frt when the issue was created after working hours and the first response is sent after a few days, on a holiday.
Test frt when the issue was created after working hours and the first response is sent after a few days, on a holiday.
"""
issue = create_issue_and_communication(get_datetime("06-25-2021 20:00"), get_datetime("06-27-2021 11:00"))
issue = create_issue_and_communication(
get_datetime("06-25-2021 20:00"), get_datetime("06-27-2021 11:00")
)
self.assertEqual(issue.first_response_time, 1.0)
def create_issue_and_communication(issue_creation, first_responded_on):
issue = make_issue(issue_creation, index=1)
sender = create_user("test@admin.com")
@@ -355,20 +417,23 @@ def create_issue_and_communication(issue_creation, first_responded_on):
return issue
def make_issue(creation=None, customer=None, index=0, priority=None, issue_type=None):
issue = frappe.get_doc({
"doctype": "Issue",
"subject": "Service Level Agreement Issue {0}".format(index),
"customer": customer,
"raised_by": "test@example.com",
"description": "Service Level Agreement Issue",
"issue_type": issue_type,
"priority": priority,
"creation": creation,
"opening_date": creation,
"service_level_agreement_creation": creation,
"company": "_Test Company"
}).insert(ignore_permissions=True)
issue = frappe.get_doc(
{
"doctype": "Issue",
"subject": "Service Level Agreement Issue {0}".format(index),
"customer": customer,
"raised_by": "test@example.com",
"description": "Service Level Agreement Issue",
"issue_type": issue_type,
"priority": priority,
"creation": creation,
"opening_date": creation,
"service_level_agreement_creation": creation,
"company": "_Test Company",
}
).insert(ignore_permissions=True)
return issue
@@ -379,45 +444,50 @@ def create_customer(name, customer_group, territory):
create_territory(territory)
if not frappe.db.exists("Customer", {"customer_name": name}):
frappe.get_doc({
"doctype": "Customer",
"customer_name": name,
"customer_group": customer_group,
"territory": territory
}).insert(ignore_permissions=True)
frappe.get_doc(
{
"doctype": "Customer",
"customer_name": name,
"customer_group": customer_group,
"territory": territory,
}
).insert(ignore_permissions=True)
def create_customer_group(customer_group):
if not frappe.db.exists("Customer Group", {"customer_group_name": customer_group}):
frappe.get_doc({
"doctype": "Customer Group",
"customer_group_name": customer_group
}).insert(ignore_permissions=True)
frappe.get_doc({"doctype": "Customer Group", "customer_group_name": customer_group}).insert(
ignore_permissions=True
)
def create_territory(territory):
if not frappe.db.exists("Territory", {"territory_name": territory}):
frappe.get_doc({
"doctype": "Territory",
"territory_name": territory,
}).insert(ignore_permissions=True)
frappe.get_doc(
{
"doctype": "Territory",
"territory_name": territory,
}
).insert(ignore_permissions=True)
def create_communication(reference_name, sender, sent_or_received, creation):
communication = frappe.get_doc({
"doctype": "Communication",
"communication_type": "Communication",
"communication_medium": "Email",
"sent_or_received": sent_or_received,
"email_status": "Open",
"subject": "Test Issue",
"sender": sender,
"content": "Test",
"status": "Linked",
"reference_doctype": "Issue",
"creation": creation,
"reference_name": reference_name
})
communication.save()
communication = frappe.get_doc(
{
"doctype": "Communication",
"communication_type": "Communication",
"communication_medium": "Email",
"sent_or_received": sent_or_received,
"email_status": "Open",
"subject": "Test Issue",
"sender": sender,
"content": "Test",
"status": "Linked",
"reference_doctype": "Issue",
"creation": creation,
"reference_name": reference_name,
}
)
communication.save()

8
erpnext/support/doctype/issue_priority/test_issue_priority.py
View File

@@ -7,7 +7,6 @@ import frappe
class TestIssuePriority(unittest.TestCase):
def test_priorities(self):
make_priorities()
priorities = frappe.get_list("Issue Priority")
@@ -15,14 +14,13 @@ class TestIssuePriority(unittest.TestCase):
for priority in priorities:
self.assertIn(priority.name, ["Low", "Medium", "High"])
def make_priorities():
insert_priority("Low")
insert_priority("Medium")
insert_priority("High")
def insert_priority(name):
if not frappe.db.exists("Issue Priority", name):
frappe.get_doc({
"doctype": "Issue Priority",
"name": name
}).insert(ignore_permissions=True)
frappe.get_doc({"doctype": "Issue Priority", "name": name}).insert(ignore_permissions=True)

164
erpnext/support/doctype/service_level_agreement/service_level_agreement.py
View File

@@ -10,7 +10,6 @@ from frappe.utils.safe_exec import get_safe_globals
class ServiceLevelAgreement(Document):
def validate(self):
self.validate_doc()
self.check_priorities()
@@ -24,7 +23,11 @@ class ServiceLevelAgreement(Document):
for priority in self.priorities:
# Check if response and resolution time is set for every priority
if not priority.response_time or not priority.resolution_time:
frappe.throw(_("Set Response Time and Resolution Time for Priority {0} in row {1}.").format(priority.priority, priority.idx))
frappe.throw(
_("Set Response Time and Resolution Time for Priority {0} in row {1}.").format(
priority.priority, priority.idx
)
)
priorities.append(priority.priority)
@@ -35,7 +38,11 @@ class ServiceLevelAgreement(Document):
resolution = priority.resolution_time
if response > resolution:
frappe.throw(_("Response Time for {0} priority in row {1} can't be greater than Resolution Time.").format(priority.priority, priority.idx))
frappe.throw(
_("Response Time for {0} priority in row {1} can't be greater than Resolution Time.").format(
priority.priority, priority.idx
)
)
# Check if repeated priority
if not len(set(priorities)) == len(priorities):
@@ -59,15 +66,27 @@ class ServiceLevelAgreement(Document):
for support_and_resolution in self.support_and_resolution:
# Check if start and end time is set for every support day
if not (support_and_resolution.start_time or support_and_resolution.end_time):
frappe.throw(_("Set Start Time and End Time for \
Support Day {0} at index {1}.".format(support_and_resolution.workday, support_and_resolution.idx)))
frappe.throw(
_(
"Set Start Time and End Time for \
Support Day {0} at index {1}.".format(
support_and_resolution.workday, support_and_resolution.idx
)
)
)
support_days.append(support_and_resolution.workday)
support_and_resolution.idx = week.index(support_and_resolution.workday) + 1
if support_and_resolution.start_time >= support_and_resolution.end_time:
frappe.throw(_("Start Time can't be greater than or equal to End Time \
for {0}.".format(support_and_resolution.workday)))
frappe.throw(
_(
"Start Time can't be greater than or equal to End Time \
for {0}.".format(
support_and_resolution.workday
)
)
)
# Check for repeated workday
if not len(set(support_days)) == len(support_days):
@@ -75,12 +94,21 @@ class ServiceLevelAgreement(Document):
frappe.throw(_("Workday {0} has been repeated.").format(repeated_days))
def validate_doc(self):
if not frappe.db.get_single_value("Support Settings", "track_service_level_agreement") and self.enable:
frappe.throw(_("{0} is not enabled in {1}").format(frappe.bold("Track Service Level Agreement"),
get_link_to_form("Support Settings", "Support Settings")))
if (
not frappe.db.get_single_value("Support Settings", "track_service_level_agreement")
and self.enable
):
frappe.throw(
_("{0} is not enabled in {1}").format(
frappe.bold("Track Service Level Agreement"),
get_link_to_form("Support Settings", "Support Settings"),
)
)
if self.default_service_level_agreement:
if frappe.db.exists("Service Level Agreement", {"default_service_level_agreement": "1", "name": ["!=", self.name]}):
if frappe.db.exists(
"Service Level Agreement", {"default_service_level_agreement": "1", "name": ["!=", self.name]}
):
frappe.throw(_("A Default Service Level Agreement already exists."))
else:
if self.start_date and self.end_date:
@@ -91,11 +119,18 @@ class ServiceLevelAgreement(Document):
frappe.throw(_("End Date of Agreement can't be less than today."))
if self.entity_type and self.entity:
if frappe.db.exists("Service Level Agreement", {"entity_type": self.entity_type, "entity": self.entity, "name": ["!=", self.name]}):
frappe.throw(_("Service Level Agreement with Entity Type {0} and Entity {1} already exists.").format(self.entity_type, self.entity))
if frappe.db.exists(
"Service Level Agreement",
{"entity_type": self.entity_type, "entity": self.entity, "name": ["!=", self.name]},
):
frappe.throw(
_("Service Level Agreement with Entity Type {0} and Entity {1} already exists.").format(
self.entity_type, self.entity
)
)
def validate_condition(self):
temp_doc = frappe.new_doc('Issue')
temp_doc = frappe.new_doc("Issue")
if self.condition:
try:
frappe.safe_eval(self.condition, None, get_context(temp_doc))
@@ -105,58 +140,77 @@ class ServiceLevelAgreement(Document):
def get_service_level_agreement_priority(self, priority):
priority = frappe.get_doc("Service Level Priority", {"priority": priority, "parent": self.name})
return frappe._dict({
"priority": priority.priority,
"response_time": priority.response_time,
"resolution_time": priority.resolution_time
})
return frappe._dict(
{
"priority": priority.priority,
"response_time": priority.response_time,
"resolution_time": priority.resolution_time,
}
)
def check_agreement_status():
service_level_agreements = frappe.get_list("Service Level Agreement", filters=[
{"active": 1},
{"default_service_level_agreement": 0}
], fields=["name"])
service_level_agreements = frappe.get_list(
"Service Level Agreement",
filters=[{"active": 1}, {"default_service_level_agreement": 0}],
fields=["name"],
)
for service_level_agreement in service_level_agreements:
doc = frappe.get_doc("Service Level Agreement", service_level_agreement.name)
if doc.end_date and getdate(doc.end_date) < getdate(frappe.utils.getdate()):
frappe.db.set_value("Service Level Agreement", service_level_agreement.name, "active", 0)
def get_active_service_level_agreement_for(doc):
if not frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
return
filters = [
["Service Level Agreement", "active", "=", 1],
["Service Level Agreement", "enable", "=", 1]
["Service Level Agreement", "enable", "=", 1],
]
if doc.get('priority'):
filters.append(["Service Level Priority", "priority", "=", doc.get('priority')])
if doc.get("priority"):
filters.append(["Service Level Priority", "priority", "=", doc.get("priority")])
customer = doc.get('customer')
customer = doc.get("customer")
or_filters = [
["Service Level Agreement", "entity", "in", [customer, get_customer_group(customer), get_customer_territory(customer)]]
[
"Service Level Agreement",
"entity",
"in",
[customer, get_customer_group(customer), get_customer_territory(customer)],
]
]
service_level_agreement = doc.get('service_level_agreement')
service_level_agreement = doc.get("service_level_agreement")
if service_level_agreement:
or_filters = [
["Service Level Agreement", "name", "=", doc.get('service_level_agreement')],
["Service Level Agreement", "name", "=", doc.get("service_level_agreement")],
]
default_sla_filter = filters + [["Service Level Agreement", "default_service_level_agreement", "=", 1]]
default_sla = frappe.get_all("Service Level Agreement", filters=default_sla_filter,
fields=["name", "default_priority", "condition"])
default_sla_filter = filters + [
["Service Level Agreement", "default_service_level_agreement", "=", 1]
]
default_sla = frappe.get_all(
"Service Level Agreement",
filters=default_sla_filter,
fields=["name", "default_priority", "condition"],
)
filters += [["Service Level Agreement", "default_service_level_agreement", "=", 0]]
agreements = frappe.get_all("Service Level Agreement", filters=filters, or_filters=or_filters,
fields=["name", "default_priority", "condition"])
agreements = frappe.get_all(
"Service Level Agreement",
filters=filters,
or_filters=or_filters,
fields=["name", "default_priority", "condition"],
)
# check if the current document on which SLA is to be applied fulfills all the conditions
filtered_agreements = []
for agreement in agreements:
condition = agreement.get('condition')
condition = agreement.get("condition")
if not condition or (condition and frappe.safe_eval(condition, None, get_context(doc))):
filtered_agreements.append(agreement)
@@ -165,17 +219,25 @@ def get_active_service_level_agreement_for(doc):
return filtered_agreements[0] if filtered_agreements else None
def get_context(doc):
return {"doc": doc.as_dict(), "nowdate": nowdate, "frappe": frappe._dict(utils=get_safe_globals().get("frappe").get("utils"))}
return {
"doc": doc.as_dict(),
"nowdate": nowdate,
"frappe": frappe._dict(utils=get_safe_globals().get("frappe").get("utils")),
}
def get_customer_group(customer):
if customer:
return frappe.db.get_value("Customer", customer, "customer_group")
def get_customer_territory(customer):
if customer:
return frappe.db.get_value("Customer", customer, "territory")
@frappe.whitelist()
def get_service_level_agreement_filters(name, customer=None):
if not frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
@@ -183,25 +245,37 @@ def get_service_level_agreement_filters(name, customer=None):
filters = [
["Service Level Agreement", "active", "=", 1],
["Service Level Agreement", "enable", "=", 1]
["Service Level Agreement", "enable", "=", 1],
]
if not customer:
or_filters = [
["Service Level Agreement", "default_service_level_agreement", "=", 1]
]
or_filters = [["Service Level Agreement", "default_service_level_agreement", "=", 1]]
else:
# Include SLA with No Entity and Entity Type
or_filters = [
["Service Level Agreement", "entity", "in", [customer, get_customer_group(customer), get_customer_territory(customer), ""]],
["Service Level Agreement", "default_service_level_agreement", "=", 1]
[
"Service Level Agreement",
"entity",
"in",
[customer, get_customer_group(customer), get_customer_territory(customer), ""],
],
["Service Level Agreement", "default_service_level_agreement", "=", 1],
]
return {
"priority": [priority.priority for priority in frappe.get_list("Service Level Priority", filters={"parent": name}, fields=["priority"])],
"service_level_agreements": [d.name for d in frappe.get_list("Service Level Agreement", filters=filters, or_filters=or_filters)]
"priority": [
priority.priority
for priority in frappe.get_list(
"Service Level Priority", filters={"parent": name}, fields=["priority"]
)
],
"service_level_agreements": [
d.name
for d in frappe.get_list("Service Level Agreement", filters=filters, or_filters=or_filters)
],
}
def get_repeated(values):
unique_list = []
diff = []

9
erpnext/support/doctype/service_level_agreement/service_level_agreement_dashboard.py
View File

@@ -3,11 +3,6 @@ from frappe import _
def get_data():
return {
'fieldname': 'service_level_agreement',
'transactions': [
{
'label': _('Issue'),
'items': ['Issue']
}
]
"fieldname": "service_level_agreement",
"transactions": [{"label": _("Issue"), "items": ["Issue"]}],
}

411
erpnext/support/doctype/service_level_agreement/test_service_level_agreement.py
View File

@@ -16,55 +16,131 @@ class TestServiceLevelAgreement(unittest.TestCase):
def test_service_level_agreement(self):
# Default Service Level Agreement
create_default_service_level_agreement = create_service_level_agreement(default_service_level_agreement=1,
holiday_list="__Test Holiday List", employee_group="_Test Employee Group",
entity_type=None, entity=None, response_time=14400, resolution_time=21600)
create_default_service_level_agreement = create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
)
get_default_service_level_agreement = get_service_level_agreement(default_service_level_agreement=1)
get_default_service_level_agreement = get_service_level_agreement(
default_service_level_agreement=1
)
self.assertEqual(create_default_service_level_agreement.name, get_default_service_level_agreement.name)
self.assertEqual(create_default_service_level_agreement.entity_type, get_default_service_level_agreement.entity_type)
self.assertEqual(create_default_service_level_agreement.entity, get_default_service_level_agreement.entity)
self.assertEqual(create_default_service_level_agreement.default_service_level_agreement, get_default_service_level_agreement.default_service_level_agreement)
self.assertEqual(
create_default_service_level_agreement.name, get_default_service_level_agreement.name
)
self.assertEqual(
create_default_service_level_agreement.entity_type,
get_default_service_level_agreement.entity_type,
)
self.assertEqual(
create_default_service_level_agreement.entity, get_default_service_level_agreement.entity
)
self.assertEqual(
create_default_service_level_agreement.default_service_level_agreement,
get_default_service_level_agreement.default_service_level_agreement,
)
# Service Level Agreement for Customer
customer = create_customer()
create_customer_service_level_agreement = create_service_level_agreement(default_service_level_agreement=0,
holiday_list="__Test Holiday List", employee_group="_Test Employee Group",
entity_type="Customer", entity=customer, response_time=7200, resolution_time=10800)
get_customer_service_level_agreement = get_service_level_agreement(entity_type="Customer", entity=customer)
create_customer_service_level_agreement = create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Customer",
entity=customer,
response_time=7200,
resolution_time=10800,
)
get_customer_service_level_agreement = get_service_level_agreement(
entity_type="Customer", entity=customer
)
self.assertEqual(create_customer_service_level_agreement.name, get_customer_service_level_agreement.name)
self.assertEqual(create_customer_service_level_agreement.entity_type, get_customer_service_level_agreement.entity_type)
self.assertEqual(create_customer_service_level_agreement.entity, get_customer_service_level_agreement.entity)
self.assertEqual(create_customer_service_level_agreement.default_service_level_agreement, get_customer_service_level_agreement.default_service_level_agreement)
self.assertEqual(
create_customer_service_level_agreement.name, get_customer_service_level_agreement.name
)
self.assertEqual(
create_customer_service_level_agreement.entity_type,
get_customer_service_level_agreement.entity_type,
)
self.assertEqual(
create_customer_service_level_agreement.entity, get_customer_service_level_agreement.entity
)
self.assertEqual(
create_customer_service_level_agreement.default_service_level_agreement,
get_customer_service_level_agreement.default_service_level_agreement,
)
# Service Level Agreement for Customer Group
customer_group = create_customer_group()
create_customer_group_service_level_agreement = create_service_level_agreement(default_service_level_agreement=0,
holiday_list="__Test Holiday List", employee_group="_Test Employee Group",
entity_type="Customer Group", entity=customer_group, response_time=7200, resolution_time=10800)
get_customer_group_service_level_agreement = get_service_level_agreement(entity_type="Customer Group", entity=customer_group)
create_customer_group_service_level_agreement = create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Customer Group",
entity=customer_group,
response_time=7200,
resolution_time=10800,
)
get_customer_group_service_level_agreement = get_service_level_agreement(
entity_type="Customer Group", entity=customer_group
)
self.assertEqual(create_customer_group_service_level_agreement.name, get_customer_group_service_level_agreement.name)
self.assertEqual(create_customer_group_service_level_agreement.entity_type, get_customer_group_service_level_agreement.entity_type)
self.assertEqual(create_customer_group_service_level_agreement.entity, get_customer_group_service_level_agreement.entity)
self.assertEqual(create_customer_group_service_level_agreement.default_service_level_agreement, get_customer_group_service_level_agreement.default_service_level_agreement)
self.assertEqual(
create_customer_group_service_level_agreement.name,
get_customer_group_service_level_agreement.name,
)
self.assertEqual(
create_customer_group_service_level_agreement.entity_type,
get_customer_group_service_level_agreement.entity_type,
)
self.assertEqual(
create_customer_group_service_level_agreement.entity,
get_customer_group_service_level_agreement.entity,
)
self.assertEqual(
create_customer_group_service_level_agreement.default_service_level_agreement,
get_customer_group_service_level_agreement.default_service_level_agreement,
)
# Service Level Agreement for Territory
territory = create_territory()
create_territory_service_level_agreement = create_service_level_agreement(default_service_level_agreement=0,
holiday_list="__Test Holiday List", employee_group="_Test Employee Group",
entity_type="Territory", entity=territory, response_time=7200, resolution_time=10800)
get_territory_service_level_agreement = get_service_level_agreement(entity_type="Territory", entity=territory)
create_territory_service_level_agreement = create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Territory",
entity=territory,
response_time=7200,
resolution_time=10800,
)
get_territory_service_level_agreement = get_service_level_agreement(
entity_type="Territory", entity=territory
)
self.assertEqual(create_territory_service_level_agreement.name, get_territory_service_level_agreement.name)
self.assertEqual(create_territory_service_level_agreement.entity_type, get_territory_service_level_agreement.entity_type)
self.assertEqual(create_territory_service_level_agreement.entity, get_territory_service_level_agreement.entity)
self.assertEqual(create_territory_service_level_agreement.default_service_level_agreement, get_territory_service_level_agreement.default_service_level_agreement)
self.assertEqual(
create_territory_service_level_agreement.name, get_territory_service_level_agreement.name
)
self.assertEqual(
create_territory_service_level_agreement.entity_type,
get_territory_service_level_agreement.entity_type,
)
self.assertEqual(
create_territory_service_level_agreement.entity, get_territory_service_level_agreement.entity
)
self.assertEqual(
create_territory_service_level_agreement.default_service_level_agreement,
get_territory_service_level_agreement.default_service_level_agreement,
)
def get_service_level_agreement(default_service_level_agreement=None, entity_type=None, entity=None):
def get_service_level_agreement(
default_service_level_agreement=None, entity_type=None, entity=None
):
if default_service_level_agreement:
filters = {"default_service_level_agreement": default_service_level_agreement}
else:
@@ -73,93 +149,96 @@ def get_service_level_agreement(default_service_level_agreement=None, entity_typ
service_level_agreement = frappe.get_doc("Service Level Agreement", filters)
return service_level_agreement
def create_service_level_agreement(default_service_level_agreement, holiday_list, employee_group,
response_time, entity_type, entity, resolution_time):
def create_service_level_agreement(
default_service_level_agreement,
holiday_list,
employee_group,
response_time,
entity_type,
entity,
resolution_time,
):
employee_group = make_employee_group()
make_holiday_list()
make_priorities()
service_level_agreement = frappe.get_doc({
"doctype": "Service Level Agreement",
"enable": 1,
"service_level": "__Test Service Level",
"default_service_level_agreement": default_service_level_agreement,
"default_priority": "Medium",
"holiday_list": holiday_list,
"employee_group": employee_group,
"entity_type": entity_type,
"entity": entity,
"start_date": frappe.utils.getdate(),
"end_date": frappe.utils.add_to_date(frappe.utils.getdate(), days=100),
"priorities": [
{
"priority": "Low",
"response_time": response_time,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
},
{
"priority": "Medium",
"response_time": response_time,
"default_priority": 1,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
},
{
"priority": "High",
"response_time": response_time,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
}
],
"pause_sla_on": [
{
"status": "Replied"
}
],
"support_and_resolution": [
{
"workday": "Monday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Tuesday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Wednesday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Thursday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Friday",
"start_time": "10:00:00",
"end_time": "18:00:00",
}
]
})
service_level_agreement = frappe.get_doc(
{
"doctype": "Service Level Agreement",
"enable": 1,
"service_level": "__Test Service Level",
"default_service_level_agreement": default_service_level_agreement,
"default_priority": "Medium",
"holiday_list": holiday_list,
"employee_group": employee_group,
"entity_type": entity_type,
"entity": entity,
"start_date": frappe.utils.getdate(),
"end_date": frappe.utils.add_to_date(frappe.utils.getdate(), days=100),
"priorities": [
{
"priority": "Low",
"response_time": response_time,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
},
{
"priority": "Medium",
"response_time": response_time,
"default_priority": 1,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
},
{
"priority": "High",
"response_time": response_time,
"response_time_period": "Hour",
"resolution_time": resolution_time,
"resolution_time_period": "Hour",
},
],
"pause_sla_on": [{"status": "Replied"}],
"support_and_resolution": [
{
"workday": "Monday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Tuesday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Wednesday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Thursday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
{
"workday": "Friday",
"start_time": "10:00:00",
"end_time": "18:00:00",
},
],
}
)
filters = {
"default_service_level_agreement": service_level_agreement.default_service_level_agreement,
"service_level": service_level_agreement.service_level
"service_level": service_level_agreement.service_level,
}
if not default_service_level_agreement:
filters.update({
"entity_type": entity_type,
"entity": entity
})
filters.update({"entity_type": entity_type, "entity": entity})
service_level_agreement_exists = frappe.db.exists("Service Level Agreement", filters)
@@ -171,24 +250,26 @@ def create_service_level_agreement(default_service_level_agreement, holiday_list
def create_customer():
customer = frappe.get_doc({
"doctype": "Customer",
"customer_name": "_Test Customer",
"customer_group": "Commercial",
"customer_type": "Individual",
"territory": "Rest Of The World"
})
customer = frappe.get_doc(
{
"doctype": "Customer",
"customer_name": "_Test Customer",
"customer_group": "Commercial",
"customer_type": "Individual",
"territory": "Rest Of The World",
}
)
if not frappe.db.exists("Customer", "_Test Customer"):
customer.insert(ignore_permissions=True)
return customer.name
else:
return frappe.db.exists("Customer", "_Test Customer")
def create_customer_group():
customer_group = frappe.get_doc({
"doctype": "Customer Group",
"customer_group_name": "_Test SLA Customer Group"
})
customer_group = frappe.get_doc(
{"doctype": "Customer Group", "customer_group_name": "_Test SLA Customer Group"}
)
if not frappe.db.exists("Customer Group", {"customer_group_name": "_Test SLA Customer Group"}):
customer_group.insert()
@@ -196,11 +277,14 @@ def create_customer_group():
else:
return frappe.db.exists("Customer Group", {"customer_group_name": "_Test SLA Customer Group"})
def create_territory():
territory = frappe.get_doc({
"doctype": "Territory",
"territory_name": "_Test SLA Territory",
})
territory = frappe.get_doc(
{
"doctype": "Territory",
"territory_name": "_Test SLA Territory",
}
)
if not frappe.db.exists("Territory", {"territory_name": "_Test SLA Territory"}):
territory.insert()
@@ -208,42 +292,65 @@ def create_territory():
else:
return frappe.db.exists("Territory", {"territory_name": "_Test SLA Territory"})
def create_service_level_agreements_for_issues():
create_service_level_agreement(default_service_level_agreement=1, holiday_list="__Test Holiday List",
employee_group="_Test Employee Group", entity_type=None, entity=None, response_time=14400, resolution_time=21600)
create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
)
create_customer()
create_service_level_agreement(default_service_level_agreement=0, holiday_list="__Test Holiday List",
employee_group="_Test Employee Group", entity_type="Customer", entity="_Test Customer", response_time=7200, resolution_time=10800)
create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Customer",
entity="_Test Customer",
response_time=7200,
resolution_time=10800,
)
create_customer_group()
create_service_level_agreement(default_service_level_agreement=0, holiday_list="__Test Holiday List",
employee_group="_Test Employee Group", entity_type="Customer Group", entity="_Test SLA Customer Group", response_time=7200, resolution_time=10800)
create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Customer Group",
entity="_Test SLA Customer Group",
response_time=7200,
resolution_time=10800,
)
create_territory()
create_service_level_agreement(default_service_level_agreement=0, holiday_list="__Test Holiday List",
employee_group="_Test Employee Group", entity_type="Territory", entity="_Test SLA Territory", response_time=7200, resolution_time=10800)
create_service_level_agreement(
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
employee_group="_Test Employee Group",
entity_type="Territory",
entity="_Test SLA Territory",
response_time=7200,
resolution_time=10800,
)
def make_holiday_list():
holiday_list = frappe.db.exists("Holiday List", "__Test Holiday List")
if not holiday_list:
holiday_list = frappe.get_doc({
"doctype": "Holiday List",
"holiday_list_name": "__Test Holiday List",
"from_date": "2019-01-01",
"to_date": "2019-12-31",
"holidays": [
{
"description": "Test Holiday 1",
"holiday_date": "2019-03-05"
},
{
"description": "Test Holiday 2",
"holiday_date": "2019-03-07"
},
{
"description": "Test Holiday 3",
"holiday_date": "2019-02-11"
},
]
}).insert()
holiday_list = frappe.get_doc(
{
"doctype": "Holiday List",
"holiday_list_name": "__Test Holiday List",
"from_date": "2019-01-01",
"to_date": "2019-12-31",
"holidays": [
{"description": "Test Holiday 1", "holiday_date": "2019-03-05"},
{"description": "Test Holiday 2", "holiday_date": "2019-03-07"},
{"description": "Test Holiday 3", "holiday_date": "2019-02-11"},
],
}
).insert()

3
erpnext/support/doctype/warranty_claim/test_warranty_claim.py
View File

@@ -5,7 +5,8 @@ import unittest
import frappe
test_records = frappe.get_test_records('Warranty Claim')
test_records = frappe.get_test_records("Warranty Claim")
class TestWarrantyClaim(unittest.TestCase):
pass

45
erpnext/support/doctype/warranty_claim/warranty_claim.py
View File

@@ -2,7 +2,6 @@
# License: GNU General Public License v3. See license.txt
import frappe
from frappe import _, session
from frappe.utils import now_datetime
@@ -15,27 +14,33 @@ class WarrantyClaim(TransactionBase):
return _("{0}: From {1}").format(self.status, self.customer_name)
def validate(self):
if session['user'] != 'Guest' and not self.customer:
if session["user"] != "Guest" and not self.customer:
frappe.throw(_("Customer is required"))
if self.status=="Closed" and not self.resolution_date and \
frappe.db.get_value("Warranty Claim", self.name, "status")!="Closed":
if (
self.status == "Closed"
and not self.resolution_date
and frappe.db.get_value("Warranty Claim", self.name, "status") != "Closed"
):
self.resolution_date = now_datetime()
def on_cancel(self):
lst = frappe.db.sql("""select t1.name
lst = frappe.db.sql(
"""select t1.name
from `tabMaintenance Visit` t1, `tabMaintenance Visit Purpose` t2
where t2.parent = t1.name and t2.prevdoc_docname = %s and t1.docstatus!=2""",
(self.name))
(self.name),
)
if lst:
lst1 = ','.join(x[0] for x in lst)
lst1 = ",".join(x[0] for x in lst)
frappe.throw(_("Cancel Material Visit {0} before cancelling this Warranty Claim").format(lst1))
else:
frappe.db.set(self, 'status', 'Cancelled')
frappe.db.set(self, "status", "Cancelled")
def on_update(self):
pass
@frappe.whitelist()
def make_maintenance_visit(source_name, target_doc=None):
from frappe.model.mapper import get_mapped_doc, map_child_doc
@@ -44,25 +49,25 @@ def make_maintenance_visit(source_name, target_doc=None):
target_doc.prevdoc_doctype = source_parent.doctype
target_doc.prevdoc_docname = source_parent.name
visit = frappe.db.sql("""select t1.name
visit = frappe.db.sql(
"""select t1.name
from `tabMaintenance Visit` t1, `tabMaintenance Visit Purpose` t2
where t2.parent=t1.name and t2.prevdoc_docname=%s
and t1.docstatus=1 and t1.completion_status='Fully Completed'""", source_name)
and t1.docstatus=1 and t1.completion_status='Fully Completed'""",
source_name,
)
if not visit:
target_doc = get_mapped_doc("Warranty Claim", source_name, {
"Warranty Claim": {
"doctype": "Maintenance Visit",
"field_map": {}
}
}, target_doc)
target_doc = get_mapped_doc(
"Warranty Claim",
source_name,
{"Warranty Claim": {"doctype": "Maintenance Visit", "field_map": {}}},
target_doc,
)
source_doc = frappe.get_doc("Warranty Claim", source_name)
if source_doc.get("item_code"):
table_map = {
"doctype": "Maintenance Visit Purpose",
"postprocess": _update_links
}
table_map = {"doctype": "Maintenance Visit Purpose", "postprocess": _update_links}
map_child_doc(source_doc, target_doc, table_map, source_doc)
return target_doc

22
erpnext/support/report/first_response_time_for_issues/first_response_time_for_issues.py
View File

@@ -7,21 +7,17 @@ import frappe
def execute(filters=None):
columns = [
{"fieldname": "creation_date", "label": "Date", "fieldtype": "Date", "width": 300},
{
'fieldname': 'creation_date',
'label': 'Date',
'fieldtype': 'Date',
'width': 300
},
{
'fieldname': 'first_response_time',
'fieldtype': 'Duration',
'label': 'First Response Time',
'width': 300
"fieldname": "first_response_time",
"fieldtype": "Duration",
"label": "First Response Time",
"width": 300,
},
]
data = frappe.db.sql('''
data = frappe.db.sql(
"""
SELECT
date(creation) as creation_date,
avg(first_response_time) as avg_response_time
@@ -31,6 +27,8 @@ def execute(filters=None):
and first_response_time > 0
GROUP BY creation_date
ORDER BY creation_date desc
''', (filters.from_date, filters.to_date))
""",
(filters.from_date, filters.to_date),
)
return columns, data

177
erpnext/support/report/issue_analytics/issue_analytics.py
View File

@@ -15,6 +15,7 @@ from erpnext.accounts.utils import get_fiscal_year
def execute(filters=None):
return IssueAnalytics(filters).run()
class IssueAnalytics(object):
def __init__(self, filters=None):
"""Issue Analytics Report"""
@@ -31,101 +32,98 @@ class IssueAnalytics(object):
def get_columns(self):
self.columns = []
if self.filters.based_on == 'Customer':
self.columns.append({
'label': _('Customer'),
'options': 'Customer',
'fieldname': 'customer',
'fieldtype': 'Link',
'width': 200
})
if self.filters.based_on == "Customer":
self.columns.append(
{
"label": _("Customer"),
"options": "Customer",
"fieldname": "customer",
"fieldtype": "Link",
"width": 200,
}
)
elif self.filters.based_on == 'Assigned To':
self.columns.append({
'label': _('User'),
'fieldname': 'user',
'fieldtype': 'Link',
'options': 'User',
'width': 200
})
elif self.filters.based_on == "Assigned To":
self.columns.append(
{"label": _("User"), "fieldname": "user", "fieldtype": "Link", "options": "User", "width": 200}
)
elif self.filters.based_on == 'Issue Type':
self.columns.append({
'label': _('Issue Type'),
'fieldname': 'issue_type',
'fieldtype': 'Link',
'options': 'Issue Type',
'width': 200
})
elif self.filters.based_on == "Issue Type":
self.columns.append(
{
"label": _("Issue Type"),
"fieldname": "issue_type",
"fieldtype": "Link",
"options": "Issue Type",
"width": 200,
}
)
elif self.filters.based_on == 'Issue Priority':
self.columns.append({
'label': _('Issue Priority'),
'fieldname': 'priority',
'fieldtype': 'Link',
'options': 'Issue Priority',
'width': 200
})
elif self.filters.based_on == "Issue Priority":
self.columns.append(
{
"label": _("Issue Priority"),
"fieldname": "priority",
"fieldtype": "Link",
"options": "Issue Priority",
"width": 200,
}
)
for end_date in self.periodic_daterange:
period = self.get_period(end_date)
self.columns.append({
'label': _(period),
'fieldname': scrub(period),
'fieldtype': 'Int',
'width': 120
})
self.columns.append(
{"label": _(period), "fieldname": scrub(period), "fieldtype": "Int", "width": 120}
)
self.columns.append({
'label': _('Total'),
'fieldname': 'total',
'fieldtype': 'Int',
'width': 120
})
self.columns.append(
{"label": _("Total"), "fieldname": "total", "fieldtype": "Int", "width": 120}
)
def get_data(self):
self.get_issues()
self.get_rows()
def get_period(self, date):
months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
if self.filters.range == 'Weekly':
period = 'Week ' + str(date.isocalendar()[1])
elif self.filters.range == 'Monthly':
if self.filters.range == "Weekly":
period = "Week " + str(date.isocalendar()[1])
elif self.filters.range == "Monthly":
period = str(months[date.month - 1])
elif self.filters.range == 'Quarterly':
period = 'Quarter ' + str(((date.month - 1) // 3) + 1)
elif self.filters.range == "Quarterly":
period = "Quarter " + str(((date.month - 1) // 3) + 1)
else:
year = get_fiscal_year(date, self.filters.company)
period = str(year[0])
if getdate(self.filters.from_date).year != getdate(self.filters.to_date).year and self.filters.range != 'Yearly':
period += ' ' + str(date.year)
if (
getdate(self.filters.from_date).year != getdate(self.filters.to_date).year
and self.filters.range != "Yearly"
):
period += " " + str(date.year)
return period
def get_period_date_ranges(self):
from dateutil.relativedelta import MO, relativedelta
from_date, to_date = getdate(self.filters.from_date), getdate(self.filters.to_date)
increment = {
'Monthly': 1,
'Quarterly': 3,
'Half-Yearly': 6,
'Yearly': 12
}.get(self.filters.range, 1)
increment = {"Monthly": 1, "Quarterly": 3, "Half-Yearly": 6, "Yearly": 12}.get(
self.filters.range, 1
)
if self.filters.range in ['Monthly', 'Quarterly']:
if self.filters.range in ["Monthly", "Quarterly"]:
from_date = from_date.replace(day=1)
elif self.filters.range == 'Yearly':
elif self.filters.range == "Yearly":
from_date = get_fiscal_year(from_date)[1]
else:
from_date = from_date + relativedelta(from_date, weekday=MO(-1))
self.periodic_daterange = []
for dummy in range(1, 53):
if self.filters.range == 'Weekly':
if self.filters.range == "Weekly":
period_end_date = add_days(from_date, 6)
else:
period_end_date = add_to_date(from_date, months=increment, days=-1)
@@ -142,25 +140,26 @@ class IssueAnalytics(object):
def get_issues(self):
filters = self.get_common_filters()
self.field_map = {
'Customer': 'customer',
'Issue Type': 'issue_type',
'Issue Priority': 'priority',
'Assigned To': '_assign'
"Customer": "customer",
"Issue Type": "issue_type",
"Issue Priority": "priority",
"Assigned To": "_assign",
}
self.entries = frappe.db.get_all('Issue',
fields=[self.field_map.get(self.filters.based_on), 'name', 'opening_date'],
filters=filters
self.entries = frappe.db.get_all(
"Issue",
fields=[self.field_map.get(self.filters.based_on), "name", "opening_date"],
filters=filters,
)
def get_common_filters(self):
filters = {}
filters['opening_date'] = ('between', [self.filters.from_date, self.filters.to_date])
filters["opening_date"] = ("between", [self.filters.from_date, self.filters.to_date])
if self.filters.get('assigned_to'):
filters['_assign'] = ('like', '%' + self.filters.get('assigned_to') + '%')
if self.filters.get("assigned_to"):
filters["_assign"] = ("like", "%" + self.filters.get("assigned_to") + "%")
for entry in ['company', 'status', 'priority', 'customer', 'project']:
for entry in ["company", "status", "priority", "customer", "project"]:
if self.filters.get(entry):
filters[entry] = self.filters.get(entry)
@@ -171,14 +170,14 @@ class IssueAnalytics(object):
self.get_periodic_data()
for entity, period_data in iteritems(self.issue_periodic_data):
if self.filters.based_on == 'Customer':
row = {'customer': entity}
elif self.filters.based_on == 'Assigned To':
row = {'user': entity}
elif self.filters.based_on == 'Issue Type':
row = {'issue_type': entity}
elif self.filters.based_on == 'Issue Priority':
row = {'priority': entity}
if self.filters.based_on == "Customer":
row = {"customer": entity}
elif self.filters.based_on == "Assigned To":
row = {"user": entity}
elif self.filters.based_on == "Issue Type":
row = {"issue_type": entity}
elif self.filters.based_on == "Issue Priority":
row = {"priority": entity}
total = 0
for end_date in self.periodic_daterange:
@@ -187,7 +186,7 @@ class IssueAnalytics(object):
row[scrub(period)] = amount
total += amount
row['total'] = total
row["total"] = total
self.data.append(row)
@@ -195,9 +194,9 @@ class IssueAnalytics(object):
self.issue_periodic_data = frappe._dict()
for d in self.entries:
period = self.get_period(d.get('opening_date'))
period = self.get_period(d.get("opening_date"))
if self.filters.based_on == 'Assigned To':
if self.filters.based_on == "Assigned To":
if d._assign:
for entry in json.loads(d._assign):
self.issue_periodic_data.setdefault(entry, frappe._dict()).setdefault(period, 0.0)
@@ -207,18 +206,12 @@ class IssueAnalytics(object):
field = self.field_map.get(self.filters.based_on)
value = d.get(field)
if not value:
value = _('Not Specified')
value = _("Not Specified")
self.issue_periodic_data.setdefault(value, frappe._dict()).setdefault(period, 0.0)
self.issue_periodic_data[value][period] += 1
def get_chart_data(self):
length = len(self.columns)
labels = [d.get('label') for d in self.columns[1:length-1]]
self.chart = {
'data': {
'labels': labels,
'datasets': []
},
'type': 'line'
}
labels = [d.get("label") for d in self.columns[1 : length - 1]]
self.chart = {"data": {"labels": labels, "datasets": []}, "type": "line"}

176
erpnext/support/report/issue_analytics/test_issue_analytics.py
View File

@@ -1,4 +1,3 @@
import unittest
import frappe
@@ -11,7 +10,8 @@ from erpnext.support.doctype.service_level_agreement.test_service_level_agreemen
)
from erpnext.support.report.issue_analytics.issue_analytics import execute
months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']
months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
class TestIssueAnalytics(unittest.TestCase):
@classmethod
@@ -24,8 +24,8 @@ class TestIssueAnalytics(unittest.TestCase):
self.current_month = str(months[current_month_date.month - 1]).lower()
self.last_month = str(months[last_month_date.month - 1]).lower()
if current_month_date.year != last_month_date.year:
self.current_month += '_' + str(current_month_date.year)
self.last_month += '_' + str(last_month_date.year)
self.current_month += "_" + str(current_month_date.year)
self.last_month += "_" + str(last_month_date.year)
def test_issue_analytics(self):
create_service_level_agreements_for_issues()
@@ -39,146 +39,88 @@ class TestIssueAnalytics(unittest.TestCase):
def compare_result_for_customer(self):
filters = {
'company': '_Test Company',
'based_on': 'Customer',
'from_date': add_months(getdate(), -1),
'to_date': getdate(),
'range': 'Monthly'
"company": "_Test Company",
"based_on": "Customer",
"from_date": add_months(getdate(), -1),
"to_date": getdate(),
"range": "Monthly",
}
report = execute(filters)
expected_data = [
{
'customer': '__Test Customer 2',
self.last_month: 1.0,
self.current_month: 0.0,
'total': 1.0
},
{
'customer': '__Test Customer 1',
self.last_month: 0.0,
self.current_month: 1.0,
'total': 1.0
},
{
'customer': '__Test Customer',
self.last_month: 1.0,
self.current_month: 1.0,
'total': 2.0
}
{"customer": "__Test Customer 2", self.last_month: 1.0, self.current_month: 0.0, "total": 1.0},
{"customer": "__Test Customer 1", self.last_month: 0.0, self.current_month: 1.0, "total": 1.0},
{"customer": "__Test Customer", self.last_month: 1.0, self.current_month: 1.0, "total": 2.0},
]
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
def compare_result_for_issue_type(self):
filters = {
'company': '_Test Company',
'based_on': 'Issue Type',
'from_date': add_months(getdate(), -1),
'to_date': getdate(),
'range': 'Monthly'
"company": "_Test Company",
"based_on": "Issue Type",
"from_date": add_months(getdate(), -1),
"to_date": getdate(),
"range": "Monthly",
}
report = execute(filters)
expected_data = [
{
'issue_type': 'Discomfort',
self.last_month: 1.0,
self.current_month: 0.0,
'total': 1.0
},
{
'issue_type': 'Service Request',
self.last_month: 0.0,
self.current_month: 1.0,
'total': 1.0
},
{
'issue_type': 'Bug',
self.last_month: 1.0,
self.current_month: 1.0,
'total': 2.0
}
{"issue_type": "Discomfort", self.last_month: 1.0, self.current_month: 0.0, "total": 1.0},
{"issue_type": "Service Request", self.last_month: 0.0, self.current_month: 1.0, "total": 1.0},
{"issue_type": "Bug", self.last_month: 1.0, self.current_month: 1.0, "total": 2.0},
]
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
def compare_result_for_issue_priority(self):
filters = {
'company': '_Test Company',
'based_on': 'Issue Priority',
'from_date': add_months(getdate(), -1),
'to_date': getdate(),
'range': 'Monthly'
"company": "_Test Company",
"based_on": "Issue Priority",
"from_date": add_months(getdate(), -1),
"to_date": getdate(),
"range": "Monthly",
}
report = execute(filters)
expected_data = [
{
'priority': 'Medium',
self.last_month: 1.0,
self.current_month: 1.0,
'total': 2.0
},
{
'priority': 'Low',
self.last_month: 1.0,
self.current_month: 0.0,
'total': 1.0
},
{
'priority': 'High',
self.last_month: 0.0,
self.current_month: 1.0,
'total': 1.0
}
{"priority": "Medium", self.last_month: 1.0, self.current_month: 1.0, "total": 2.0},
{"priority": "Low", self.last_month: 1.0, self.current_month: 0.0, "total": 1.0},
{"priority": "High", self.last_month: 0.0, self.current_month: 1.0, "total": 1.0},
]
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
def compare_result_for_assignment(self):
filters = {
'company': '_Test Company',
'based_on': 'Assigned To',
'from_date': add_months(getdate(), -1),
'to_date': getdate(),
'range': 'Monthly'
"company": "_Test Company",
"based_on": "Assigned To",
"from_date": add_months(getdate(), -1),
"to_date": getdate(),
"range": "Monthly",
}
report = execute(filters)
expected_data = [
{
'user': 'test@example.com',
self.last_month: 1.0,
self.current_month: 1.0,
'total': 2.0
},
{
'user': 'test1@example.com',
self.last_month: 2.0,
self.current_month: 1.0,
'total': 3.0
}
{"user": "test@example.com", self.last_month: 1.0, self.current_month: 1.0, "total": 2.0},
{"user": "test1@example.com", self.last_month: 2.0, self.current_month: 1.0, "total": 3.0},
]
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
self.assertEqual(expected_data, report[1]) # rows
self.assertEqual(len(report[0]), 4) # cols
def create_issue_types():
for entry in ['Bug', 'Service Request', 'Discomfort']:
if not frappe.db.exists('Issue Type', entry):
frappe.get_doc({
'doctype': 'Issue Type',
'__newname': entry
}).insert()
for entry in ["Bug", "Service Request", "Discomfort"]:
if not frappe.db.exists("Issue Type", entry):
frappe.get_doc({"doctype": "Issue Type", "__newname": entry}).insert()
def create_records():
@@ -190,29 +132,15 @@ def create_records():
last_month_date = add_months(current_month_date, -1)
issue = make_issue(current_month_date, "__Test Customer", 2, "High", "Bug")
add_assignment({
"assign_to": ["test@example.com"],
"doctype": "Issue",
"name": issue.name
})
add_assignment({"assign_to": ["test@example.com"], "doctype": "Issue", "name": issue.name})
issue = make_issue(last_month_date, "__Test Customer", 2, "Low", "Bug")
add_assignment({
"assign_to": ["test1@example.com"],
"doctype": "Issue",
"name": issue.name
})
add_assignment({"assign_to": ["test1@example.com"], "doctype": "Issue", "name": issue.name})
issue = make_issue(current_month_date, "__Test Customer 1", 2, "Medium", "Service Request")
add_assignment({
"assign_to": ["test1@example.com"],
"doctype": "Issue",
"name": issue.name
})
add_assignment({"assign_to": ["test1@example.com"], "doctype": "Issue", "name": issue.name})
issue = make_issue(last_month_date, "__Test Customer 2", 2, "Medium", "Discomfort")
add_assignment({
"assign_to": ["test@example.com", "test1@example.com"],
"doctype": "Issue",
"name": issue.name
})
add_assignment(
{"assign_to": ["test@example.com", "test1@example.com"], "doctype": "Issue", "name": issue.name}
)

341
erpnext/support/report/issue_summary/issue_summary.py
View File

@@ -13,6 +13,7 @@ from six import iteritems
def execute(filters=None):
return IssueSummary(filters).run()
class IssueSummary(object):
def __init__(self, filters=None):
self.filters = frappe._dict(filters or {})
@@ -28,82 +29,77 @@ class IssueSummary(object):
def get_columns(self):
self.columns = []
if self.filters.based_on == 'Customer':
self.columns.append({
'label': _('Customer'),
'options': 'Customer',
'fieldname': 'customer',
'fieldtype': 'Link',
'width': 200
})
if self.filters.based_on == "Customer":
self.columns.append(
{
"label": _("Customer"),
"options": "Customer",
"fieldname": "customer",
"fieldtype": "Link",
"width": 200,
}
)
elif self.filters.based_on == 'Assigned To':
self.columns.append({
'label': _('User'),
'fieldname': 'user',
'fieldtype': 'Link',
'options': 'User',
'width': 200
})
elif self.filters.based_on == "Assigned To":
self.columns.append(
{"label": _("User"), "fieldname": "user", "fieldtype": "Link", "options": "User", "width": 200}
)
elif self.filters.based_on == 'Issue Type':
self.columns.append({
'label': _('Issue Type'),
'fieldname': 'issue_type',
'fieldtype': 'Link',
'options': 'Issue Type',
'width': 200
})
elif self.filters.based_on == "Issue Type":
self.columns.append(
{
"label": _("Issue Type"),
"fieldname": "issue_type",
"fieldtype": "Link",
"options": "Issue Type",
"width": 200,
}
)
elif self.filters.based_on == 'Issue Priority':
self.columns.append({
'label': _('Issue Priority'),
'fieldname': 'priority',
'fieldtype': 'Link',
'options': 'Issue Priority',
'width': 200
})
elif self.filters.based_on == "Issue Priority":
self.columns.append(
{
"label": _("Issue Priority"),
"fieldname": "priority",
"fieldtype": "Link",
"options": "Issue Priority",
"width": 200,
}
)
self.statuses = ['Open', 'Replied', 'On Hold', 'Resolved', 'Closed']
self.statuses = ["Open", "Replied", "On Hold", "Resolved", "Closed"]
for status in self.statuses:
self.columns.append({
'label': _(status),
'fieldname': scrub(status),
'fieldtype': 'Int',
'width': 80
})
self.columns.append(
{"label": _(status), "fieldname": scrub(status), "fieldtype": "Int", "width": 80}
)
self.columns.append({
'label': _('Total Issues'),
'fieldname': 'total_issues',
'fieldtype': 'Int',
'width': 100
})
self.columns.append(
{"label": _("Total Issues"), "fieldname": "total_issues", "fieldtype": "Int", "width": 100}
)
self.sla_status_map = {
'SLA Failed': 'failed',
'SLA Fulfilled': 'fulfilled',
'SLA Ongoing': 'ongoing'
"SLA Failed": "failed",
"SLA Fulfilled": "fulfilled",
"SLA Ongoing": "ongoing",
}
for label, fieldname in self.sla_status_map.items():
self.columns.append({
'label': _(label),
'fieldname': fieldname,
'fieldtype': 'Int',
'width': 100
})
self.columns.append(
{"label": _(label), "fieldname": fieldname, "fieldtype": "Int", "width": 100}
)
self.metrics = ['Avg First Response Time', 'Avg Response Time', 'Avg Hold Time',
'Avg Resolution Time', 'Avg User Resolution Time']
self.metrics = [
"Avg First Response Time",
"Avg Response Time",
"Avg Hold Time",
"Avg Resolution Time",
"Avg User Resolution Time",
]
for metric in self.metrics:
self.columns.append({
'label': _(metric),
'fieldname': scrub(metric),
'fieldtype': 'Duration',
'width': 170
})
self.columns.append(
{"label": _(metric), "fieldname": scrub(metric), "fieldtype": "Duration", "width": 170}
)
def get_data(self):
self.get_issues()
@@ -112,26 +108,37 @@ class IssueSummary(object):
def get_issues(self):
filters = self.get_common_filters()
self.field_map = {
'Customer': 'customer',
'Issue Type': 'issue_type',
'Issue Priority': 'priority',
'Assigned To': '_assign'
"Customer": "customer",
"Issue Type": "issue_type",
"Issue Priority": "priority",
"Assigned To": "_assign",
}
self.entries = frappe.db.get_all('Issue',
fields=[self.field_map.get(self.filters.based_on), 'name', 'opening_date', 'status', 'avg_response_time',
'first_response_time', 'total_hold_time', 'user_resolution_time', 'resolution_time', 'agreement_status'],
filters=filters
self.entries = frappe.db.get_all(
"Issue",
fields=[
self.field_map.get(self.filters.based_on),
"name",
"opening_date",
"status",
"avg_response_time",
"first_response_time",
"total_hold_time",
"user_resolution_time",
"resolution_time",
"agreement_status",
],
filters=filters,
)
def get_common_filters(self):
filters = {}
filters['opening_date'] = ('between', [self.filters.from_date, self.filters.to_date])
filters["opening_date"] = ("between", [self.filters.from_date, self.filters.to_date])
if self.filters.get('assigned_to'):
filters['_assign'] = ('like', '%' + self.filters.get('assigned_to') + '%')
if self.filters.get("assigned_to"):
filters["_assign"] = ("like", "%" + self.filters.get("assigned_to") + "%")
for entry in ['company', 'status', 'priority', 'customer', 'project']:
for entry in ["company", "status", "priority", "customer", "project"]:
if self.filters.get(entry):
filters[entry] = self.filters.get(entry)
@@ -142,20 +149,20 @@ class IssueSummary(object):
self.get_summary_data()
for entity, data in iteritems(self.issue_summary_data):
if self.filters.based_on == 'Customer':
row = {'customer': entity}
elif self.filters.based_on == 'Assigned To':
row = {'user': entity}
elif self.filters.based_on == 'Issue Type':
row = {'issue_type': entity}
elif self.filters.based_on == 'Issue Priority':
row = {'priority': entity}
if self.filters.based_on == "Customer":
row = {"customer": entity}
elif self.filters.based_on == "Assigned To":
row = {"user": entity}
elif self.filters.based_on == "Issue Type":
row = {"issue_type": entity}
elif self.filters.based_on == "Issue Priority":
row = {"priority": entity}
for status in self.statuses:
count = flt(data.get(status, 0.0))
row[scrub(status)] = count
row['total_issues'] = data.get('total_issues', 0.0)
row["total_issues"] = data.get("total_issues", 0.0)
for sla_status in self.sla_status_map.values():
value = flt(data.get(sla_status), 0.0)
@@ -174,36 +181,41 @@ class IssueSummary(object):
status = d.status
agreement_status = scrub(d.agreement_status)
if self.filters.based_on == 'Assigned To':
if self.filters.based_on == "Assigned To":
if d._assign:
for entry in json.loads(d._assign):
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault(status, 0.0)
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault(agreement_status, 0.0)
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault('total_issues', 0.0)
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault("total_issues", 0.0)
self.issue_summary_data[entry][status] += 1
self.issue_summary_data[entry][agreement_status] += 1
self.issue_summary_data[entry]['total_issues'] += 1
self.issue_summary_data[entry]["total_issues"] += 1
else:
field = self.field_map.get(self.filters.based_on)
value = d.get(field)
if not value:
value = _('Not Specified')
value = _("Not Specified")
self.issue_summary_data.setdefault(value, frappe._dict()).setdefault(status, 0.0)
self.issue_summary_data.setdefault(value, frappe._dict()).setdefault(agreement_status, 0.0)
self.issue_summary_data.setdefault(value, frappe._dict()).setdefault('total_issues', 0.0)
self.issue_summary_data.setdefault(value, frappe._dict()).setdefault("total_issues", 0.0)
self.issue_summary_data[value][status] += 1
self.issue_summary_data[value][agreement_status] += 1
self.issue_summary_data[value]['total_issues'] += 1
self.issue_summary_data[value]["total_issues"] += 1
self.get_metrics_data()
def get_metrics_data(self):
issues = []
metrics_list = ['avg_response_time', 'avg_first_response_time', 'avg_hold_time',
'avg_resolution_time', 'avg_user_resolution_time']
metrics_list = [
"avg_response_time",
"avg_first_response_time",
"avg_hold_time",
"avg_resolution_time",
"avg_user_resolution_time",
]
for entry in self.entries:
issues.append(entry.name)
@@ -211,7 +223,7 @@ class IssueSummary(object):
field = self.field_map.get(self.filters.based_on)
if issues:
if self.filters.based_on == 'Assigned To':
if self.filters.based_on == "Assigned To":
assignment_map = frappe._dict()
for d in self.entries:
if d._assign:
@@ -219,11 +231,15 @@ class IssueSummary(object):
for metric in metrics_list:
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault(metric, 0.0)
self.issue_summary_data[entry]['avg_response_time'] += d.get('avg_response_time') or 0.0
self.issue_summary_data[entry]['avg_first_response_time'] += d.get('first_response_time') or 0.0
self.issue_summary_data[entry]['avg_hold_time'] += d.get('total_hold_time') or 0.0
self.issue_summary_data[entry]['avg_resolution_time'] += d.get('resolution_time') or 0.0
self.issue_summary_data[entry]['avg_user_resolution_time'] += d.get('user_resolution_time') or 0.0
self.issue_summary_data[entry]["avg_response_time"] += d.get("avg_response_time") or 0.0
self.issue_summary_data[entry]["avg_first_response_time"] += (
d.get("first_response_time") or 0.0
)
self.issue_summary_data[entry]["avg_hold_time"] += d.get("total_hold_time") or 0.0
self.issue_summary_data[entry]["avg_resolution_time"] += d.get("resolution_time") or 0.0
self.issue_summary_data[entry]["avg_user_resolution_time"] += (
d.get("user_resolution_time") or 0.0
)
if not assignment_map.get(entry):
assignment_map[entry] = 0
@@ -234,7 +250,8 @@ class IssueSummary(object):
self.issue_summary_data[entry][metric] /= flt(assignment_map.get(entry))
else:
data = frappe.db.sql("""
data = frappe.db.sql(
"""
SELECT
{0}, AVG(first_response_time) as avg_frt,
AVG(avg_response_time) as avg_resp_time,
@@ -245,21 +262,30 @@ class IssueSummary(object):
WHERE
name IN %(issues)s
GROUP BY {0}
""".format(field), {'issues': issues}, as_dict=1)
""".format(
field
),
{"issues": issues},
as_dict=1,
)
for entry in data:
value = entry.get(field)
if not value:
value = _('Not Specified')
value = _("Not Specified")
for metric in metrics_list:
self.issue_summary_data.setdefault(value, frappe._dict()).setdefault(metric, 0.0)
self.issue_summary_data[value]['avg_response_time'] = entry.get('avg_resp_time') or 0.0
self.issue_summary_data[value]['avg_first_response_time'] = entry.get('avg_frt') or 0.0
self.issue_summary_data[value]['avg_hold_time'] = entry.get('avg_hold_time') or 0.0
self.issue_summary_data[value]['avg_resolution_time'] = entry.get('avg_resolution_time') or 0.0
self.issue_summary_data[value]['avg_user_resolution_time'] = entry.get('avg_user_resolution_time') or 0.0
self.issue_summary_data[value]["avg_response_time"] = entry.get("avg_resp_time") or 0.0
self.issue_summary_data[value]["avg_first_response_time"] = entry.get("avg_frt") or 0.0
self.issue_summary_data[value]["avg_hold_time"] = entry.get("avg_hold_time") or 0.0
self.issue_summary_data[value]["avg_resolution_time"] = (
entry.get("avg_resolution_time") or 0.0
)
self.issue_summary_data[value]["avg_user_resolution_time"] = (
entry.get("avg_user_resolution_time") or 0.0
)
def get_chart_data(self):
self.chart = []
@@ -273,47 +299,30 @@ class IssueSummary(object):
entity = self.filters.based_on
entity_field = self.field_map.get(entity)
if entity == 'Assigned To':
entity_field = 'user'
if entity == "Assigned To":
entity_field = "user"
for entry in self.data:
labels.append(entry.get(entity_field))
open_issues.append(entry.get('open'))
replied_issues.append(entry.get('replied'))
on_hold_issues.append(entry.get('on_hold'))
resolved_issues.append(entry.get('resolved'))
closed_issues.append(entry.get('closed'))
open_issues.append(entry.get("open"))
replied_issues.append(entry.get("replied"))
on_hold_issues.append(entry.get("on_hold"))
resolved_issues.append(entry.get("resolved"))
closed_issues.append(entry.get("closed"))
self.chart = {
'data': {
'labels': labels[:30],
'datasets': [
{
'name': 'Open',
'values': open_issues[:30]
},
{
'name': 'Replied',
'values': replied_issues[:30]
},
{
'name': 'On Hold',
'values': on_hold_issues[:30]
},
{
'name': 'Resolved',
'values': resolved_issues[:30]
},
{
'name': 'Closed',
'values': closed_issues[:30]
}
]
"data": {
"labels": labels[:30],
"datasets": [
{"name": "Open", "values": open_issues[:30]},
{"name": "Replied", "values": replied_issues[:30]},
{"name": "On Hold", "values": on_hold_issues[:30]},
{"name": "Resolved", "values": resolved_issues[:30]},
{"name": "Closed", "values": closed_issues[:30]},
],
},
'type': 'bar',
'barOptions': {
'stacked': True
}
"type": "bar",
"barOptions": {"stacked": True},
}
def get_report_summary(self):
@@ -326,41 +335,41 @@ class IssueSummary(object):
closed = 0
for entry in self.data:
open_issues += entry.get('open')
replied += entry.get('replied')
on_hold += entry.get('on_hold')
resolved += entry.get('resolved')
closed += entry.get('closed')
open_issues += entry.get("open")
replied += entry.get("replied")
on_hold += entry.get("on_hold")
resolved += entry.get("resolved")
closed += entry.get("closed")
self.report_summary = [
{
'value': open_issues,
'indicator': 'Red',
'label': _('Open'),
'datatype': 'Int',
"value": open_issues,
"indicator": "Red",
"label": _("Open"),
"datatype": "Int",
},
{
'value': replied,
'indicator': 'Grey',
'label': _('Replied'),
'datatype': 'Int',
"value": replied,
"indicator": "Grey",
"label": _("Replied"),
"datatype": "Int",
},
{
'value': on_hold,
'indicator': 'Grey',
'label': _('On Hold'),
'datatype': 'Int',
"value": on_hold,
"indicator": "Grey",
"label": _("On Hold"),
"datatype": "Int",
},
{
'value': resolved,
'indicator': 'Green',
'label': _('Resolved'),
'datatype': 'Int',
"value": resolved,
"indicator": "Green",
"label": _("Resolved"),
"datatype": "Int",
},
{
'value': closed,
'indicator': 'Green',
'label': _('Closed'),
'datatype': 'Int',
}
"value": closed,
"indicator": "Green",
"label": _("Closed"),
"datatype": "Int",
},
]

94
erpnext/support/report/support_hour_distribution/support_hour_distribution.py
View File

@@ -8,34 +8,36 @@ from frappe.utils import add_to_date, get_datetime, getdate
from six import iteritems
time_slots = {
'12AM - 3AM': '00:00:00-03:00:00',
'3AM - 6AM': '03:00:00-06:00:00',
'6AM - 9AM': '06:00:00-09:00:00',
'9AM - 12PM': '09:00:00-12:00:00',
'12PM - 3PM': '12:00:00-15:00:00',
'3PM - 6PM': '15:00:00-18:00:00',
'6PM - 9PM': '18:00:00-21:00:00',
'9PM - 12AM': '21:00:00-23:00:00'
"12AM - 3AM": "00:00:00-03:00:00",
"3AM - 6AM": "03:00:00-06:00:00",
"6AM - 9AM": "06:00:00-09:00:00",
"9AM - 12PM": "09:00:00-12:00:00",
"12PM - 3PM": "12:00:00-15:00:00",
"3PM - 6PM": "15:00:00-18:00:00",
"6PM - 9PM": "18:00:00-21:00:00",
"9PM - 12AM": "21:00:00-23:00:00",
}
def execute(filters=None):
columns, data = [], []
if not filters.get('periodicity'):
filters['periodicity'] = 'Daily'
if not filters.get("periodicity"):
filters["periodicity"] = "Daily"
columns = get_columns()
data, timeslot_wise_count = get_data(filters)
chart = get_chart_data(timeslot_wise_count)
return columns, data, None, chart
def get_data(filters):
start_date = getdate(filters.from_date)
data = []
time_slot_wise_total_count = {}
while(start_date <= getdate(filters.to_date)):
hours_count = {'date': start_date}
while start_date <= getdate(filters.to_date):
hours_count = {"date": start_date}
for key, value in iteritems(time_slots):
start_time, end_time = value.split('-')
start_time, end_time = value.split("-")
start_time = get_datetime("{0} {1}".format(start_date.strftime("%Y-%m-%d"), start_time))
end_time = get_datetime("{0} {1}".format(start_date.strftime("%Y-%m-%d"), end_time))
hours_count[key] = get_hours_count(start_time, end_time)
@@ -48,49 +50,57 @@ def get_data(filters):
return data, time_slot_wise_total_count
def get_hours_count(start_time, end_time):
data = frappe.db.sql(""" select count(*) from `tabIssue` where creation
between %(start_time)s and %(end_time)s""", {
'start_time': start_time,
'end_time': end_time
}, as_list=1) or []
data = (
frappe.db.sql(
""" select count(*) from `tabIssue` where creation
between %(start_time)s and %(end_time)s""",
{"start_time": start_time, "end_time": end_time},
as_list=1,
)
or []
)
return data[0][0] if len(data) > 0 else 0
def get_columns():
columns = [{
"fieldname": "date",
"label": _("Date"),
"fieldtype": "Date",
"width": 100
}]
for label in ['12AM - 3AM', '3AM - 6AM', '6AM - 9AM',
'9AM - 12PM', '12PM - 3PM', '3PM - 6PM', '6PM - 9PM', '9PM - 12AM']:
columns.append({
"fieldname": label,
"label": _(label),
"fieldtype": "Data",
"width": 120
})
def get_columns():
columns = [{"fieldname": "date", "label": _("Date"), "fieldtype": "Date", "width": 100}]
for label in [
"12AM - 3AM",
"3AM - 6AM",
"6AM - 9AM",
"9AM - 12PM",
"12PM - 3PM",
"3PM - 6PM",
"6PM - 9PM",
"9PM - 12AM",
]:
columns.append({"fieldname": label, "label": _(label), "fieldtype": "Data", "width": 120})
return columns
def get_chart_data(timeslot_wise_count):
total_count = []
timeslots = ['12AM - 3AM', '3AM - 6AM', '6AM - 9AM',
'9AM - 12PM', '12PM - 3PM', '3PM - 6PM', '6PM - 9PM', '9PM - 12AM']
timeslots = [
"12AM - 3AM",
"3AM - 6AM",
"6AM - 9AM",
"9AM - 12PM",
"12PM - 3PM",
"3PM - 6PM",
"6PM - 9PM",
"9PM - 12AM",
]
datasets = []
for data in timeslots:
total_count.append(timeslot_wise_count.get(data, 0))
datasets.append({'values': total_count})
datasets.append({"values": total_count})
chart = {
"data": {
'labels': timeslots,
'datasets': datasets
}
}
chart = {"data": {"labels": timeslots, "datasets": datasets}}
chart["type"] = "line"
return chart

2
erpnext/support/web_form/issues/issues.py
View File

@@ -1,5 +1,3 @@
def get_context(context):
# do your magic here
pass