refactor(treewide): formatting and ruff fixes, + manually enabled F401

Signed-off-by: Akhil Narang <me@akhilnarang.dev>
This commit is contained in:
Akhil Narang
2024-03-27 11:37:26 +05:30
parent 8afb7790de
commit 3effaf21ef
574 changed files with 4135 additions and 6276 deletions

View File

@@ -163,7 +163,7 @@ class Issue(Document):
"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(
"content": " - Split the Issue from <a href='/app/Form/Issue/{}'>{}</a>".format(
self.name, frappe.bold(self.name)
),
}
@@ -217,7 +217,6 @@ def get_issue_list(doctype, txt, filters, limit_start, limit_page_length=20, ord
@frappe.whitelist()
def set_multiple_status(names, status):
for name in json.loads(names):
frappe.db.set_value("Issue", name, "status", status)
@@ -229,9 +228,7 @@ def set_status(name, status):
def auto_close_tickets():
"""Auto-close replied support tickets after 7 days"""
auto_close_after_days = (
frappe.db.get_single_value("Support Settings", "close_issue_after_days") or 7
)
auto_close_after_days = frappe.db.get_single_value("Support Settings", "close_issue_after_days") or 7
table = frappe.qb.DocType("Issue")
issues = (
@@ -299,9 +296,7 @@ def set_first_response_time(communication, method):
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)

View File

@@ -29,23 +29,21 @@ class TestIssue(TestSetUp):
creation = get_datetime("2019-03-04 12:00")
# make issue with customer specific SLA
customer = create_customer("_Test Customer", "__Test SLA Customer Group", "__Test SLA Territory")
create_customer("_Test Customer", "__Test SLA Customer Group", "__Test SLA Territory")
issue = make_issue(creation, "_Test Customer", 1)
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 customer_group specific SLA
customer = create_customer("__Test Customer", "_Test SLA Customer Group", "__Test SLA Territory")
create_customer("__Test Customer", "_Test SLA Customer Group", "__Test SLA Territory")
issue = make_issue(creation, "__Test Customer", 2)
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"
)
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"))
@@ -147,16 +145,15 @@ class TestIssue(TestSetUp):
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
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
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"))
self.assertEqual(issue.agreement_status, "First Response Due")
self.assertEqual(issue.response_by, get_datetime("2021-11-01 17:00"))
self.assertEqual(issue.resolution_by, get_datetime("2021-11-01 19:00"))
# Replied on → 2 pm
frappe.flags.current_time = get_datetime("2021-11-01 14:00")
@@ -164,19 +161,19 @@ class TestIssue(TestSetUp):
issue.reload()
issue.status = "Replied"
issue.save()
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)
self.assertEqual(issue.agreement_status, "Resolution Due")
self.assertEqual(issue.on_hold_since, frappe.flags.current_time)
self.assertEqual(issue.first_responded_on, frappe.flags.current_time)
# Customer Replied → 3 pm
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.assertEqual(issue.status, "Open")
# Hold Time + 1 Hrs
self.assertEquals(issue.total_hold_time, 3600)
self.assertEqual(issue.total_hold_time, 3600)
# Resolution By should increase by one hrs
self.assertEquals(issue.resolution_by, get_datetime("2021-11-01 20:00"))
self.assertEqual(issue.resolution_by, get_datetime("2021-11-01 20:00"))
# Replied on → 4 pm, Open → 1 hr, Resolution Due → 8 pm
frappe.flags.current_time = get_datetime("2021-11-01 16:00")
@@ -184,37 +181,37 @@ class TestIssue(TestSetUp):
issue.reload()
issue.status = "Replied"
issue.save()
self.assertEquals(issue.agreement_status, "Resolution Due")
self.assertEqual(issue.agreement_status, "Resolution Due")
# Customer Closed → 10 pm
frappe.flags.current_time = get_datetime("2021-11-01 22:00")
issue.status = "Closed"
issue.save()
# Hold Time + 6 Hrs
self.assertEquals(issue.total_hold_time, 3600 + 21600)
self.assertEqual(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.resolution_date, frappe.flags.current_time)
self.assertEqual(issue.resolution_by, get_datetime("2021-11-02 02:00"))
self.assertEqual(issue.agreement_status, "Fulfilled")
self.assertEqual(issue.resolution_date, frappe.flags.current_time)
# Customer Open → 3 am i.e after resolution by is crossed
frappe.flags.current_time = get_datetime("2021-11-02 03:00")
create_communication(issue.name, "test@example.com", "Received", frappe.flags.current_time)
issue.reload()
# Since issue was Resolved, Resolution By should be increased by 5 hrs (3am - 10pm)
self.assertEquals(issue.total_hold_time, 3600 + 21600 + 18000)
self.assertEqual(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.assertEqual(issue.resolution_by, get_datetime("2021-11-02 07:00"))
self.assertEqual(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.save()
self.assertEquals(issue.resolution_by, get_datetime("2021-11-02 07:00"))
self.assertEquals(issue.agreement_status, "Fulfilled")
self.assertEquals(issue.resolution_date, frappe.flags.current_time)
self.assertEqual(issue.resolution_by, get_datetime("2021-11-02 07:00"))
self.assertEqual(issue.agreement_status, "Fulfilled")
self.assertEqual(issue.resolution_date, frappe.flags.current_time)
def test_recording_of_assignment_on_first_reponse_failure(self):
from frappe.desk.form.assign_to import add as add_assignment
@@ -255,8 +252,8 @@ class TestIssue(TestSetUp):
create_communication(issue.name, "test@admin.com", "Sent", frappe.flags.current_time)
issue.reload()
self.assertEquals(issue.first_responded_on, frappe.flags.current_time)
self.assertEquals(issue.agreement_status, "Resolution Due")
self.assertEqual(issue.first_responded_on, frappe.flags.current_time)
self.assertEqual(issue.agreement_status, "Resolution Due")
class TestFirstResponseTime(TestSetUp):
@@ -547,7 +544,7 @@ def make_issue(creation=None, customer=None, index=0, priority=None, issue_type=
issue = frappe.get_doc(
{
"doctype": "Issue",
"subject": "Service Level Agreement Issue {0}".format(index),
"subject": f"Service Level Agreement Issue {index}",
"customer": customer,
"raised_by": "test@example.com",
"description": "Service Level Agreement Issue",
@@ -564,7 +561,6 @@ def make_issue(creation=None, customer=None, index=0, priority=None, issue_type=
def create_customer(name, customer_group, territory):
create_customer_group(customer_group)
create_territory(territory)
@@ -580,7 +576,6 @@ def create_customer(name, customer_group, territory):
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
@@ -588,7 +583,6 @@ def create_customer_group(customer_group):
def create_territory(territory):
if not frappe.db.exists("Territory", {"territory_name": territory}):
frappe.get_doc(
{

View File

@@ -79,22 +79,26 @@ class ServiceLevelAgreement(Document):
# 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)
_("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)
_("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
)
_(
"Response Time for {0} priority in row {1} can't be greater than Resolution Time."
).format(priority.priority, priority.idx)
)
priorities.append(priority.priority)
@@ -358,7 +362,7 @@ def get_active_service_level_agreement_for(doc):
"Service Level Agreement",
"entity",
"in",
[customer] + get_customer_group(customer) + get_customer_territory(customer),
[customer, *get_customer_group(customer), *get_customer_territory(customer)],
],
["Service Level Agreement", "entity_type", "is", "not set"],
]
@@ -366,9 +370,7 @@ def get_active_service_level_agreement_for(doc):
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_filter = [*filters, ["Service Level Agreement", "default_service_level_agreement", "=", 1]]
default_sla = frappe.get_all(
"Service Level Agreement",
filters=default_sla_filter,
@@ -409,7 +411,7 @@ def get_customer_group(customer):
customer_group = frappe.db.get_value("Customer", customer, "customer_group") if customer else None
if customer_group:
ancestors = get_ancestors_of("Customer Group", customer_group)
customer_groups = [customer_group] + ancestors
customer_groups = [customer_group, *ancestors]
return customer_groups
@@ -419,7 +421,7 @@ def get_customer_territory(customer):
customer_territory = frappe.db.get_value("Customer", customer, "territory") if customer else None
if customer_territory:
ancestors = get_ancestors_of("Territory", customer_territory)
customer_territories = [customer_territory] + ancestors
customer_territories = [customer_territory, *ancestors]
return customer_territories
@@ -443,7 +445,7 @@ def get_service_level_agreement_filters(doctype, name, customer=None):
"Service Level Agreement",
"entity",
"in",
[""] + [customer] + get_customer_group(customer) + get_customer_territory(customer),
["", customer, *get_customer_group(customer), *get_customer_territory(customer)],
]
)
@@ -455,8 +457,7 @@ def get_service_level_agreement_filters(doctype, name, customer=None):
)
],
"service_level_agreements": [
d.name
for d in frappe.get_all("Service Level Agreement", filters=filters, or_filters=or_filters)
d.name for d in frappe.get_all("Service Level Agreement", filters=filters, or_filters=or_filters)
],
}
@@ -517,7 +518,6 @@ def remove_sla_if_applied(doc):
def process_sla(doc, sla):
if not doc.creation:
doc.creation = now_datetime(doc.get("owner"))
if doc.meta.has_field("service_level_agreement_creation"):
@@ -747,16 +747,13 @@ def change_service_level_agreement_and_priority(self):
and frappe.db.exists("Issue", self.name)
and frappe.db.get_single_value("Support Settings", "track_service_level_agreement")
):
if self.priority != frappe.db.get_value("Issue", self.name, "priority"):
self.set_response_and_resolution_time(
priority=self.priority, service_level_agreement=self.service_level_agreement
)
frappe.msgprint(_("Priority has been changed to {0}.").format(self.priority))
if self.service_level_agreement != frappe.db.get_value(
"Issue", self.name, "service_level_agreement"
):
if 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
)
@@ -768,9 +765,7 @@ def change_service_level_agreement_and_priority(self):
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
@@ -787,7 +782,7 @@ def reset_service_level_agreement(doctype: str, docname: str, reason, user):
"reference_doctype": doc.doctype,
"reference_name": doc.name,
"comment_email": user,
"content": " resetted Service Level Agreement - {0}".format(_(reason)),
"content": f" resetted Service Level Agreement - {_(reason)}",
}
).insert(ignore_permissions=True)
@@ -895,7 +890,7 @@ def record_assigned_users_on_failure(doc):
if assigned_users:
from frappe.utils import get_fullname
assigned_users = ", ".join((get_fullname(user) for user in assigned_users))
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)

View File

@@ -51,9 +51,7 @@ class TestServiceLevelAgreement(unittest.TestCase):
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
@@ -198,9 +196,7 @@ class TestServiceLevelAgreement(unittest.TestCase):
)
# 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
@@ -390,14 +386,17 @@ def create_service_level_agreement(
resolution_time=0,
doctype="Issue",
condition="",
sla_fulfilled_on=[],
pause_sla_on=[],
sla_fulfilled_on=None,
pause_sla_on=None,
apply_sla_for_resolution=1,
service_level=None,
start_time="10:00:00",
end_time="18:00:00",
):
if pause_sla_on is None:
pause_sla_on = []
if sla_fulfilled_on is None:
sla_fulfilled_on = []
make_holiday_list()
make_priorities()
@@ -484,9 +483,7 @@ def create_service_level_agreement(
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():
@@ -615,7 +612,12 @@ def create_custom_doctype():
"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": "Email ID",
"fieldname": "email_id",
"fieldtype": "Link",
"options": "Customer",
},
{
"label": "Status",
"fieldname": "status",
@@ -637,8 +639,8 @@ def make_lead(creation=None, index=0, company=None):
return frappe.get_doc(
{
"doctype": "Lead",
"email_id": "test_lead1@example{0}.com".format(index),
"lead_name": "_Test Lead {0}".format(index),
"email_id": f"test_lead1@example{index}.com",
"lead_name": f"_Test Lead {index}",
"status": "Open",
"creation": creation,
"service_level_agreement_creation": creation,

View File

@@ -45,9 +45,7 @@ class WarrantyClaim(TransactionBase):
service_address: DF.SmallText | None
status: DF.Literal["", "Open", "Closed", "Work In Progress", "Cancelled"]
territory: DF.Link | None
warranty_amc_status: DF.Literal[
"", "Under Warranty", "Out of Warranty", "Under AMC", "Out of AMC"
]
warranty_amc_status: DF.Literal["", "Under Warranty", "Out of Warranty", "Under AMC", "Out of AMC"]
warranty_expiry_date: DF.Date | None
# end: auto-generated types

View File

@@ -15,7 +15,7 @@ def execute(filters=None):
return IssueAnalytics(filters).run()
class IssueAnalytics(object):
class IssueAnalytics:
def __init__(self, filters=None):
"""Issue Analytics Report"""
self.filters = frappe._dict(filters or {})
@@ -44,7 +44,13 @@ class IssueAnalytics(object):
elif self.filters.based_on == "Assigned To":
self.columns.append(
{"label": _("User"), "fieldname": "user", "fieldtype": "Link", "options": "User", "width": 200}
{
"label": _("User"),
"fieldname": "user",
"fieldtype": "Link",
"options": "User",
"width": 200,
}
)
elif self.filters.based_on == "Issue Type":
@@ -75,9 +81,7 @@ class IssueAnalytics(object):
{"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()
@@ -109,9 +113,7 @@ class IssueAnalytics(object):
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"]:
from_date = from_date.replace(day=1)
@@ -121,7 +123,7 @@ class IssueAnalytics(object):
from_date = from_date + relativedelta(from_date, weekday=MO(-1))
self.periodic_daterange = []
for dummy in range(1, 53):
for _dummy in range(1, 53):
if self.filters.range == "Weekly":
period_end_date = add_days(from_date, 6)
else:

View File

@@ -13,7 +13,7 @@ def execute(filters=None):
return IssueSummary(filters).run()
class IssueSummary(object):
class IssueSummary:
def __init__(self, filters=None):
self.filters = frappe._dict(filters or {})
@@ -41,7 +41,13 @@ class IssueSummary(object):
elif self.filters.based_on == "Assigned To":
self.columns.append(
{"label": _("User"), "fieldname": "user", "fieldtype": "Link", "options": "User", "width": 200}
{
"label": _("User"),
"fieldname": "user",
"fieldtype": "Link",
"options": "User",
"width": 200,
}
)
elif self.filters.based_on == "Issue Type":
@@ -84,9 +90,7 @@ class IssueSummary(object):
}
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",
@@ -185,8 +189,12 @@ class IssueSummary(object):
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(
agreement_status, 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
@@ -229,14 +237,20 @@ class IssueSummary(object):
if d._assign:
for entry in json.loads(d._assign):
for metric in metrics_list:
self.issue_summary_data.setdefault(entry, frappe._dict()).setdefault(metric, 0.0)
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_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_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
)
@@ -251,9 +265,9 @@ class IssueSummary(object):
else:
data = frappe.db.sql(
"""
f"""
SELECT
{0}, AVG(first_response_time) as avg_frt,
{field}, AVG(first_response_time) as avg_frt,
AVG(avg_response_time) as avg_resp_time,
AVG(total_hold_time) as avg_hold_time,
AVG(resolution_time) as avg_resolution_time,
@@ -261,10 +275,8 @@ class IssueSummary(object):
FROM `tabIssue`
WHERE
name IN %(issues)s
GROUP BY {0}
""".format(
field
),
GROUP BY {field}
""",
{"issues": issues},
as_dict=1,
)

View File

@@ -37,8 +37,8 @@ def get_data(filters):
hours_count = {"date": start_date}
for key, value in time_slots.items():
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))
start_time = get_datetime("{} {}".format(start_date.strftime("%Y-%m-%d"), start_time))
end_time = get_datetime("{} {}".format(start_date.strftime("%Y-%m-%d"), end_time))
hours_count[key] = get_hours_count(start_time, end_time)
time_slot_wise_total_count[key] = time_slot_wise_total_count.get(key, 0) + hours_count[key]