feat(SLA): Apply SLA to any document (#22449)

This commit is contained in:
Himanshu 2021-06-14 19:05:52 +05:30 committed by GitHub
parent 326a2e6a99
commit ec25d5938b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
16 changed files with 1358 additions and 603 deletions

View File

@ -228,6 +228,7 @@ standard_queries = {
doc_events = {
"*": {
"validate": "erpnext.support.doctype.service_level_agreement.service_level_agreement.apply",
"on_submit": "erpnext.healthcare.doctype.patient_history_settings.patient_history_settings.create_medical_record",
"on_update_after_submit": "erpnext.healthcare.doctype.patient_history_settings.patient_history_settings.update_medical_record",
"on_cancel": "erpnext.healthcare.doctype.patient_history_settings.patient_history_settings.delete_medical_record"
@ -242,6 +243,9 @@ doc_events = {
"on_update": ["erpnext.hr.doctype.employee.employee.update_user_permissions",
"erpnext.portal.utils.set_default_role"]
},
"Communication": {
"on_update": "erpnext.support.doctype.service_level_agreement.service_level_agreement.update_hold_time"
},
("Sales Taxes and Charges Template", 'Price List'): {
"on_update": "erpnext.shopping_cart.doctype.shopping_cart_settings.shopping_cart_settings.validate_cart_settings"
},
@ -332,8 +336,8 @@ scheduler_events = {
"erpnext.projects.doctype.project.project.hourly_reminder",
"erpnext.projects.doctype.project.project.collect_project_status",
"erpnext.hr.doctype.shift_type.shift_type.process_auto_attendance_for_all_shifts",
"erpnext.support.doctype.issue.issue.set_service_level_agreement_variance",
"erpnext.erpnext_integrations.connectors.shopify_connection.sync_old_orders"
"erpnext.erpnext_integrations.connectors.shopify_connection.sync_old_orders",
"erpnext.support.doctype.service_level_agreement.service_level_agreement.set_service_level_agreement_variance"
],
"hourly_long": [
"erpnext.stock.doctype.repost_item_valuation.repost_item_valuation.repost_entries"

View File

@ -285,4 +285,5 @@ erpnext.patches.v13_0.germany_make_custom_fields
erpnext.patches.v13_0.germany_fill_debtor_creditor_number
erpnext.patches.v13_0.set_pos_closing_as_failed
erpnext.patches.v13_0.update_timesheet_changes
erpnext.patches.v13_0.add_doctype_to_sla
erpnext.patches.v13_0.set_training_event_attendance

View File

@ -0,0 +1,20 @@
# Copyright (c) 2020, Frappe and Contributors
# License: GNU General Public License v3. See license.txt
from __future__ import unicode_literals
import frappe
from frappe.model.utils.rename_field import rename_field
def execute():
frappe.reload_doc('support', 'doctype', 'service_level_agreement')
if frappe.db.has_column('Service Level Agreement', 'enable'):
rename_field('Service Level Agreement', 'enable', 'enabled')
for sla in frappe.get_all('Service Level Agreement'):
agreement = frappe.get_doc('Service Level Agreement', sla.name)
agreement.document_type = 'Issue'
agreement.apply_sla_for_resolution = 1
agreement.append('sla_fulfilled_on', {'status': 'Resolved'})
agreement.append('sla_fulfilled_on', {'status': 'Closed'})
agreement.save()

View File

@ -749,6 +749,151 @@ $(document).on('app_ready', function() {
}
});
// Show SLA dashboard
$(document).on('app_ready', function() {
frappe.call({
method: 'erpnext.support.doctype.service_level_agreement.service_level_agreement.get_sla_doctypes',
callback: function(r) {
if (!r.message)
return;
$.each(r.message, function(_i, d) {
frappe.ui.form.on(d, {
onload: function(frm) {
if (!frm.doc.service_level_agreement)
return;
frappe.call({
method: 'erpnext.support.doctype.service_level_agreement.service_level_agreement.get_service_level_agreement_filters',
args: {
doctype: frm.doc.doctype,
name: frm.doc.service_level_agreement,
customer: frm.doc.customer
},
callback: function (r) {
if (r && r.message) {
frm.set_query('priority', function() {
return {
filters: {
'name': ['in', r.message.priority],
}
};
});
frm.set_query('service_level_agreement', function() {
return {
filters: {
'name': ['in', r.message.service_level_agreements],
}
};
});
}
}
});
},
refresh: function(frm) {
if (frm.doc.status !== 'Closed' && frm.doc.service_level_agreement
&& frm.doc.agreement_status === 'Ongoing') {
frappe.call({
'method': 'frappe.client.get',
args: {
doctype: 'Service Level Agreement',
name: frm.doc.service_level_agreement
},
callback: function(data) {
let statuses = data.message.pause_sla_on;
const hold_statuses = [];
$.each(statuses, (_i, entry) => {
hold_statuses.push(entry.status);
});
if (hold_statuses.includes(frm.doc.status)) {
frm.dashboard.clear_headline();
let message = {'indicator': 'orange', 'msg': __('SLA is on hold since {0}', [moment(frm.doc.on_hold_since).fromNow(true)])};
frm.dashboard.set_headline_alert(
'<div class="row">' +
'<div class="col-xs-12">' +
'<span class="indicator whitespace-nowrap '+ message.indicator +'"><span>'+ message.msg +'</span></span> ' +
'</div>' +
'</div>'
);
} else {
set_time_to_resolve_and_response(frm, data.message.apply_sla_for_resolution);
}
}
});
} else if (frm.doc.service_level_agreement) {
frm.dashboard.clear_headline();
let agreement_status = (frm.doc.agreement_status == 'Fulfilled') ?
{'indicator': 'green', 'msg': 'Service Level Agreement has been fulfilled'} :
{'indicator': 'red', 'msg': 'Service Level Agreement Failed'};
frm.dashboard.set_headline_alert(
'<div class="row">' +
'<div class="col-xs-12">' +
'<span class="indicator whitespace-nowrap '+ agreement_status.indicator +'"><span class="hidden-xs">'+ agreement_status.msg +'</span></span> ' +
'</div>' +
'</div>'
);
}
},
});
});
}
});
});
function set_time_to_resolve_and_response(frm, apply_sla_for_resolution) {
frm.dashboard.clear_headline();
let time_to_respond = get_status(frm.doc.response_by_variance);
if (!frm.doc.first_responded_on && frm.doc.agreement_status === 'Ongoing') {
time_to_respond = get_time_left(frm.doc.response_by, frm.doc.agreement_status);
}
let alert = `
<div class="row">
<div class="col-xs-12 col-sm-6">
<span class="indicator whitespace-nowrap ${time_to_respond.indicator}">
<span>Time to Respond: ${time_to_respond.diff_display}</span>
</span>
</div>`;
if (apply_sla_for_resolution) {
let time_to_resolve = get_status(frm.doc.resolution_by_variance);
if (!frm.doc.resolution_date && frm.doc.agreement_status === 'Ongoing') {
time_to_resolve = get_time_left(frm.doc.resolution_by, frm.doc.agreement_status);
}
alert += `
<div class="col-xs-12 col-sm-6">
<span class="indicator whitespace-nowrap ${time_to_resolve.indicator}">
<span>Time to Resolve: ${time_to_resolve.diff_display}</span>
</span>
</div>`;
}
alert += '</div>';
frm.dashboard.set_headline_alert(alert);
}
function get_time_left(timestamp, agreement_status) {
const diff = moment(timestamp).diff(moment());
const diff_display = diff >= 44500 ? moment.duration(diff).humanize() : 'Failed';
let indicator = (diff_display == 'Failed' && agreement_status != 'Fulfilled') ? 'red' : 'green';
return {'diff_display': diff_display, 'indicator': indicator};
}
function get_status(variance) {
if (variance > 0) {
return {'diff_display': 'Fulfilled', 'indicator': 'green'};
} else {
return {'diff_display': 'Failed', 'indicator': 'red'};
}
}
function attach_selector_button(inner_text, append_loction, context, grid_row) {
let $btn_div = $("<div>").css({"margin-bottom": "10px", "margin-top": "10px"})
.appendTo(append_loction);

View File

@ -9,94 +9,15 @@ frappe.ui.form.on("Issue", {
};
});
if (frappe.model.can_read("Support Settings")) {
frappe.db.get_value("Support Settings", {name: "Support Settings"},
["allow_resetting_service_level_agreement", "track_service_level_agreement"], (r) => {
if (r && r.track_service_level_agreement == "0") {
frm.set_df_property("service_level_section", "hidden", 1);
}
if (r && r.allow_resetting_service_level_agreement == "0") {
frm.set_df_property("reset_service_level_agreement", "hidden", 1);
}
});
}
if (frm.doc.service_level_agreement) {
frappe.call({
method: "erpnext.support.doctype.service_level_agreement.service_level_agreement.get_service_level_agreement_filters",
args: {
name: frm.doc.service_level_agreement,
customer: frm.doc.customer
},
callback: function (r) {
if (r && r.message) {
frm.set_query("priority", function() {
return {
filters: {
"name": ["in", r.message.priority],
}
};
});
frm.set_query("service_level_agreement", function() {
return {
filters: {
"name": ["in", r.message.service_level_agreements],
}
};
});
}
frappe.db.get_value("Support Settings", {name: "Support Settings"},
["allow_resetting_service_level_agreement", "track_service_level_agreement"], (r) => {
if (r && r.track_service_level_agreement == "0") {
frm.set_df_property("service_level_section", "hidden", 1);
}
if (r && r.allow_resetting_service_level_agreement == "0") {
frm.set_df_property("reset_service_level_agreement", "hidden", 1);
}
});
}
},
refresh: function(frm) {
// alert messages
if (frm.doc.status !== "Closed" && frm.doc.service_level_agreement
&& frm.doc.agreement_status === "Ongoing") {
frappe.call({
"method": "frappe.client.get",
args: {
doctype: "Service Level Agreement",
name: frm.doc.service_level_agreement
},
callback: function(data) {
let statuses = data.message.pause_sla_on;
const hold_statuses = [];
$.each(statuses, (_i, entry) => {
hold_statuses.push(entry.status);
});
if (hold_statuses.includes(frm.doc.status)) {
frm.dashboard.clear_headline();
let message = { "indicator": "orange", "msg": __("SLA is on hold since {0}", [moment(frm.doc.on_hold_since).fromNow(true)]) };
frm.dashboard.set_headline_alert(
'<div class="row">' +
'<div class="col-xs-12">' +
'<span class="indicator whitespace-nowrap ' + message.indicator + '"><span>' + message.msg + '</span></span> ' +
'</div>' +
'</div>'
);
} else {
set_time_to_resolve_and_response(frm);
}
}
});
} else if (frm.doc.service_level_agreement) {
frm.dashboard.clear_headline();
let agreement_status = (frm.doc.agreement_status == "Fulfilled") ?
{ "indicator": "green", "msg": "Service Level Agreement has been fulfilled" } :
{ "indicator": "red", "msg": "Service Level Agreement Failed" };
frm.dashboard.set_headline_alert(
'<div class="row">' +
'<div class="col-xs-12">' +
'<span class="indicator whitespace-nowrap ' + agreement_status.indicator + '"><span class="hidden-xs">' + agreement_status.msg + '</span></span> ' +
'</div>' +
'</div>'
);
}
// buttons
if (frm.doc.status !== "Closed") {
@ -142,7 +63,7 @@ frappe.ui.form.on("Issue", {
message: __("Resetting Service Level Agreement.")
});
frm.call("reset_service_level_agreement", {
frappe.call("erpnext.support.doctype.service_level_agreement.service_level_agreement.reset_service_level_agreement", {
reason: values.reason,
user: frappe.session.user_email
}, () => {
@ -224,44 +145,4 @@ frappe.ui.form.on("Issue", {
// frm.timeline.wrapper.data("help-article-event-attached", true);
// }
},
});
function set_time_to_resolve_and_response(frm) {
frm.dashboard.clear_headline();
var time_to_respond = get_status(frm.doc.response_by_variance);
if (!frm.doc.first_responded_on && frm.doc.agreement_status === "Ongoing") {
time_to_respond = get_time_left(frm.doc.response_by, frm.doc.agreement_status);
}
var time_to_resolve = get_status(frm.doc.resolution_by_variance);
if (!frm.doc.resolution_date && frm.doc.agreement_status === "Ongoing") {
time_to_resolve = get_time_left(frm.doc.resolution_by, frm.doc.agreement_status);
}
frm.dashboard.set_headline_alert(
'<div class="row">' +
'<div class="col-xs-12 col-sm-6">' +
'<span class="indicator whitespace-nowrap '+ time_to_respond.indicator +'"><span>Time to Respond: '+ time_to_respond.diff_display +'</span></span> ' +
'</div>' +
'<div class="col-xs-12 col-sm-6">' +
'<span class="indicator whitespace-nowrap '+ time_to_resolve.indicator +'"><span>Time to Resolve: '+ time_to_resolve.diff_display +'</span></span> ' +
'</div>' +
'</div>'
);
}
function get_time_left(timestamp, agreement_status) {
const diff = moment(timestamp).diff(moment());
const diff_display = diff >= 44500 ? moment.duration(diff).humanize() : "Failed";
let indicator = (diff_display == "Failed" && agreement_status != "Fulfilled") ? "red" : "green";
return {"diff_display": diff_display, "indicator": indicator};
}
function get_status(variance) {
if (variance > 0) {
return {"diff_display": "Fulfilled", "indicator": "green"};
} else {
return {"diff_display": "Failed", "indicator": "red"};
}
}
});

View File

@ -7,11 +7,10 @@ import json
from frappe import _
from frappe import utils
from frappe.model.document import Document
from frappe.utils import cint, now_datetime, getdate, get_weekdays, add_to_date, get_time, get_datetime, time_diff_in_seconds
from frappe.utils import now_datetime
from datetime import datetime, timedelta
from frappe.model.mapper import get_mapped_doc
from frappe.utils.user import is_website_user
from erpnext.support.doctype.service_level_agreement.service_level_agreement import get_active_service_level_agreement_for
from frappe.email.inbox import link_communication_to_document
class Issue(Document):
@ -25,8 +24,6 @@ class Issue(Document):
if not self.raised_by:
self.raised_by = frappe.session.user
self.change_service_level_agreement_and_priority()
self.update_status()
self.set_lead_contact(self.raised_by)
def on_update(self):
@ -54,99 +51,6 @@ class Issue(Document):
self.company = frappe.db.get_value("Lead", self.lead, "company") or \
frappe.db.get_default("Company")
def update_status(self):
status = frappe.db.get_value("Issue", self.name, "status")
if self.status != "Open" and status == "Open" and not self.first_responded_on:
self.first_responded_on = frappe.flags.current_time or now_datetime()
if self.status in ["Closed", "Resolved"] and status not in ["Resolved", "Closed"]:
self.resolution_date = frappe.flags.current_time or now_datetime()
if frappe.db.get_value("Issue", self.name, "agreement_status") == "Ongoing":
set_service_level_agreement_variance(issue=self.name)
self.update_agreement_status()
set_resolution_time(issue=self)
set_user_resolution_time(issue=self)
if self.status == "Open" and status != "Open":
# if no date, it should be set as None and not a blank string "", as per mysql strict config
self.resolution_date = None
self.reset_issue_metrics()
# enable SLA and variance on Reopen
self.agreement_status = "Ongoing"
set_service_level_agreement_variance(issue=self.name)
self.handle_hold_time(status)
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})
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()
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
# 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:
hold_time = self.total_hold_time if self.total_hold_time else 0
now_time = frappe.flags.current_time or now_datetime()
last_hold_time = 0
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
# re-calculate SLA variables after issue changes from any hold status to any non-hold status
# add hold time to SLA variables
start_date_time = get_datetime(self.service_level_agreement_creation)
priority = get_priority(self)
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 = 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
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
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:
self.agreement_status = "Failed"
else:
self.agreement_status = "Fulfilled"
def update_agreement_status_on_custom_status(self):
"""
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
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
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"
def create_communication(self):
communication = frappe.new_doc("Communication")
communication.update({
@ -213,194 +117,6 @@ class Issue(Document):
return replicated_issue.name
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)
else:
self.set_response_and_resolution_time()
def set_response_and_resolution_time(self, priority=None, service_level_agreement=None):
service_level_agreement = get_active_service_level_agreement_for(priority=priority,
customer=self.customer, service_level_agreement=service_level_agreement)
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))
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))
self.service_level_agreement = service_level_agreement.name
self.priority = service_level_agreement.default_priority if not priority else priority
priority = get_priority(self)
if not self.creation:
self.creation = now_datetime()
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_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 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)
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))
@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)
self.service_level_agreement_creation = now_datetime()
self.set_response_and_resolution_time(priority=self.priority, service_level_agreement=self.service_level_agreement)
self.agreement_status = "Ongoing"
self.save()
def reset_issue_metrics(self):
self.db_set("resolution_time", None)
self.db_set("user_resolution_time", None)
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
})
return priority
def get_expected_time_for(parameter, service_level, start_date_time):
current_date_time = start_date_time
expected_time = current_date_time
start_time = None
end_time = None
if parameter == "response":
allotted_seconds = service_level.get("response_time")
elif parameter == "resolution":
allotted_seconds = service_level.get("resolution_time")
else:
frappe.throw(_("{0} parameter is invalid").format(parameter))
expected_time_is_set = 0
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,
})
holidays = get_holidays(service_level.get("holiday_list"))
weekdays = get_weekdays()
while not expected_time_is_set:
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 \
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
elif allotted_seconds:
if time_left_today >= allotted_seconds:
expected_time = datetime.combine(getdate(current_date_time), get_time(start_time))
expected_time = add_to_date(expected_time, seconds=allotted_seconds)
expected_time_is_set = 1
else:
allotted_seconds = allotted_seconds - time_left_today
if not expected_time_is_set:
current_date_time = add_to_date(current_date_time, days=1)
if end_time and allotted_seconds >= 86400:
current_date_time = datetime.combine(getdate(current_date_time), get_time(end_time))
else:
current_date_time = expected_time
return current_date_time
def set_service_level_agreement_variance(issue=None):
current_time = frappe.flags.current_time or now_datetime()
filters = {"status": "Open", "agreement_status": "Ongoing"}
if issue:
filters = {"name": issue}
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
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)
if variance < 0:
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
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)
if variance < 0:
frappe.db.set_value(dt="Issue", dn=doc.name, field="agreement_status", val="Failed", update_modified=False)
def set_resolution_time(issue):
# total time taken from issue creation to closing
resolution_time = time_diff_in_seconds(issue.resolution_date, issue.creation)
issue.db_set("resolution_time", resolution_time)
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
},
fields=["sent_or_received", "name", "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 wait_time > 0:
pending_time.append(wait_time)
total_pending_time = sum(pending_time)
resolution_time_in_secs = time_diff_in_seconds(issue.resolution_date, issue.creation)
user_resolution_time = resolution_time_in_secs - total_pending_time
issue.db_set("user_resolution_time", user_resolution_time)
def get_list_context(context=None):
return {
"title": _("Issues"),
@ -439,15 +155,13 @@ def get_issue_list(doctype, txt, filters, limit_start, limit_page_length=20, ord
@frappe.whitelist()
def set_multiple_status(names, status):
names = json.loads(names)
for name in names:
set_status(name, status)
for name in json.loads(names):
frappe.db.set_value("Issue", name, "status", status)
@frappe.whitelist()
def set_status(name, status):
st = frappe.get_doc("Issue", name)
st.status = status
st.save()
frappe.db.set_value("Issue", name, "status", status)
def auto_close_tickets():
"""Auto-close replied support tickets after 7 days"""
@ -473,14 +187,6 @@ 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, {
@ -506,9 +212,7 @@ def make_issue_from_communication(communication, ignore_communication_links=Fals
return issue.name
def get_time_in_timedelta(time):
"""
Converts datetime.time(10, 36, 55, 961454) to datetime.timedelta(seconds=38215)
"""
import datetime
return datetime.timedelta(hours=time.hour, minutes=time.minute, seconds=time.second)
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

View File

@ -68,7 +68,7 @@ class TestIssue(unittest.TestCase):
self.assertEqual(issue.resolution_by, datetime.datetime(2019, 3, 6, 12, 0))
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 15, 0)
issue.reload()
issue.status = 'Closed'
issue.save()

View File

@ -1,4 +1,5 @@
{
"actions": [],
"creation": "2019-03-04 12:55:36.403035",
"doctype": "DocType",
"editable_grid": 1,
@ -16,7 +17,8 @@
"fieldtype": "Select",
"in_list_view": 1,
"label": "Workday",
"options": "Monday\nTuesday\nWednesday\nThursday\nFriday\nSaturday\nSunday"
"options": "Monday\nTuesday\nWednesday\nThursday\nFriday\nSaturday\nSunday",
"reqd": 1
},
{
"fieldname": "section_break_2",
@ -26,7 +28,8 @@
"fieldname": "start_time",
"fieldtype": "Time",
"in_list_view": 1,
"label": "Start Time"
"label": "Start Time",
"reqd": 1
},
{
"fieldname": "column_break_3",
@ -36,11 +39,13 @@
"fieldname": "end_time",
"fieldtype": "Time",
"in_list_view": 1,
"label": "End Time"
"label": "End Time",
"reqd": 1
}
],
"istable": 1,
"modified": "2019-05-05 19:15:08.999579",
"links": [],
"modified": "2020-07-06 13:28:47.303873",
"modified_by": "Administrator",
"module": "Support",
"name": "Service Day",

View File

@ -3,16 +3,87 @@
frappe.ui.form.on('Service Level Agreement', {
setup: function(frm) {
let allow_statuses = [];
const exclude_statuses = ['Open', 'Closed', 'Resolved'];
if (cint(frm.doc.apply_sla_for_resolution) === 1) {
frm.get_field('priorities').grid.editable_fields = [
{fieldname: 'priority', columns: 1},
{fieldname: 'default_priority', columns: 1},
{fieldname: 'response_time', columns: 2},
{fieldname: 'resolution_time', columns: 2}
];
} else {
frm.get_field('priorities').grid.editable_fields = [
{fieldname: 'priority', columns: 1},
{fieldname: 'default_priority', columns: 1},
{fieldname: 'response_time', columns: 3},
];
}
},
frappe.model.with_doctype('Issue', () => {
let statuses = frappe.meta.get_docfield('Issue', 'status', frm.doc.name).options;
statuses = statuses.split('\n');
allow_statuses = statuses.filter((status) => !exclude_statuses.includes(status));
frm.fields_dict.pause_sla_on.grid.update_docfield_property(
'status', 'options', [''].concat(allow_statuses)
);
refresh: function(frm) {
frm.trigger('fetch_status_fields');
frm.trigger('toggle_resolution_fields');
},
document_type: function(frm) {
frm.trigger('fetch_status_fields');
},
fetch_status_fields: function(frm) {
let allow_statuses = [];
let exclude_statuses = [];
if (frm.doc.document_type) {
frappe.model.with_doctype(frm.doc.document_type, () => {
let statuses = frappe.meta.get_docfield(frm.doc.document_type, 'status', frm.doc.name).options;
statuses = statuses.split('\n');
exclude_statuses = ['Open', 'Closed'];
allow_statuses = statuses.filter((status) => !exclude_statuses.includes(status));
frm.fields_dict.pause_sla_on.grid.update_docfield_property(
'status', 'options', [''].concat(allow_statuses)
);
exclude_statuses = ['Open'];
allow_statuses = statuses.filter((status) => !exclude_statuses.includes(status));
frm.fields_dict.sla_fulfilled_on.grid.update_docfield_property(
'status', 'options', [''].concat(allow_statuses)
);
});
}
frm.refresh_field('pause_sla_on');
},
apply_sla_for_resolution: function(frm) {
frm.trigger('toggle_resolution_fields');
},
toggle_resolution_fields: function(frm) {
if (cint(frm.doc.apply_sla_for_resolution) === 1) {
frm.fields_dict.priorities.grid.update_docfield_property('resolution_time', 'hidden', 0);
frm.fields_dict.priorities.grid.update_docfield_property('resolution_time', 'reqd', 1);
} else {
frm.fields_dict.priorities.grid.update_docfield_property('resolution_time', 'hidden', 1);
frm.fields_dict.priorities.grid.update_docfield_property('resolution_time', 'reqd', 0);
}
frm.refresh_field('priorities');
},
onload: function(frm) {
frm.set_query("document_type", function() {
let invalid_doctypes = frappe.model.core_doctypes_list;
invalid_doctypes.push(frm.doc.doctype, 'Cost Center', 'Company');
return {
filters: [
['DocType', 'issingle', '=', 0],
['DocType', 'istable', '=', 0],
['DocType', 'name', 'not in', invalid_doctypes],
['DocType', 'module', 'not in', ["Email", "Core", "Custom", "Event Streaming", "Social", "Data Migration", "Geo", "Desk"]]
]
};
});
}
});

View File

@ -1,18 +1,18 @@
{
"actions": [],
"autoname": "format:SLA-{service_level}-{####}",
"autoname": "format:SLA-{document_type}-{service_level}-{####}",
"creation": "2018-12-26 21:08:15.448812",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"enable",
"enabled",
"section_break_2",
"service_level",
"default_priority",
"document_type",
"default_service_level_agreement",
"default_priority",
"column_break_2",
"employee_group",
"service_level",
"holiday_list",
"entity_section",
"entity_type",
@ -20,13 +20,14 @@
"entity",
"agreement_details_section",
"start_date",
"active",
"column_break_7",
"end_date",
"section_break_18",
"pause_sla_on",
"response_and_resolution_time_section",
"apply_sla_for_resolution",
"priorities",
"status_details",
"sla_fulfilled_on",
"pause_sla_on",
"support_and_resolution_section_break",
"support_and_resolution"
],
@ -36,7 +37,7 @@
"fieldtype": "Data",
"in_list_view": 1,
"in_standard_filter": 1,
"label": "Service Level",
"label": "Service Level Name",
"reqd": 1
},
{
@ -51,20 +52,12 @@
"fieldtype": "Column Break"
},
{
"fieldname": "employee_group",
"fieldtype": "Link",
"in_list_view": 1,
"in_standard_filter": 1,
"label": "Employee Group",
"options": "Employee Group"
},
{
"depends_on": "eval: !doc.default_service_level_agreement",
"fieldname": "agreement_details_section",
"fieldtype": "Section Break",
"label": "Agreement Details"
},
{
"depends_on": "eval: !doc.default_service_level_agreement",
"fieldname": "start_date",
"fieldtype": "Date",
"label": "Start Date"
@ -81,21 +74,18 @@
"label": "End Date"
},
{
"collapsible": 1,
"fieldname": "response_and_resolution_time_section",
"fieldtype": "Section Break",
"label": "Response and Resolution Time"
},
{
"collapsible": 1,
"fieldname": "support_and_resolution_section_break",
"fieldtype": "Section Break",
"label": "Support Hours"
"label": "Working Hours"
},
{
"fieldname": "support_and_resolution",
"fieldtype": "Table",
"label": "Support and Resolution",
"options": "Service Day",
"reqd": 1
},
@ -106,13 +96,6 @@
"options": "Service Level Priority",
"reqd": 1
},
{
"default": "1",
"fieldname": "active",
"fieldtype": "Check",
"label": "Active",
"read_only": 1
},
{
"fieldname": "column_break_10",
"fieldtype": "Column Break"
@ -138,15 +121,10 @@
"label": "Entity Type",
"options": "\nCustomer\nCustomer Group\nTerritory"
},
{
"default": "1",
"fieldname": "enable",
"fieldtype": "Check",
"label": "Enable"
},
{
"fieldname": "section_break_2",
"fieldtype": "Section Break"
"fieldtype": "Section Break",
"hide_border": 1
},
{
"default": "0",
@ -161,20 +139,46 @@
"options": "Issue Priority",
"read_only": 1
},
{
"fieldname": "section_break_18",
"fieldtype": "Section Break",
"hide_border": 1
},
{
"fieldname": "pause_sla_on",
"fieldtype": "Table",
"label": "Pause SLA On",
"label": "SLA Paused On",
"options": "Pause SLA On Status"
},
{
"fieldname": "document_type",
"fieldtype": "Link",
"label": "Document Type",
"options": "DocType",
"reqd": 1
},
{
"default": "1",
"fieldname": "enabled",
"fieldtype": "Check",
"label": "Enabled"
},
{
"fieldname": "status_details",
"fieldtype": "Section Break",
"label": "Status Details"
},
{
"fieldname": "sla_fulfilled_on",
"fieldtype": "Table",
"label": "SLA Fulfilled On",
"options": "SLA Fulfilled On Status",
"reqd": 1
},
{
"default": "1",
"fieldname": "apply_sla_for_resolution",
"fieldtype": "Check",
"label": "Apply SLA for Resolution Time"
}
],
"links": [],
"modified": "2020-06-10 12:30:15.050785",
"modified": "2021-05-29 13:35:41.956849",
"modified_by": "Administrator",
"module": "Support",
"name": "Service Level Agreement",

View File

@ -6,44 +6,43 @@ from __future__ import unicode_literals
import frappe
from frappe.model.document import Document
from frappe import _
from frappe.utils import getdate, get_weekdays, get_link_to_form
from frappe.core.utils import get_parent_doc
from frappe.utils import time_diff_in_seconds, getdate, get_weekdays, add_to_date, get_time, get_datetime, \
get_time_zone, to_timedelta, get_datetime_str, get_link_to_form, cint
from datetime import datetime
from erpnext.support.doctype.issue.issue import get_holidays
class ServiceLevelAgreement(Document):
def validate(self):
self.validate_doc()
self.validate_status_field()
self.check_priorities()
self.check_support_and_resolution()
def check_priorities(self):
default_priority = []
priorities = []
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))
if not priority.response_time:
frappe.throw(_("Set Response Time for Priority {0} in row {1}.").format(priority.priority, priority.idx))
if self.apply_sla_for_resolution:
if not priority.resolution_time:
frappe.throw(_("Set Response Time for Priority {0} in row {1}.").format(priority.priority, priority.idx))
response = priority.response_time
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))
priorities.append(priority.priority)
if priority.default_priority:
default_priority.append(priority.default_priority)
response = priority.response_time
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))
# Check if repeated priority
if not len(set(priorities)) == len(priorities):
repeated_priority = get_repeated(priorities)
frappe.throw(_("Priority {0} has been repeated.").format(repeated_priority))
# Check if repeated default priority
if not len(set(default_priority)) == len(default_priority):
frappe.throw(_("Select only one Priority as Default."))
# set default priority from priorities
try:
self.default_priority = next(d.priority for d in self.priorities if d.default_priority)
@ -55,17 +54,12 @@ class ServiceLevelAgreement(Document):
support_days = []
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)))
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)))
if to_timedelta(support_and_resolution.start_time) >= to_timedelta(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))
# Check for repeated workday
if not len(set(support_days)) == len(support_days):
@ -73,24 +67,34 @@ 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:
if self.enabled and self.document_type == "Issue" \
and not frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
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]}):
frappe.throw(_("A Default Service Level Agreement already exists."))
else:
if self.start_date and self.end_date:
if getdate(self.start_date) >= getdate(self.end_date):
frappe.throw(_("Start Date of Agreement can't be greater than or equal to End Date."))
if self.default_service_level_agreement and frappe.db.exists("Service Level Agreement", {
"document_type": self.document_type,
"default_service_level_agreement": "1",
"name": ["!=", self.name]
}):
frappe.throw(_("Default Service Level Agreement for {0} already exists.").format(self.document_type))
if getdate(self.end_date) < getdate(frappe.utils.getdate()):
frappe.throw(_("End Date of Agreement can't be less than today."))
if self.start_date and self.end_date:
self.validate_from_to_dates(self.start_date, self.end_date)
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 self.entity_type and self.entity and frappe.db.exists("Service Level Agreement", {
"entity_type": self.entity_type,
"entity": self.entity,
"name": ["!=", self.name]
}):
frappe.throw(_("Service Level Agreement for {0} {1} already exists.").format(
frappe.bold(self.entity_type), frappe.bold(self.entity)))
def validate_status_field(self):
meta = frappe.get_meta(self.document_type)
if not meta.get_field("status"):
frappe.throw(_("The Document Type {0} must have a Status field to configure Service Level Agreement").format(
frappe.bold(self.document_type)))
def get_service_level_agreement_priority(self, priority):
priority = frappe.get_doc("Service Level Priority", {"priority": priority, "parent": self.name})
@ -101,78 +105,169 @@ class ServiceLevelAgreement(Document):
"resolution_time": priority.resolution_time
})
def before_insert(self):
# no need to set up SLA fields for Issue dt as they are standard fields in Issue
if self.document_type == "Issue":
return
service_level_agreement_fields = get_service_level_agreement_fields()
meta = frappe.get_meta(self.document_type, cached=False)
if meta.custom:
self.create_docfields(meta, service_level_agreement_fields)
else:
self.create_custom_fields(meta, service_level_agreement_fields)
def on_trash(self):
set_documents_with_active_service_level_agreement()
def after_insert(self):
set_documents_with_active_service_level_agreement()
def on_update(self):
set_documents_with_active_service_level_agreement()
def create_docfields(self, meta, service_level_agreement_fields):
last_index = len(meta.fields)
for field in service_level_agreement_fields:
if not meta.has_field(field.get("fieldname")):
last_index += 1
frappe.get_doc({
"doctype": "DocField",
"idx": last_index,
"parenttype": "DocType",
"parentfield": "fields",
"parent": self.document_type,
"label": field.get("label"),
"fieldname": field.get("fieldname"),
"fieldtype": field.get("fieldtype"),
"collapsible": field.get("collapsible"),
"options": field.get("options"),
"read_only": field.get("read_only"),
"hidden": field.get("hidden"),
"description": field.get("description"),
"default": field.get("default"),
}).insert(ignore_permissions=True)
else:
existing_field = meta.get_field(field.get("fieldname"))
self.reset_field_properties(existing_field, "DocField", field)
# to update meta and modified timestamp
frappe.get_doc('DocType', self.document_type).save(ignore_permissions=True)
def create_custom_fields(self, meta, service_level_agreement_fields):
for field in service_level_agreement_fields:
if not meta.has_field(field.get("fieldname")):
frappe.get_doc({
"doctype": "Custom Field",
"dt": self.document_type,
"label": field.get("label"),
"fieldname": field.get("fieldname"),
"fieldtype": field.get("fieldtype"),
"insert_after": "append",
"collapsible": field.get("collapsible"),
"options": field.get("options"),
"read_only": field.get("read_only"),
"hidden": field.get("hidden"),
"description": field.get("description"),
"default": field.get("default"),
}).insert(ignore_permissions=True)
else:
existing_field = meta.get_field(field.get("fieldname"))
self.reset_field_properties(existing_field, "Custom Field", field)
def reset_field_properties(self, field, field_dt, sla_field):
field = frappe.get_doc(field_dt, {"fieldname": field.fieldname})
field.label = sla_field.get("label")
field.fieldname = sla_field.get("fieldname")
field.fieldtype = sla_field.get("fieldtype")
field.collapsible = sla_field.get("collapsible")
field.hidden = sla_field.get("hidden")
field.options = sla_field.get("options")
field.read_only = sla_field.get("read_only")
field.hidden = sla_field.get("hidden")
field.description = sla_field.get("description")
field.default = sla_field.get("default")
field.save(ignore_permissions=True)
def check_agreement_status():
service_level_agreements = frappe.get_list("Service Level Agreement", filters=[
{"active": 1},
service_level_agreements = frappe.get_all("Service Level Agreement", filters=[
{"enabled": 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)
frappe.db.set_value("Service Level Agreement", service_level_agreement.name, "enabled", 0)
def get_active_service_level_agreement_for(priority, customer=None, service_level_agreement=None):
if not frappe.db.get_single_value("Support Settings", "track_service_level_agreement"):
def get_active_service_level_agreement_for(doctype, priority, customer=None, service_level_agreement=None):
if doctype == "Issue" and 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", "document_type", "=", doctype],
["Service Level Agreement", "enabled", "=", 1]
]
if priority:
filters.append(["Service Level Priority", "priority", "=", priority])
or_filters = [
["Service Level Agreement", "entity", "in", [customer, get_customer_group(customer), get_customer_territory(customer)]]
]
or_filters = []
if service_level_agreement:
or_filters = [
["Service Level Agreement", "name", "=", service_level_agreement],
]
if customer:
or_filters.append(
["Service Level Agreement", "entity", "in", [customer, get_customer_group(customer), get_customer_territory(customer)]]
)
or_filters.append(["Service Level Agreement", "default_service_level_agreement", "=", 1])
agreement = frappe.get_list("Service Level Agreement", filters=filters, or_filters=or_filters,
fields=["name", "default_priority"])
agreement = frappe.get_all("Service Level Agreement", filters=filters, or_filters=or_filters,
fields=["name", "default_priority", "apply_sla_for_resolution"])
return agreement[0] if agreement else None
def get_customer_group(customer):
if customer:
return frappe.db.get_value("Customer", customer, "customer_group")
return frappe.db.get_value("Customer", customer, "customer_group") if customer else None
def get_customer_territory(customer):
if customer:
return frappe.db.get_value("Customer", customer, "territory")
return frappe.db.get_value("Customer", customer, "territory") if customer else None
@frappe.whitelist()
def get_service_level_agreement_filters(name, customer=None):
def get_service_level_agreement_filters(doctype, name, customer=None):
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", "document_type", "=", doctype],
["Service Level Agreement", "enabled", "=", 1]
]
if not customer:
or_filters = [
["Service Level Agreement", "default_service_level_agreement", "=", 1]
]
else:
or_filters = [
["Service Level Agreement", "default_service_level_agreement", "=", 1]
]
if customer:
# 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]
]
or_filters.append(
["Service Level Agreement", "entity", "in", [customer, get_customer_group(customer), get_customer_territory(customer), ""]]
)
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_all("Service Level Priority", filters={"parent": name}, fields=["priority"])],
"service_level_agreements": [d.name for d in frappe.get_all("Service Level Agreement", filters=filters, or_filters=or_filters)]
}
def get_repeated(values):
unique_list = []
diff = []
@ -183,3 +278,573 @@ def get_repeated(values):
if value not in diff:
diff.append(str(value))
return " ".join(diff)
def get_documents_with_active_service_level_agreement():
if not frappe.cache().hget("service_level_agreement", "active"):
set_documents_with_active_service_level_agreement()
return frappe.cache().hget("service_level_agreement", "active")
def set_documents_with_active_service_level_agreement():
active = [sla.document_type for sla in frappe.get_all("Service Level Agreement", fields=["document_type"])]
frappe.cache().hset("service_level_agreement", "active", active)
def apply(doc, method=None):
# Applies SLA to document on validate
if frappe.flags.in_patch or frappe.flags.in_install or frappe.flags.in_setup_wizard or \
doc.doctype not in get_documents_with_active_service_level_agreement():
return
service_level_agreement = get_active_service_level_agreement_for(doctype=doc.get("doctype"), priority=doc.get("priority"),
customer=doc.get("customer"), service_level_agreement=doc.get("service_level_agreement"))
if not service_level_agreement:
return
set_sla_properties(doc, service_level_agreement)
def set_sla_properties(doc, service_level_agreement):
if frappe.db.exists(doc.doctype, doc.name):
from_db = frappe.get_doc(doc.doctype, doc.name)
else:
from_db = frappe._dict({})
meta = frappe.get_meta(doc.doctype)
if meta.has_field("customer") and service_level_agreement.customer and doc.get("customer") and \
not service_level_agreement.customer == doc.get("customer"):
frappe.throw(_("Service Level Agreement {0} is specific to Customer {1}").format(service_level_agreement.name,
service_level_agreement.customer))
doc.service_level_agreement = service_level_agreement.name
doc.priority = doc.get("priority") or service_level_agreement.default_priority
priority = get_priority(doc)
if not doc.creation:
doc.creation = now_datetime(doc.get("owner"))
if meta.has_field("service_level_agreement_creation"):
doc.service_level_agreement_creation = now_datetime(doc.get("owner"))
start_date_time = get_datetime(doc.get("service_level_agreement_creation") or doc.creation)
set_response_by_and_variance(doc, meta, start_date_time, priority)
if service_level_agreement.apply_sla_for_resolution:
set_resolution_by_and_variance(doc, meta, start_date_time, priority)
update_status(doc, from_db, meta)
def update_status(doc, from_db, meta):
if meta.has_field("status"):
if meta.has_field("first_responded_on") and doc.status != "Open" and \
from_db.status == "Open" and not doc.first_responded_on:
doc.first_responded_on = frappe.flags.current_time or now_datetime(doc.get("owner"))
if meta.has_field("service_level_agreement") and doc.service_level_agreement:
# mark sla status as fulfilled based on the configuration
fulfillment_statuses = [entry.status for entry in frappe.db.get_all("SLA Fulfilled On Status", filters={
"parent": doc.service_level_agreement
}, fields=["status"])]
if doc.status in fulfillment_statuses and from_db.status not in fulfillment_statuses:
apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", doc.service_level_agreement,
"apply_sla_for_resolution")
if apply_sla_for_resolution and meta.has_field("resolution_date"):
doc.resolution_date = frappe.flags.current_time or now_datetime(doc.get("owner"))
if meta.has_field("agreement_status") and from_db.agreement_status == "Ongoing":
set_service_level_agreement_variance(doc.doctype, doc.name)
update_agreement_status(doc, meta)
if apply_sla_for_resolution:
set_resolution_time(doc, meta)
set_user_resolution_time(doc, meta)
if doc.status == "Open" and from_db.status != "Open":
# if no date, it should be set as None and not a blank string "", as per mysql strict config
# enable SLA and variance on Reopen
reset_metrics(doc, meta)
set_service_level_agreement_variance(doc.doctype, doc.name)
handle_hold_time(doc, meta, from_db.status)
def get_expected_time_for(parameter, service_level, start_date_time):
current_date_time = start_date_time
expected_time = current_date_time
start_time = end_time = None
expected_time_is_set = 0
allotted_seconds = get_allotted_seconds(parameter, service_level)
support_days = get_support_days(service_level)
holidays = get_holidays(service_level.get("holiday_list"))
weekdays = get_weekdays()
while not expected_time_is_set:
current_weekday = weekdays[current_date_time.weekday()]
if not is_holiday(current_date_time, holidays) and current_weekday in support_days:
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)
else:
start_time = 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
elif allotted_seconds:
if time_left_today >= allotted_seconds:
expected_time = datetime.combine(getdate(current_date_time), get_time(start_time))
expected_time = add_to_date(expected_time, seconds=allotted_seconds)
expected_time_is_set = 1
else:
allotted_seconds = allotted_seconds - time_left_today
if not expected_time_is_set:
current_date_time = add_to_date(current_date_time, days=1)
if end_time and allotted_seconds >= 86400:
current_date_time = datetime.combine(getdate(current_date_time), get_time(end_time))
else:
current_date_time = expected_time
return current_date_time
def get_allotted_seconds(parameter, service_level):
allotted_seconds = 0
if parameter == "response":
allotted_seconds = service_level.get("response_time")
elif parameter == "resolution":
allotted_seconds = service_level.get("resolution_time")
else:
frappe.throw(_("{0} parameter is invalid").format(parameter))
return allotted_seconds
def get_support_days(service_level):
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,
})
return support_days
def set_service_level_agreement_variance(doctype, doc=None):
filters = {"status": "Open", "agreement_status": "Ongoing"}
if doc:
filters = {"name": doc}
for entry in frappe.get_all(doctype, filters=filters):
current_doc = frappe.get_doc(doctype, entry.name)
current_time = frappe.flags.current_time or now_datetime(current_doc.get("owner"))
apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", current_doc.service_level_agreement,
"apply_sla_for_resolution")
if not current_doc.first_responded_on: # first_responded_on set when first reply is sent to customer
variance = round(time_diff_in_seconds(current_doc.response_by, current_time), 2)
frappe.db.set_value(current_doc.doctype, current_doc.name, "response_by_variance", variance, update_modified=False)
if variance < 0:
frappe.db.set_value(current_doc.doctype, current_doc.name, "agreement_status", "Failed", update_modified=False)
if apply_sla_for_resolution and not current_doc.get("resolution_date"): # resolution_date set when issue has been closed
variance = round(time_diff_in_seconds(current_doc.resolution_by, current_time), 2)
frappe.db.set_value(current_doc.doctype, current_doc.name, "resolution_by_variance", variance, update_modified=False)
if variance < 0:
frappe.db.set_value(current_doc.doctype, current_doc.name, "agreement_status", "Failed", update_modified=False)
def set_user_resolution_time(doc, meta):
# total time taken by a user to close the issue apart from wait_time
if not meta.has_field("user_resolution_time"):
return
communications = frappe.get_all("Communication", filters={
"reference_doctype": doc.doctype,
"reference_name": doc.name
}, fields=["sent_or_received", "name", "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 wait_time > 0:
pending_time.append(wait_time)
total_pending_time = sum(pending_time)
resolution_time_in_secs = time_diff_in_seconds(doc.resolution_date, doc.creation)
doc.user_resolution_time = resolution_time_in_secs - total_pending_time
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 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)
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))
def get_priority(doc):
service_level_agreement = frappe.get_doc("Service Level Agreement", doc.service_level_agreement)
priority = service_level_agreement.get_service_level_agreement_priority(doc.priority)
priority.update({
"support_and_resolution": service_level_agreement.support_and_resolution,
"holiday_list": service_level_agreement.holiday_list
})
return priority
def reset_service_level_agreement(doc, 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": doc.doctype,
"reference_name": doc.name,
"comment_email": user,
"content": " resetted Service Level Agreement - {0}".format(_(reason)),
}).insert(ignore_permissions=True)
doc.service_level_agreement_creation = now_datetime(doc.get("owner"))
doc.set_response_and_resolution_time(priority=doc.priority, service_level_agreement=doc.service_level_agreement)
doc.agreement_status = "Ongoing"
doc.save()
def reset_metrics(doc, meta):
if meta.has_field("resolution_date"):
doc.resolution_date = None
if not meta.has_field("resolution_time"):
doc.resolution_time = None
if not meta.has_field("user_resolution_time"):
doc.user_resolution_time = None
if meta.has_field("agreement_status"):
doc.agreement_status = "Ongoing"
def set_resolution_time(doc, meta):
# total time taken from issue creation to closing
if not meta.has_field("resolution_time"):
return
doc.resolution_time = time_diff_in_seconds(doc.resolution_date, doc.creation)
# called via hooks on communication update
def update_hold_time(doc, status):
parent = get_parent_doc(doc)
if not parent:
return
if doc.communication_type == "Comment":
return
status_field = parent.meta.get_field("status")
if status_field:
options = (status_field.options or "").splitlines()
# if status has a "Replied" option, then handle hold time
if ("Replied" in options) and doc.sent_or_received == "Received":
meta = frappe.get_meta(parent.doctype)
handle_hold_time(parent, meta, 'Replied')
def handle_hold_time(doc, meta, status):
if meta.has_field("service_level_agreement") and doc.service_level_agreement:
# set response and resolution variance as None as the issue is on Hold for status as Replied
hold_statuses = [entry.status for entry in frappe.db.get_all("Pause SLA On Status", filters={
"parent": doc.service_level_agreement
}, fields=["status"])]
if not hold_statuses:
return
if meta.has_field("status") and doc.status in hold_statuses and status not in hold_statuses:
apply_hold_status(doc, meta)
# calculate hold time when status is changed from any hold status to any non-hold status
if meta.has_field("status") and doc.status not in hold_statuses and status in hold_statuses:
reset_hold_status_and_update_hold_time(doc, meta)
def apply_hold_status(doc, meta):
update_values = {'on_hold_since': frappe.flags.current_time or now_datetime(doc.get("owner"))}
if meta.has_field("first_responded_on") and not doc.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
doc.db_set(update_values)
def reset_hold_status_and_update_hold_time(doc, meta):
hold_time = doc.total_hold_time if meta.has_field("total_hold_time") and doc.total_hold_time else 0
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
last_hold_time = 0
update_values = {}
if meta.has_field("on_hold_since") and doc.on_hold_since:
# last_hold_time will be added to the sla variables
last_hold_time = time_diff_in_seconds(now_time, doc.on_hold_since)
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
start_date_time = get_datetime(doc.get("service_level_agreement_creation") or doc.creation)
priority = get_priority(doc)
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
# add hold time to response by variance
if meta.has_field("first_responded_on") and not doc.first_responded_on:
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
# add hold time to resolution by variance
if frappe.db.get_value("Service Level Agreement", doc.service_level_agreement, "apply_sla_for_resolution"):
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
doc.db_set(update_values)
def get_service_level_agreement_fields():
return [
{
"collapsible": 1,
"fieldname": "service_level_section",
"fieldtype": "Section Break",
"label": "Service Level"
},
{
"fieldname": "service_level_agreement",
"fieldtype": "Link",
"label": "Service Level Agreement",
"options": "Service Level Agreement"
},
{
"fieldname": "priority",
"fieldtype": "Link",
"label": "Priority",
"options": "Issue Priority"
},
{
"fieldname": "response_by",
"fieldtype": "Datetime",
"label": "Response By",
"read_only": 1
},
{
"fieldname": "response_by_variance",
"fieldtype": "Duration",
"hide_seconds": 1,
"label": "Response By Variance",
"read_only": 1
},
{
"fieldname": "first_responded_on",
"fieldtype": "Datetime",
"label": "First Responded On",
"read_only": 1
},
{
"fieldname": "on_hold_since",
"fieldtype": "Datetime",
"hidden": 1,
"label": "On Hold Since",
"read_only": 1
},
{
"fieldname": "total_hold_time",
"fieldtype": "Duration",
"label": "Total Hold Time",
"read_only": 1
},
{
"fieldname": "cb",
"fieldtype": "Column Break",
"read_only": 1
},
{
"default": "Ongoing",
"fieldname": "agreement_status",
"fieldtype": "Select",
"label": "Service Level Agreement Status",
"options": "Ongoing\nFulfilled\nFailed",
"read_only": 1
},
{
"fieldname": "resolution_by",
"fieldtype": "Datetime",
"label": "Resolution By",
"read_only": 1
},
{
"fieldname": "resolution_by_variance",
"fieldtype": "Duration",
"hide_seconds": 1,
"label": "Resolution By Variance",
"read_only": 1
},
{
"fieldname": "service_level_agreement_creation",
"fieldtype": "Datetime",
"hidden": 1,
"label": "Service Level Agreement Creation",
"read_only": 1
},
{
"depends_on": "eval:!doc.__islocal",
"fieldname": "resolution_date",
"fieldtype": "Datetime",
"label": "Resolution Date",
"no_copy": 1,
"read_only": 1
}
]
def update_agreement_status_on_custom_status(doc):
# Update Agreement Fulfilled status using Custom Scripts for Custom Status
meta = frappe.get_meta(doc.doctype)
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
if meta.has_field("first_responded_on") and not doc.first_responded_on:
# first_responded_on set when first reply is sent to customer
doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, now_time), 2)
if meta.has_field("resolution_date") and not doc.resolution_date:
# resolution_date set when issue has been closed
doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, now_time), 2)
if meta.has_field("agreement_status"):
doc.agreement_status = "Fulfilled" if doc.response_by_variance > 0 and doc.resolution_by_variance > 0 else "Failed"
def update_agreement_status(doc, meta):
if meta.has_field("service_level_agreement") and meta.has_field("agreement_status") and \
doc.service_level_agreement and doc.agreement_status == "Ongoing":
apply_sla_for_resolution = frappe.db.get_value("Service Level Agreement", doc.service_level_agreement,
"apply_sla_for_resolution")
# if SLA is applied for resolution check for response and resolution, else only response
if apply_sla_for_resolution:
if meta.has_field("response_by_variance") and meta.has_field("resolution_by_variance"):
if cint(frappe.db.get_value(doc.doctype, doc.name, "response_by_variance")) < 0 or \
cint(frappe.db.get_value(doc.doctype, doc.name, "resolution_by_variance")) < 0:
doc.agreement_status = "Failed"
else:
doc.agreement_status = "Fulfilled"
else:
if meta.has_field("response_by_variance") and \
cint(frappe.db.get_value(doc.doctype, doc.name, "response_by_variance")) < 0:
doc.agreement_status = "Failed"
else:
doc.agreement_status = "Fulfilled"
def is_holiday(date, holidays):
return getdate(date) in holidays
def get_time_in_timedelta(time):
"""Converts datetime.time(10, 36, 55, 961454) to datetime.timedelta(seconds=38215)."""
import datetime
return datetime.timedelta(hours=time.hour, minutes=time.minute, seconds=time.second)
def set_response_by_and_variance(doc, meta, start_date_time, priority):
if meta.has_field("response_by"):
doc.response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
if meta.has_field("response_by_variance"):
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
doc.response_by_variance = round(time_diff_in_seconds(doc.response_by, now_time), 2)
def set_resolution_by_and_variance(doc, meta, start_date_time, priority):
if meta.has_field("resolution_by"):
doc.resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
if meta.has_field("resolution_by_variance"):
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
doc.resolution_by_variance = round(time_diff_in_seconds(doc.resolution_by, now_time), 2)
def now_datetime(user):
dt = convert_utc_to_user_timezone(datetime.utcnow(), user)
return dt.replace(tzinfo=None)
def convert_utc_to_user_timezone(utc_timestamp, user):
from pytz import timezone, UnknownTimeZoneError
user_tz = get_tz(user)
utcnow = timezone('UTC').localize(utc_timestamp)
try:
return utcnow.astimezone(timezone(user_tz))
except UnknownTimeZoneError:
return utcnow
def get_tz(user):
return frappe.db.get_value("User", user, "time_zone") or get_time_zone()
@frappe.whitelist()
def get_user_time(user, to_string=False):
return get_datetime_str(now_datetime(user)) if to_string else now_datetime(user)
@frappe.whitelist()
def get_sla_doctypes():
doctypes = []
data = frappe.get_list('Service Level Agreement',
{'enabled': 1},
['document_type'],
distinct=1
)
for entry in data:
doctypes.append(entry.document_type)
return doctypes

View File

@ -5,19 +5,20 @@ from __future__ import unicode_literals
import frappe
import unittest
from erpnext.hr.doctype.employee_group.test_employee_group import make_employee_group
import datetime
from frappe.utils import flt
from erpnext.support.doctype.issue_priority.test_issue_priority import make_priorities
from erpnext.support.doctype.service_level_agreement.service_level_agreement import get_service_level_agreement_fields
class TestServiceLevelAgreement(unittest.TestCase):
def setUp(self):
frappe.db.sql("delete from `tabService Level Agreement`")
frappe.db.set_value("Support Settings", None, "track_service_level_agreement", 1)
frappe.db.sql("delete from `tabLead`")
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)
holiday_list="__Test Holiday List", 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)
@ -29,8 +30,8 @@ class TestServiceLevelAgreement(unittest.TestCase):
# 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)
holiday_list="__Test Holiday List", 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)
@ -41,8 +42,8 @@ class TestServiceLevelAgreement(unittest.TestCase):
# 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)
holiday_list="__Test Holiday List", 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)
@ -53,7 +54,7 @@ class TestServiceLevelAgreement(unittest.TestCase):
# 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",
holiday_list="__Test Holiday List",
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)
@ -62,64 +63,223 @@ class TestServiceLevelAgreement(unittest.TestCase):
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 test_custom_field_creation_for_sla_on_standard_dt(self):
# Default Service Level Agreement
doctype = "Lead"
lead_sla = create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400, resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}]
)
def get_service_level_agreement(default_service_level_agreement=None, entity_type=None, entity=None):
# check default SLA for lead
default_sla = get_service_level_agreement(default_service_level_agreement=1, doctype=doctype)
self.assertEqual(lead_sla.name, default_sla.name)
# check SLA custom fields created for leads
sla_fields = get_service_level_agreement_fields()
meta = frappe.get_meta(doctype, cached=False)
for field in sla_fields:
self.assertTrue(meta.has_field(field.get("fieldname")))
def test_docfield_creation_for_sla_on_custom_dt(self):
doctype = create_custom_doctype()
sla = create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400, resolution_time=21600,
doctype=doctype.name
)
# check default SLA for custom dt
default_sla = get_service_level_agreement(default_service_level_agreement=1, doctype=doctype.name)
self.assertEqual(sla.name, default_sla.name)
# check SLA docfields created
sla_fields = get_service_level_agreement_fields()
meta = frappe.get_meta(doctype.name, cached=False)
for field in sla_fields:
self.assertTrue(meta.has_field(field.get("fieldname")))
def test_sla_application(self):
# Default Service Level Agreement
doctype = "Lead"
lead_sla = create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400, resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}]
)
# make lead with default SLA
creation = datetime.datetime(2019, 3, 4, 12, 0)
lead = make_lead(creation=creation, index=1)
self.assertEqual(lead.service_level_agreement, lead_sla.name)
self.assertEqual(lead.response_by, datetime.datetime(2019, 3, 4, 16, 0))
self.assertEqual(lead.resolution_by, datetime.datetime(2019, 3, 4, 18, 0))
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 15, 0)
lead.reload()
lead.status = 'Converted'
lead.save()
self.assertEqual(lead.agreement_status, 'Fulfilled')
def test_hold_time(self):
doctype = "Lead"
create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400, resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}],
pause_sla_on=[{"status": "Replied"}]
)
creation = datetime.datetime(2020, 3, 4, 4, 0)
lead = make_lead(creation, index=2)
frappe.flags.current_time = datetime.datetime(2020, 3, 4, 4, 15)
lead.reload()
lead.status = 'Replied'
lead.save()
lead.reload()
self.assertEqual(lead.on_hold_since, frappe.flags.current_time)
frappe.flags.current_time = datetime.datetime(2020, 3, 4, 5, 5)
lead.reload()
lead.status = 'Converted'
lead.save()
lead.reload()
self.assertEqual(flt(lead.total_hold_time, 2), 3000)
self.assertEqual(lead.resolution_by, datetime.datetime(2020, 3, 4, 16, 50))
def test_failed_sla_for_response_only(self):
doctype = "Lead"
create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400,
doctype=doctype,
sla_fulfilled_on=[{"status": "Replied"}],
pause_sla_on=[],
apply_sla_for_resolution=0
)
creation = datetime.datetime(2019, 3, 4, 12, 0)
lead = make_lead(creation=creation, index=1)
self.assertEqual(lead.response_by, datetime.datetime(2019, 3, 4, 16, 0))
# failed with response time only
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 16, 5)
lead.reload()
lead.status = 'Replied'
lead.save()
lead.reload()
self.assertEqual(lead.agreement_status, 'Failed')
def test_fulfilled_sla_for_response_only(self):
doctype = "Lead"
lead_sla = create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
response_time=14400,
doctype=doctype,
sla_fulfilled_on=[{"status": "Replied"}],
apply_sla_for_resolution=0
)
# fulfilled with response time only
creation = datetime.datetime(2019, 3, 4, 12, 0)
lead = make_lead(creation=creation, index=2)
self.assertEqual(lead.service_level_agreement, lead_sla.name)
self.assertEqual(lead.response_by, datetime.datetime(2019, 3, 4, 16, 0))
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 15, 30)
lead.reload()
lead.status = 'Replied'
lead.save()
lead.reload()
self.assertEqual(lead.agreement_status, 'Fulfilled')
def tearDown(self):
for d in frappe.get_all("Service Level Agreement"):
frappe.delete_doc("Service Level Agreement", d.name, force=1)
def get_service_level_agreement(default_service_level_agreement=None, entity_type=None, entity=None, doctype="Issue"):
if default_service_level_agreement:
filters = {"default_service_level_agreement": default_service_level_agreement}
filters = {"default_service_level_agreement": default_service_level_agreement, "document_type": doctype}
else:
filters = {"entity_type": entity_type, "entity": entity}
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, response_time, entity_type,
entity, resolution_time=0, doctype="Issue", sla_fulfilled_on=[], pause_sla_on=[], apply_sla_for_resolution=1):
employee_group = make_employee_group()
make_holiday_list()
make_priorities()
service_level_agreement = frappe.get_doc({
if not sla_fulfilled_on:
sla_fulfilled_on = [
{"status": "Resolved"},
{"status": "Closed"}
]
pause_sla_on = [{"status": "Replied"}] if doctype == "Issue" else pause_sla_on
service_level_agreement = frappe._dict({
"doctype": "Service Level Agreement",
"enable": 1,
"enabled": 1,
"document_type": doctype,
"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),
"apply_sla_for_resolution": apply_sla_for_resolution,
"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"
}
],
"sla_fulfilled_on": sla_fulfilled_on,
"pause_sla_on": pause_sla_on,
"support_and_resolution": [
{
"workday": "Monday",
@ -173,10 +333,13 @@ def create_service_level_agreement(default_service_level_agreement, holiday_list
service_level_agreement_exists = frappe.db.exists("Service Level Agreement", filters)
if not service_level_agreement_exists:
service_level_agreement.insert(ignore_permissions=True)
return service_level_agreement
doc = frappe.get_doc(service_level_agreement).insert(ignore_permissions=True)
else:
return frappe.get_doc("Service Level Agreement", service_level_agreement_exists)
doc = frappe.get_doc("Service Level Agreement", service_level_agreement_exists)
doc.update(service_level_agreement)
doc.save()
return doc
def create_customer():
@ -219,19 +382,19 @@ def create_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)
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)
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)
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)
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")
@ -256,3 +419,55 @@ def make_holiday_list():
},
]
}).insert()
def create_custom_doctype():
if not frappe.db.exists("DocType", "Test SLA on Custom Dt"):
doc = frappe.get_doc({
"doctype": "DocType",
"module": "Support",
"custom": 1,
"fields": [
{
"label": "Date",
"fieldname": "date",
"fieldtype": "Date"
},
{
"label": "Description",
"fieldname": "desc",
"fieldtype": "Long Text"
},
{
"label": "Email ID",
"fieldname": "email_id",
"fieldtype": "Link",
"options": "Customer"
},
{
"label": "Status",
"fieldname": "status",
"fieldtype": "Select",
"options": "Open\nReplied\nClosed"
}
],
"permissions": [{
"role": "System Manager",
"read": 1,
"write": 1
}],
"name": "Test SLA on Custom Dt",
})
doc.insert()
return doc
else:
return frappe.get_doc("DocType", "Test SLA on Custom Dt")
def make_lead(creation=None, index=0):
return frappe.get_doc({
"doctype": "Lead",
"email_id": "test_lead1@example{0}.com".format(index),
"lead_name": "_Test Lead {0}".format(index),
"status": "Open",
"creation": creation,
"service_level_agreement_creation": creation
}).insert(ignore_permissions=True)

View File

@ -15,12 +15,13 @@
],
"fields": [
{
"columns": 2,
"columns": 1,
"fieldname": "priority",
"fieldtype": "Link",
"in_list_view": 1,
"label": "Priority",
"options": "Issue Priority"
"options": "Issue Priority",
"reqd": 1
},
{
"fieldname": "sb_00",
@ -32,7 +33,6 @@
"fieldtype": "Duration",
"hide_days": 1,
"hide_seconds": 1,
"in_list_view": 1,
"label": "Resolution Time"
},
{
@ -58,12 +58,13 @@
"hide_days": 1,
"hide_seconds": 1,
"in_list_view": 1,
"label": "First Response Time"
"label": "First Response Time",
"reqd": 1
}
],
"istable": 1,
"links": [],
"modified": "2020-06-10 12:45:47.545915",
"modified": "2021-05-29 19:52:51.733248",
"modified_by": "Administrator",
"module": "Support",
"name": "Service Level Priority",
@ -73,4 +74,4 @@
"sort_field": "modified",
"sort_order": "DESC",
"track_changes": 1
}
}

View File

@ -0,0 +1,31 @@
{
"actions": [],
"creation": "2021-05-26 21:11:29.176369",
"doctype": "DocType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
"status"
],
"fields": [
{
"fieldname": "status",
"fieldtype": "Select",
"in_list_view": 1,
"label": "Status",
"reqd": 1
}
],
"istable": 1,
"links": [],
"modified": "2021-05-26 21:11:29.176369",
"modified_by": "Administrator",
"module": "Support",
"name": "SLA Fulfilled On Status",
"owner": "Administrator",
"permissions": [],
"quick_entry": 1,
"sort_field": "modified",
"sort_order": "DESC",
"track_changes": 1
}

View File

@ -0,0 +1,8 @@
# Copyright (c) 2021, Frappe Technologies Pvt. Ltd. and contributors
# For license information, please see license.txt
# import frappe
from frappe.model.document import Document
class SLAFulfilledOnStatus(Document):
pass