c52e5157b4
* feat: allow users to email prepared report * fix: add export in background checkbox in export dialog * chore: use a common function to send report email * feat: delete generated files from system after user defined duration * feat: add exported reports in a folder and periodically clean it * test: add test for export via report * fix: add exported folder path as constant * chore: un-edit report.json * refactor: for readability * chore: type hint * refactor: let the BG job fail if email fails * refactor: consistent `report_name` * it reads better now, e.g. attached_to_name=report_name instead of attached_to_name=title * refactor: `return_file` to its inverse `populate_response` * chore: more specific error message --------- Co-authored-by: Rahul Agrawal <deathstarconsole@Rahuls-MacBook-Air.local> Co-authored-by: Hussain Nagaria <hussainbhaitech@gmail.com>
110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
# Copyright (c) 2019, Frappe Technologies Pvt. Ltd. and Contributors
|
|
# License: MIT. See LICENSE
|
|
|
|
import frappe
|
|
from frappe.desk.reportview import export_query, extract_fieldnames
|
|
from frappe.tests import IntegrationTestCase
|
|
|
|
|
|
class TestReportview(IntegrationTestCase):
|
|
def test_csv(self):
|
|
from csv import QUOTE_ALL, QUOTE_MINIMAL, QUOTE_NONE, QUOTE_NONNUMERIC, DictReader
|
|
from io import StringIO
|
|
|
|
frappe.local.form_dict = frappe._dict(
|
|
doctype="DocType",
|
|
file_format_type="CSV",
|
|
fields=("name", "module", "issingle"),
|
|
filters={"issingle": 1, "module": "Core"},
|
|
)
|
|
|
|
for delimiter in (",", ";", "\t", "|"):
|
|
frappe.local.form_dict.csv_delimiter = delimiter
|
|
for quoting in (QUOTE_ALL, QUOTE_MINIMAL, QUOTE_NONE, QUOTE_NONNUMERIC):
|
|
frappe.local.form_dict.csv_quoting = quoting
|
|
|
|
export_query()
|
|
|
|
self.assertTrue(frappe.response["filename"].endswith(".csv"))
|
|
self.assertEqual(frappe.response["type"], "binary")
|
|
with StringIO(frappe.response["filecontent"].decode("utf-8")) as result:
|
|
reader = DictReader(result, delimiter=delimiter, quoting=quoting)
|
|
for row in reader:
|
|
self.assertEqual(int(row["Is Single"]), 1)
|
|
self.assertEqual(row["Module"], "Core")
|
|
|
|
def test_extract_fieldname(self):
|
|
self.assertEqual(
|
|
extract_fieldnames("count(distinct `tabPhoto`.name) as total_count")[0], "tabPhoto.name"
|
|
)
|
|
|
|
self.assertEqual(extract_fieldnames("owner")[0], "owner")
|
|
self.assertEqual(extract_fieldnames("from")[0], "from")
|
|
|
|
self.assertEqual(extract_fieldnames("module")[0], "module")
|
|
|
|
self.assertEqual(extract_fieldnames("count(`tabPhoto`.name) as total_count")[0], "tabPhoto.name")
|
|
|
|
self.assertEqual(extract_fieldnames("count(distinct `tabPhoto`.name)")[0], "tabPhoto.name")
|
|
|
|
self.assertEqual(extract_fieldnames("count(`tabPhoto`.name)")[0], "tabPhoto.name")
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("count(distinct `tabJob Applicant`.name) as total_count")[0],
|
|
"tabJob Applicant.name",
|
|
)
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("(1 / nullif(locate('a', `tabAddress`.`name`), 0)) as `_relevance`")[0],
|
|
"tabAddress.name",
|
|
)
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("(1 / nullif(locate('(a)', `tabAddress`.`name`), 0)) as `_relevance`")[0],
|
|
"tabAddress.name",
|
|
)
|
|
|
|
self.assertEqual(extract_fieldnames("EXTRACT(MONTH FROM date_column) AS month")[0], "date_column")
|
|
|
|
self.assertEqual(extract_fieldnames("COUNT(*) AS count")[0], "*")
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("first_name + ' ' + last_name AS full_name"), ["first_name", "last_name"]
|
|
)
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("CONCAT(first_name, ' ', last_name) AS full_name"),
|
|
["first_name", "last_name"],
|
|
)
|
|
|
|
self.assertEqual(
|
|
extract_fieldnames("CONCAT(id, '/', name, '/', age, '/', marks) AS student"),
|
|
["id", "name", "age", "marks"],
|
|
)
|
|
|
|
self.assertEqual(extract_fieldnames("tablefield.fiedname")[0], "tablefield.fiedname")
|
|
|
|
self.assertEqual(extract_fieldnames("`tabChild DocType`.`fiedname`")[0], "tabChild DocType.fiedname")
|
|
|
|
self.assertEqual(extract_fieldnames("sum(1)"), [])
|
|
|
|
def test_export_report_via_email(self):
|
|
frappe.local.form_dict = frappe._dict(
|
|
doctype="DocType",
|
|
file_format_type="CSV",
|
|
fields=("name", "module", "issingle"),
|
|
filters={"issingle": 1, "module": "Core"},
|
|
export_in_background=1,
|
|
)
|
|
|
|
frappe.db.delete("Email Queue")
|
|
export_query()
|
|
jobs = frappe.get_all(
|
|
"RQ Job",
|
|
filters={"job_name": "frappe.desk.query_report.run_report_view_export_job"},
|
|
fields=["name", "status"],
|
|
)
|
|
email_queue = frappe.get_all("Email Queue")
|
|
|
|
self.assertTrue(jobs, "Background job was not enqueued")
|
|
self.assertTrue(email_queue, "Email was not enqueued")
|