style: format code with black

This commit is contained in:
Ankush Menat
2022-03-28 18:52:46 +05:30
parent 21e00da3d6
commit 494bd9ef78
1395 changed files with 91704 additions and 62532 deletions

View File

@@ -1,5 +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"},
]

View File

@@ -50,23 +50,26 @@ 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 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()
@@ -97,23 +100,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
@@ -121,6 +132,7 @@ class Issue(Document):
self.db_set("resolution_time", None)
self.db_set("user_resolution_time", None)
def get_list_context(context=None):
return {
"title": _("Issues"),
@@ -128,7 +140,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,
}
@@ -145,7 +157,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
@@ -154,7 +167,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()
@@ -163,16 +178,24 @@ def set_multiple_status(names, status):
for name in json.loads(names):
frappe.db.set_value("Issue", name, "status", status)
@frappe.whitelist()
def set_status(name, status):
frappe.db.set_value("Issue", name, "status", status)
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"))
@@ -181,72 +204,87 @@ 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)
@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.service_level_agreement_creation or 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
@@ -259,7 +297,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
@@ -269,7 +307,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):
@@ -294,9 +334,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)
@@ -304,6 +346,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:
@@ -311,6 +354,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)
@@ -318,19 +362,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)
@@ -338,6 +385,7 @@ def is_before_working_hours(date, support_hours):
return True
return False
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]

View File

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

View File

@@ -23,6 +23,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")
@@ -41,9 +42,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"))
@@ -78,10 +80,10 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2019-03-04 15:00")
issue.reload()
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
self.assertEqual(issue.agreement_status, 'Fulfilled')
self.assertEqual(issue.agreement_status, "Fulfilled")
def test_hold_time_on_replied(self):
creation = get_datetime("2020-03-04 4:00")
@@ -94,7 +96,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)
@@ -113,7 +115,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()
issue.reload()
@@ -130,27 +132,29 @@ class TestIssue(TestSetUp):
create_communication(issue.name, "test@admin.com", "Sent", frappe.flags.current_time)
issue.reload()
issue.status = 'Replied'
issue.status = "Replied"
issue.save()
self.assertEqual(issue.on_hold_since, frappe.flags.current_time)
# close the issue after being on hold for 20 days
frappe.flags.current_time = get_datetime("2021-11-22 01:00")
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
self.assertEqual(issue.resolution_by, get_datetime('2021-11-22 06:00:00'))
self.assertEqual(issue.resolution_date, get_datetime('2021-11-22 01:00:00'))
self.assertEqual(issue.agreement_status, 'Fulfilled')
self.assertEqual(issue.resolution_by, get_datetime("2021-11-22 06:00:00"))
self.assertEqual(issue.resolution_date, get_datetime("2021-11-22 01:00:00"))
self.assertEqual(issue.agreement_status, "Fulfilled")
def test_issue_open_after_closed(self):
# Created on -> 1 pm, Response Time -> 4 hrs, Resolution Time -> 6 hrs
frappe.flags.current_time = get_datetime("2021-11-01 13:00")
issue = make_issue(frappe.flags.current_time, index=1, issue_type='Critical') # Applies 24hr working time SLA
issue = make_issue(
frappe.flags.current_time, index=1, issue_type="Critical"
) # Applies 24hr working time SLA
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
self.assertEquals(issue.agreement_status, 'First Response Due')
self.assertEquals(issue.agreement_status, "First Response Due")
self.assertEquals(issue.response_by, get_datetime("2021-11-01 17:00"))
self.assertEquals(issue.resolution_by, get_datetime("2021-11-01 19:00"))
@@ -158,9 +162,9 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2021-11-01 14:00")
create_communication(issue.name, "test@admin.com", "Sent", frappe.flags.current_time)
issue.reload()
issue.status = 'Replied'
issue.status = "Replied"
issue.save()
self.assertEquals(issue.agreement_status, 'Resolution Due')
self.assertEquals(issue.agreement_status, "Resolution Due")
self.assertEquals(issue.on_hold_since, frappe.flags.current_time)
self.assertEquals(issue.first_responded_on, frappe.flags.current_time)
@@ -168,7 +172,7 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2021-11-01 15:00")
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
issue.reload()
self.assertEquals(issue.status, 'Open')
self.assertEquals(issue.status, "Open")
# Hold Time + 1 Hrs
self.assertEquals(issue.total_hold_time, 3600)
# Resolution By should increase by one hrs
@@ -178,19 +182,19 @@ class TestIssue(TestSetUp):
frappe.flags.current_time = get_datetime("2021-11-01 16:00")
create_communication(issue.name, "test@admin.com", "Sent", frappe.flags.current_time)
issue.reload()
issue.status = 'Replied'
issue.status = "Replied"
issue.save()
self.assertEquals(issue.agreement_status, 'Resolution Due')
self.assertEquals(issue.agreement_status, "Resolution Due")
# Customer Closed → 10 pm
frappe.flags.current_time = get_datetime("2021-11-01 22:00")
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
# Hold Time + 6 Hrs
self.assertEquals(issue.total_hold_time, 3600 + 21600)
# Resolution By should increase by 6 hrs
self.assertEquals(issue.resolution_by, get_datetime("2021-11-02 02:00"))
self.assertEquals(issue.agreement_status, 'Fulfilled')
self.assertEquals(issue.agreement_status, "Fulfilled")
self.assertEquals(issue.resolution_date, frappe.flags.current_time)
# Customer Open → 3 am i.e after resolution by is crossed
@@ -201,15 +205,15 @@ class TestIssue(TestSetUp):
self.assertEquals(issue.total_hold_time, 3600 + 21600 + 18000)
# Resolution By should increase by 5 hrs
self.assertEquals(issue.resolution_by, get_datetime("2021-11-02 07:00"))
self.assertEquals(issue.agreement_status, 'Resolution Due')
self.assertEquals(issue.agreement_status, "Resolution Due")
self.assertFalse(issue.resolution_date)
# We Closed → 4 am, SLA should be Fulfilled
frappe.flags.current_time = get_datetime("2021-11-02 04:00")
issue.status = 'Closed'
issue.status = "Closed"
issue.save()
self.assertEquals(issue.resolution_by, get_datetime("2021-11-02 07:00"))
self.assertEquals(issue.agreement_status, 'Fulfilled')
self.assertEquals(issue.agreement_status, "Fulfilled")
self.assertEquals(issue.resolution_date, frappe.flags.current_time)
def test_recording_of_assignment_on_first_reponse_failure(self):
@@ -219,11 +223,7 @@ class TestIssue(TestSetUp):
issue = make_issue(frappe.flags.current_time, index=1)
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
add_assignment({
'doctype': issue.doctype,
'name': issue.name,
'assign_to': ['test@admin.com']
})
add_assignment({"doctype": issue.doctype, "name": issue.name, "assign_to": ["test@admin.com"]})
issue.reload()
# send a reply failing response SLA
@@ -232,12 +232,15 @@ class TestIssue(TestSetUp):
# assert if a new timeline item has been added
# to record the assignment
comment = frappe.db.exists('Comment', {
'reference_doctype': 'Issue',
'reference_name': issue.name,
'comment_type': 'Assigned',
'content': _('First Response SLA Failed by {}').format('test')
})
comment = frappe.db.exists(
"Comment",
{
"reference_doctype": "Issue",
"reference_name": issue.name,
"comment_type": "Assigned",
"content": _("First Response SLA Failed by {}").format("test"),
},
)
self.assertTrue(comment)
def test_agreement_status_on_response(self):
@@ -245,7 +248,7 @@ class TestIssue(TestSetUp):
issue = make_issue(frappe.flags.current_time, index=1)
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
self.assertTrue(issue.status == 'Open')
self.assertTrue(issue.status == "Open")
# send a reply within response SLA
frappe.flags.current_time = get_datetime("2021-11-02 11:00")
@@ -253,7 +256,8 @@ class TestIssue(TestSetUp):
issue.reload()
self.assertEquals(issue.first_responded_on, frappe.flags.current_time)
self.assertEquals(issue.agreement_status, 'Resolution Due')
self.assertEquals(issue.agreement_status, "Resolution Due")
class TestFirstResponseTime(TestSetUp):
# working hours used in all cases: Mon-Fri, 10am to 6pm
@@ -262,209 +266,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")
@@ -474,25 +537,28 @@ 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):
if issue_type and not frappe.db.exists('Issue Type', issue_type):
doc = frappe.new_doc('Issue Type')
if issue_type and not frappe.db.exists("Issue Type", issue_type):
doc = frappe.new_doc("Issue Type")
doc.name = issue_type
doc.insert()
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
@@ -503,45 +569,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 = 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()

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)

View File

@@ -42,16 +42,24 @@ class ServiceLevelAgreement(Document):
for priority in self.priorities:
# Check if response and resolution time is set for every priority
if not priority.response_time:
frappe.throw(_("Set Response Time for Priority {0} in row {1}.").format(priority.priority, priority.idx))
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))
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))
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)
@@ -74,9 +82,14 @@ class ServiceLevelAgreement(Document):
support_days.append(support_and_resolution.workday)
support_and_resolution.idx = week.index(support_and_resolution.workday) + 1
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))
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):
@@ -84,51 +97,76 @@ class ServiceLevelAgreement(Document):
frappe.throw(_("Workday {0} has been repeated.").format(repeated_days))
def validate_doc(self):
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.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 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 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 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 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)))
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_selected_doctype(self):
invalid_doctypes = list(frappe.model.core_doctypes_list)
invalid_doctypes.extend(['Cost Center', 'Company'])
valid_document_types = frappe.get_all('DocType', {
'issingle': 0,
'istable': 0,
'is_submittable': 0,
'name': ['not in', invalid_doctypes],
'module': ['not in', ["Email", "Core", "Custom", "Event Streaming", "Social", "Data Migration", "Geo", "Desk"]]
}, pluck="name")
invalid_doctypes.extend(["Cost Center", "Company"])
valid_document_types = frappe.get_all(
"DocType",
{
"issingle": 0,
"istable": 0,
"is_submittable": 0,
"name": ["not in", invalid_doctypes],
"module": [
"not in",
["Email", "Core", "Custom", "Event Streaming", "Social", "Data Migration", "Geo", "Desk"],
],
},
pluck="name",
)
if self.document_type not in valid_document_types:
frappe.throw(
msg=_("Please select valid document type."),
title=_("Invalid Document Type")
)
frappe.throw(msg=_("Please select valid document type."), title=_("Invalid Document Type"))
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)))
frappe.throw(
_(
"The Document Type {0} must have a Status field to configure Service Level Agreement"
).format(frappe.bold(self.document_type))
)
def validate_condition(self):
temp_doc = frappe.new_doc(self.document_type)
@@ -141,11 +179,13 @@ 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 before_insert(self):
# no need to set up SLA fields for Issue dt as they are standard fields in Issue
@@ -176,46 +216,50 @@ class ServiceLevelAgreement(Document):
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)
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)
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)
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)
@@ -236,57 +280,73 @@ class ServiceLevelAgreement(Document):
def check_agreement_status():
service_level_agreements = frappe.get_all("Service Level Agreement", filters=[
{"enabled": 1},
{"default_service_level_agreement": 0}
], fields=["name"])
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, "enabled", 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", "document_type", "=", doc.get('doctype')],
["Service Level Agreement", "enabled", "=", 1]
["Service Level Agreement", "document_type", "=", doc.get("doctype")],
["Service Level Agreement", "enabled", "=", 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")])
or_filters = []
if doc.get('service_level_agreement'):
if doc.get("service_level_agreement"):
or_filters = [
["Service Level Agreement", "name", "=", doc.get('service_level_agreement')],
["Service Level Agreement", "name", "=", doc.get("service_level_agreement")],
]
customer = doc.get('customer')
customer = doc.get("customer")
if customer:
or_filters.extend([
["Service Level Agreement", "entity", "in", [customer] + get_customer_group(customer) + get_customer_territory(customer)],
["Service Level Agreement", "entity_type", "is", "not set"]
])
else:
or_filters.append(
["Service Level Agreement", "entity_type", "is", "not set"]
or_filters.extend(
[
[
"Service Level Agreement",
"entity",
"in",
[customer] + get_customer_group(customer) + get_customer_territory(customer),
],
["Service Level Agreement", "entity_type", "is", "not set"],
]
)
else:
or_filters.append(["Service Level Agreement", "entity_type", "is", "not set"])
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", "apply_sla_for_resolution", "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", "apply_sla_for_resolution", "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", "apply_sla_for_resolution", "condition"])
agreements = frappe.get_all(
"Service Level Agreement",
filters=filters,
or_filters=or_filters,
fields=["name", "default_priority", "apply_sla_for_resolution", "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)
@@ -295,8 +355,14 @@ 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):
customer_groups = []
@@ -325,22 +391,33 @@ def get_service_level_agreement_filters(doctype, name, customer=None):
filters = [
["Service Level Agreement", "document_type", "=", doctype],
["Service Level Agreement", "enabled", "=", 1]
["Service Level Agreement", "enabled", "=", 1],
]
or_filters = [
["Service Level Agreement", "default_service_level_agreement", "=", 1]
]
or_filters = [["Service Level Agreement", "default_service_level_agreement", "=", 1]]
if customer:
# Include SLA with No Entity and Entity Type
or_filters.append(
["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),
]
)
return {
"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)]
"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)
],
}
@@ -366,7 +443,9 @@ def get_documents_with_active_service_level_agreement():
def set_documents_with_active_service_level_agreement():
active = [sla.document_type for sla in frappe.get_all("Service Level Agreement", fields=["document_type"])]
active = [
sla.document_type for sla in frappe.get_all("Service Level Agreement", fields=["document_type"])
]
frappe.cache().hset("service_level_agreement", "active", active)
return active
@@ -414,7 +493,7 @@ def process_sla(doc, sla):
def handle_status_change(doc, apply_sla_for_resolution):
now_time = frappe.flags.current_time or now_datetime(doc.get("owner"))
prev_status = frappe.db.get_value(doc.doctype, doc.name, 'status')
prev_status = frappe.db.get_value(doc.doctype, doc.name, "status")
hold_statuses = get_hold_statuses(doc.service_level_agreement)
fulfillment_statuses = get_fulfillment_statuses(doc.service_level_agreement)
@@ -429,9 +508,9 @@ def handle_status_change(doc, apply_sla_for_resolution):
return status not in hold_statuses and status not in fulfillment_statuses
def set_first_response():
if doc.meta.has_field("first_responded_on") and not doc.get('first_responded_on'):
if doc.meta.has_field("first_responded_on") and not doc.get("first_responded_on"):
doc.first_responded_on = now_time
if get_datetime(doc.get('first_responded_on')) > get_datetime(doc.get('response_by')):
if get_datetime(doc.get("first_responded_on")) > get_datetime(doc.get("response_by")):
record_assigned_users_on_failure(doc)
def calculate_hold_hours():
@@ -444,7 +523,7 @@ def handle_status_change(doc, apply_sla_for_resolution):
doc.total_hold_time = (doc.total_hold_time or 0) + current_hold_hours
doc.on_hold_since = None
if ((is_open_status(prev_status) and not is_open_status(doc.status)) or doc.flags.on_first_reply):
if (is_open_status(prev_status) and not is_open_status(doc.status)) or doc.flags.on_first_reply:
set_first_response()
# Open to Replied
@@ -492,22 +571,28 @@ def handle_status_change(doc, apply_sla_for_resolution):
def get_fulfillment_statuses(service_level_agreement):
return [entry.status for entry in frappe.db.get_all("SLA Fulfilled On Status", filters={
"parent": service_level_agreement
}, fields=["status"])]
return [
entry.status
for entry in frappe.db.get_all(
"SLA Fulfilled On Status", filters={"parent": service_level_agreement}, fields=["status"]
)
]
def get_hold_statuses(service_level_agreement):
return [entry.status for entry in frappe.db.get_all("Pause SLA On Status", filters={
"parent": service_level_agreement
}, fields=["status"])]
return [
entry.status
for entry in frappe.db.get_all(
"Pause SLA On Status", filters={"parent": service_level_agreement}, fields=["status"]
)
]
def update_response_and_resolution_metrics(doc, apply_sla_for_resolution):
priority = get_response_and_resolution_duration(doc)
start_date_time = get_datetime(doc.get("service_level_agreement_creation") or doc.creation)
set_response_by(doc, start_date_time, priority)
if apply_sla_for_resolution and not doc.get('on_hold_since'): # resolution_by is reset if on hold
if apply_sla_for_resolution and not doc.get("on_hold_since"): # resolution_by is reset if on hold
set_resolution_by(doc, start_date_time, priority)
@@ -526,9 +611,13 @@ 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:
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
):
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
@@ -572,10 +661,12 @@ def get_allotted_seconds(parameter, service_level):
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,
})
support_days[service.workday] = frappe._dict(
{
"start_time": service.start_time,
"end_time": service.end_time,
}
)
return support_days
@@ -588,15 +679,20 @@ def set_resolution_time(doc):
if not doc.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")
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 (
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)
@@ -606,25 +702,35 @@ def set_resolution_time(doc):
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)
)
def get_response_and_resolution_duration(doc):
sla = frappe.get_doc("Service Level Agreement", doc.service_level_agreement)
priority = sla.get_service_level_agreement_priority(doc.priority)
priority.update({
"support_and_resolution": sla.support_and_resolution,
"holiday_list": sla.holiday_list
})
priority.update(
{"support_and_resolution": sla.support_and_resolution, "holiday_list": sla.holiday_list}
)
return priority
@@ -632,14 +738,16 @@ 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)
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.save()
@@ -665,28 +773,30 @@ def on_communication_update(doc, status):
if not parent:
return
if not parent.meta.has_field('service_level_agreement'):
if not parent.meta.has_field("service_level_agreement"):
return
if (
doc.sent_or_received == "Received" # a reply is received
and parent.get('status') == 'Open' # issue status is set as open from communication.py
doc.sent_or_received == "Received" # a reply is received
and parent.get("status") == "Open" # issue status is set as open from communication.py
and parent.get_doc_before_save()
and parent.get('status') != parent._doc_before_save.get('status') # status changed
and parent.get("status") != parent._doc_before_save.get("status") # status changed
):
# undo the status change in db
# since prev status is fetched from db
frappe.db.set_value(
parent.doctype, parent.name,
'status', parent._doc_before_save.get('status'),
update_modified=False
parent.doctype,
parent.name,
"status",
parent._doc_before_save.get("status"),
update_modified=False,
)
elif (
doc.sent_or_received == "Sent" # a reply is sent
and parent.get('first_responded_on') # first_responded_on is set from communication.py
doc.sent_or_received == "Sent" # a reply is sent
and parent.get("first_responded_on") # first_responded_on is set from communication.py
and parent.get_doc_before_save()
and not parent._doc_before_save.get('first_responded_on') # first_responded_on was not set
and not parent._doc_before_save.get("first_responded_on") # first_responded_on was not set
):
# reset first_responded_on since it will be handled/set later on
parent.first_responded_on = None
@@ -695,7 +805,9 @@ def on_communication_update(doc, status):
else:
return
for_resolution = frappe.db.get_value('Service Level Agreement', parent.service_level_agreement, 'apply_sla_for_resolution')
for_resolution = frappe.db.get_value(
"Service Level Agreement", parent.service_level_agreement, "apply_sla_for_resolution"
)
handle_status_change(parent, for_resolution)
update_response_and_resolution_metrics(parent, for_resolution)
@@ -705,36 +817,42 @@ def on_communication_update(doc, status):
def reset_expected_response_and_resolution(doc):
if doc.meta.has_field("first_responded_on") and not doc.get('first_responded_on'):
if doc.meta.has_field("first_responded_on") and not doc.get("first_responded_on"):
doc.response_by = None
if doc.meta.has_field("resolution_by") and not doc.get('resolution_date'):
if doc.meta.has_field("resolution_by") and not doc.get("resolution_date"):
doc.resolution_by = None
def set_response_by(doc, start_date_time, priority):
if doc.meta.has_field("response_by"):
doc.response_by = get_expected_time_for(parameter="response", service_level=priority, start_date_time=start_date_time)
if doc.meta.has_field("total_hold_time") and doc.get('total_hold_time') and not doc.get('first_responded_on'):
doc.response_by = add_to_date(doc.response_by, seconds=round(doc.get('total_hold_time')))
doc.response_by = get_expected_time_for(
parameter="response", service_level=priority, start_date_time=start_date_time
)
if (
doc.meta.has_field("total_hold_time")
and doc.get("total_hold_time")
and not doc.get("first_responded_on")
):
doc.response_by = add_to_date(doc.response_by, seconds=round(doc.get("total_hold_time")))
def set_resolution_by(doc, start_date_time, priority):
if doc.meta.has_field("resolution_by"):
doc.resolution_by = get_expected_time_for(parameter="resolution", service_level=priority, start_date_time=start_date_time)
if doc.meta.has_field("total_hold_time") and doc.get('total_hold_time'):
doc.resolution_by = add_to_date(doc.resolution_by, seconds=round(doc.get('total_hold_time')))
doc.resolution_by = get_expected_time_for(
parameter="resolution", service_level=priority, start_date_time=start_date_time
)
if doc.meta.has_field("total_hold_time") and doc.get("total_hold_time"):
doc.resolution_by = add_to_date(doc.resolution_by, seconds=round(doc.get("total_hold_time")))
def record_assigned_users_on_failure(doc):
assigned_users = doc.get_assigned_users()
if assigned_users:
from frappe.utils import get_fullname
assigned_users = ', '.join((get_fullname(user) for user in assigned_users))
message = _('First Response SLA Failed by {}').format(assigned_users)
doc.add_comment(
comment_type='Assigned',
text=message
)
assigned_users = ", ".join((get_fullname(user) for user in assigned_users))
message = _("First Response SLA Failed by {}").format(assigned_users)
doc.add_comment(comment_type="Assigned", text=message)
def get_service_level_agreement_fields():
@@ -743,71 +861,57 @@ def get_service_level_agreement_fields():
"collapsible": 1,
"fieldname": "service_level_section",
"fieldtype": "Section Break",
"label": "Service Level"
"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
"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": "first_responded_on",
"fieldtype": "Datetime",
"label": "First Responded On",
"no_copy": 1,
"read_only": 1
"read_only": 1,
},
{
"fieldname": "on_hold_since",
"fieldtype": "Datetime",
"hidden": 1,
"label": "On Hold Since",
"read_only": 1
"read_only": 1,
},
{
"fieldname": "total_hold_time",
"fieldtype": "Duration",
"label": "Total Hold Time",
"read_only": 1
},
{
"fieldname": "cb",
"fieldtype": "Column Break",
"read_only": 1
"read_only": 1,
},
{"fieldname": "cb", "fieldtype": "Column Break", "read_only": 1},
{
"default": "First Response Due",
"fieldname": "agreement_status",
"fieldtype": "Select",
"label": "Service Level Agreement Status",
"options": "First Response Due\nResolution Due\nFulfilled\nFailed",
"read_only": 1
"read_only": 1,
},
{
"fieldname": "resolution_by",
"fieldtype": "Datetime",
"label": "Resolution By",
"read_only": 1
"read_only": 1,
},
{
"fieldname": "service_level_agreement_creation",
"fieldtype": "Datetime",
"hidden": 1,
"label": "Service Level Agreement Creation",
"read_only": 1
"read_only": 1,
},
{
"depends_on": "eval:!doc.__islocal",
@@ -815,8 +919,8 @@ def get_service_level_agreement_fields():
"fieldtype": "Datetime",
"label": "Resolution Date",
"no_copy": 1,
"read_only": 1
}
"read_only": 1,
},
]
@@ -826,21 +930,21 @@ def update_agreement_status_on_custom_status(doc):
def update_agreement_status(doc, apply_sla_for_resolution):
if (doc.meta.has_field("agreement_status")):
if doc.meta.has_field("agreement_status"):
# if SLA is applied for resolution check for response and resolution, else only response
if apply_sla_for_resolution:
if doc.meta.has_field("first_responded_on") and not doc.get('first_responded_on'):
if doc.meta.has_field("first_responded_on") and not doc.get("first_responded_on"):
doc.agreement_status = "First Response Due"
elif doc.meta.has_field("resolution_date") and not doc.get('resolution_date'):
elif doc.meta.has_field("resolution_date") and not doc.get("resolution_date"):
doc.agreement_status = "Resolution Due"
elif get_datetime(doc.get('resolution_date')) <= get_datetime(doc.get('resolution_by')):
elif get_datetime(doc.get("resolution_date")) <= get_datetime(doc.get("resolution_by")):
doc.agreement_status = "Fulfilled"
else:
doc.agreement_status = "Failed"
else:
if doc.meta.has_field("first_responded_on") and not doc.get('first_responded_on'):
if doc.meta.has_field("first_responded_on") and not doc.get("first_responded_on"):
doc.agreement_status = "First Response Due"
elif get_datetime(doc.get('first_responded_on')) <= get_datetime(doc.get('response_by')):
elif get_datetime(doc.get("first_responded_on")) <= get_datetime(doc.get("response_by")):
doc.agreement_status = "Fulfilled"
else:
doc.agreement_status = "Failed"
@@ -853,6 +957,7 @@ def is_holiday(date, 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)
@@ -865,7 +970,7 @@ def convert_utc_to_user_timezone(utc_timestamp, user):
from pytz import UnknownTimeZoneError, timezone
user_tz = get_tz(user)
utcnow = timezone('UTC').localize(utc_timestamp)
utcnow = timezone("UTC").localize(utc_timestamp)
try:
return utcnow.astimezone(timezone(user_tz))
except UnknownTimeZoneError:
@@ -884,11 +989,7 @@ def get_user_time(user, to_string=False):
@frappe.whitelist()
def get_sla_doctypes():
doctypes = []
data = frappe.get_all('Service Level Agreement',
{'enabled': 1},
['document_type'],
distinct=1
)
data = frappe.get_all("Service Level Agreement", {"enabled": 1}, ["document_type"], distinct=1)
for entry in data:
doctypes.append(entry.document_type)

View File

@@ -20,51 +20,122 @@ 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", 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",
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", 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",
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", 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",
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,
create_territory_service_level_agreement = create_service_level_agreement(
default_service_level_agreement=0,
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)
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 test_custom_field_creation_for_sla_on_standard_dt(self):
# Default Service Level Agreement
@@ -72,10 +143,12 @@ class TestServiceLevelAgreement(unittest.TestCase):
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,
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}]
sla_fulfilled_on=[{"status": "Converted"}],
)
# check default SLA for lead
@@ -86,27 +159,35 @@ class TestServiceLevelAgreement(unittest.TestCase):
sla_fields = get_service_level_agreement_fields()
for field in sla_fields:
self.assertTrue(frappe.db.exists("Custom Field", {"dt": doctype, "fieldname": field.get("fieldname")}))
self.assertTrue(
frappe.db.exists("Custom Field", {"dt": doctype, "fieldname": 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
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)
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()
for field in sla_fields:
self.assertTrue(frappe.db.exists("DocField", {"fieldname": field.get("fieldname"), "parent": doctype.name}))
self.assertTrue(
frappe.db.exists("DocField", {"fieldname": field.get("fieldname"), "parent": doctype.name})
)
def test_sla_application(self):
# Default Service Level Agreement
@@ -114,10 +195,12 @@ class TestServiceLevelAgreement(unittest.TestCase):
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,
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}]
sla_fulfilled_on=[{"status": "Converted"}],
)
# make lead with default SLA
@@ -130,21 +213,23 @@ class TestServiceLevelAgreement(unittest.TestCase):
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 15, 0)
lead.reload()
lead.status = 'Converted'
lead.status = "Converted"
lead.save()
self.assertEqual(lead.agreement_status, 'Fulfilled')
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,
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
doctype=doctype,
sla_fulfilled_on=[{"status": "Converted"}],
pause_sla_on=[{"status": "Replied"}]
pause_sla_on=[{"status": "Replied"}],
)
creation = datetime.datetime(2020, 3, 4, 4, 0)
@@ -152,7 +237,7 @@ class TestServiceLevelAgreement(unittest.TestCase):
frappe.flags.current_time = datetime.datetime(2020, 3, 4, 4, 15)
lead.reload()
lead.status = 'Replied'
lead.status = "Replied"
lead.save()
lead.reload()
@@ -160,7 +245,7 @@ class TestServiceLevelAgreement(unittest.TestCase):
frappe.flags.current_time = datetime.datetime(2020, 3, 4, 5, 5)
lead.reload()
lead.status = 'Converted'
lead.status = "Converted"
lead.save()
lead.reload()
@@ -172,12 +257,13 @@ class TestServiceLevelAgreement(unittest.TestCase):
create_service_level_agreement(
default_service_level_agreement=1,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
entity_type=None,
entity=None,
response_time=14400,
doctype=doctype,
sla_fulfilled_on=[{"status": "Replied"}],
pause_sla_on=[],
apply_sla_for_resolution=0
apply_sla_for_resolution=0,
)
creation = datetime.datetime(2019, 3, 4, 12, 0)
@@ -187,22 +273,23 @@ class TestServiceLevelAgreement(unittest.TestCase):
# failed with response time only
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 16, 5)
lead.reload()
lead.status = 'Replied'
lead.status = "Replied"
lead.save()
lead.reload()
self.assertEqual(lead.agreement_status, 'Failed')
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,
entity_type=None,
entity=None,
response_time=14400,
doctype=doctype,
sla_fulfilled_on=[{"status": "Replied"}],
apply_sla_for_resolution=0
apply_sla_for_resolution=0,
)
# fulfilled with response time only
@@ -214,11 +301,11 @@ class TestServiceLevelAgreement(unittest.TestCase):
frappe.flags.current_time = datetime.datetime(2019, 3, 4, 15, 30)
lead.reload()
lead.status = 'Replied'
lead.status = "Replied"
lead.save()
lead.reload()
self.assertEqual(lead.agreement_status, 'Fulfilled')
self.assertEqual(lead.agreement_status, "Fulfilled")
def test_service_level_agreement_filters(self):
doctype = "Lead"
@@ -226,29 +313,30 @@ class TestServiceLevelAgreement(unittest.TestCase):
default_service_level_agreement=0,
doctype=doctype,
holiday_list="__Test Holiday List",
entity_type=None, entity=None,
entity_type=None,
entity=None,
condition='doc.source == "Test Source"',
response_time=14400,
sla_fulfilled_on=[{"status": "Replied"}],
apply_sla_for_resolution=0
apply_sla_for_resolution=0,
)
creation = datetime.datetime(2019, 3, 4, 12, 0)
lead = make_lead(creation=creation, index=4)
applied_sla = frappe.db.get_value('Lead', lead.name, 'service_level_agreement')
applied_sla = frappe.db.get_value("Lead", lead.name, "service_level_agreement")
self.assertFalse(applied_sla)
source = frappe.get_doc(doctype='Lead Source', source_name='Test Source')
source = frappe.get_doc(doctype="Lead Source", source_name="Test Source")
source.insert(ignore_if_duplicate=True)
lead.source = "Test Source"
lead.save()
applied_sla = frappe.db.get_value('Lead', lead.name, 'service_level_agreement')
applied_sla = frappe.db.get_value("Lead", lead.name, "service_level_agreement")
self.assertEqual(applied_sla, lead_sla.name)
# check if SLA is removed if condition fails
lead.reload()
lead.source = None
lead.save()
applied_sla = frappe.db.get_value('Lead', lead.name, 'service_level_agreement')
applied_sla = frappe.db.get_value("Lead", lead.name, "service_level_agreement")
self.assertFalse(applied_sla)
def tearDown(self):
@@ -256,130 +344,150 @@ class TestServiceLevelAgreement(unittest.TestCase):
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"):
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, "document_type": doctype}
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, response_time, entity_type,
entity, resolution_time=0, doctype="Issue", condition="", sla_fulfilled_on=[], pause_sla_on=[], apply_sla_for_resolution=1,
service_level=None, start_time="10:00:00", end_time="18:00:00"):
def create_service_level_agreement(
default_service_level_agreement,
holiday_list,
response_time,
entity_type,
entity,
resolution_time=0,
doctype="Issue",
condition="",
sla_fulfilled_on=[],
pause_sla_on=[],
apply_sla_for_resolution=1,
service_level=None,
start_time="10:00:00",
end_time="18:00:00",
):
make_holiday_list()
make_priorities()
if not sla_fulfilled_on:
sla_fulfilled_on = [
{"status": "Resolved"},
{"status": "Closed"}
]
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",
"enabled": 1,
"document_type": doctype,
"service_level": service_level or "__Test {} SLA".format(entity_type if entity_type else "Default"),
"default_service_level_agreement": default_service_level_agreement,
"condition": condition,
"default_priority": "Medium",
"holiday_list": holiday_list,
"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,
"resolution_time": resolution_time,
},
{
"priority": "Medium",
"response_time": response_time,
"default_priority": 1,
"resolution_time": resolution_time,
},
{
"priority": "High",
"response_time": response_time,
"resolution_time": resolution_time,
}
],
"sla_fulfilled_on": sla_fulfilled_on,
"pause_sla_on": pause_sla_on,
"support_and_resolution": [
{
"workday": "Monday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Tuesday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Wednesday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Thursday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Friday",
"start_time": start_time,
"end_time": end_time,
}
]
})
service_level_agreement = frappe._dict(
{
"doctype": "Service Level Agreement",
"enabled": 1,
"document_type": doctype,
"service_level": service_level
or "__Test {} SLA".format(entity_type if entity_type else "Default"),
"default_service_level_agreement": default_service_level_agreement,
"condition": condition,
"default_priority": "Medium",
"holiday_list": holiday_list,
"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,
"resolution_time": resolution_time,
},
{
"priority": "Medium",
"response_time": response_time,
"default_priority": 1,
"resolution_time": resolution_time,
},
{
"priority": "High",
"response_time": response_time,
"resolution_time": resolution_time,
},
],
"sla_fulfilled_on": sla_fulfilled_on,
"pause_sla_on": pause_sla_on,
"support_and_resolution": [
{
"workday": "Monday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Tuesday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Wednesday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Thursday",
"start_time": start_time,
"end_time": end_time,
},
{
"workday": "Friday",
"start_time": start_time,
"end_time": end_time,
},
],
}
)
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})
sla = frappe.db.exists("Service Level Agreement", filters)
if sla:
frappe.delete_doc("Service Level Agreement", sla, force=1)
return frappe.get_doc(service_level_agreement).insert(ignore_permissions=True, ignore_if_duplicate=True)
return frappe.get_doc(service_level_agreement).insert(
ignore_permissions=True, ignore_if_duplicate=True
)
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()
@@ -387,11 +495,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()
@@ -399,102 +510,116 @@ 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",
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",
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",
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",
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",
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",
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",
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",
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",
entity_type=None, entity=None, response_time=14400, resolution_time=21600,
service_level="24-hour-SLA", start_time="00:00:00", end_time="23:59:59",
condition="doc.issue_type == 'Critical'"
default_service_level_agreement=0,
holiday_list="__Test Holiday List",
entity_type=None,
entity=None,
response_time=14400,
resolution_time=21600,
service_level="24-hour-SLA",
start_time="00:00:00",
end_time="23:59:59",
condition="doc.issue_type == 'Critical'",
)
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()
def create_custom_doctype():
if not frappe.db.exists("DocType", "Test SLA on Custom Dt"):
doc = frappe.get_doc({
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": "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"
}
"options": "Open\nReplied\nClosed",
},
],
"permissions": [{
"role": "System Manager",
"read": 1,
"write": 1
}],
"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,
"priority": "Medium"
}).insert(ignore_permissions=True)
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,
"priority": "Medium",
}
).insert(ignore_permissions=True)

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

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

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

View File

@@ -14,6 +14,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"""
@@ -30,101 +31,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)
@@ -141,25 +139,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)
@@ -170,14 +169,14 @@ class IssueAnalytics(object):
self.get_periodic_data()
for entity, period_data in self.issue_periodic_data.items():
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:
@@ -186,7 +185,7 @@ class IssueAnalytics(object):
row[scrub(period)] = amount
total += amount
row['total'] = total
row["total"] = total
self.data.append(row)
@@ -194,9 +193,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)
@@ -206,18 +205,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"}

View File

@@ -10,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
@@ -23,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()
@@ -38,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():
@@ -189,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}
)

View File

@@ -12,6 +12,7 @@ from frappe.utils import flt
def execute(filters=None):
return IssueSummary(filters).run()
class IssueSummary(object):
def __init__(self, filters=None):
self.filters = frappe._dict(filters or {})
@@ -27,83 +28,78 @@ 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',
'First Response Due': 'first_response_due',
'Resolution Due': 'resolution_due'
"SLA Failed": "failed",
"SLA Fulfilled": "fulfilled",
"First Response Due": "first_response_due",
"Resolution Due": "resolution_due",
}
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 self.issue_summary_data.items():
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",
},
]

View File

@@ -7,34 +7,36 @@ from frappe import _
from frappe.utils import add_to_date, get_datetime, getdate
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 time_slots.items():
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)
@@ -47,49 +49,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