Merge pull request #17968 from jmchilton/error_reporting_unit_tests

Error reporting unit tests

lib/galaxy/tools/error_reports/plugins/email.py

@@ -35,7 +35,9 @@ def submit_report(self, dataset, job, tool, **kwargs):
             )
             return ("Your error report has been sent", "success")
         except Exception as e:
-            return (f"An error occurred sending the report by email: {unicodify(e)}", "danger")
+            msg = f"An error occurred sending the report by email: {unicodify(e)}"
+            log.exception(msg)
+            return (msg, "danger")
 
 
 __all__ = ("EmailPlugin",)

lib/galaxy/util/__init__.py

@@ -644,12 +644,6 @@ def pretty_print_time_interval(time=False, precise=False, utc=False):
         return "a few years ago"
 
 
-def pretty_print_json(json_data, is_json_string=False):
-    if is_json_string:
-        json_data = json.loads(json_data)
-    return json.dumps(json_data, sort_keys=True, indent=4)
-
-
 # characters that are valid
 valid_chars = set(string.ascii_letters + string.digits + " -=_.()/+*^,:?!")
 
@@ -1621,6 +1615,21 @@ def send_mail(frm, to, subject, body, config, html=None, reply_to=None):
     :type  reply_to: str
     :param reply_to: Reply-to address (Default None)
     """
+    smtp_server = config.smtp_server
+    if smtp_server and isinstance(smtp_server, str) and smtp_server.startswith("mock_emails_to_path://"):
+        path = config.smtp_server[len("mock_emails_to_path://") :]
+        email_dict = {
+            "from": frm,
+            "to": to,
+            "subject": subject,
+            "body": body,
+            "html": html,
+            "reply_to": reply_to,
+        }
+        email_json = json.to_json_string(email_dict)
+        with open(path, "w") as f:
+            f.write(email_json)
+        return
 
     to = listify(to)
     if html:

lib/galaxy_test/base/populators.py

@@ -1048,6 +1048,36 @@ def new_error_dataset(self, history_id: str) -> str:
         assert output_details["state"] == "error", output_details
         return output_details["id"]
 
+    def report_job_error_raw(
+        self, job_id: str, dataset_id: str, message: str = "", email: Optional[str] = None
+    ) -> Response:
+        url = f"jobs/{job_id}/error"
+        payload = dict(
+            dataset_id=dataset_id,
+            message=message,
+        )
+        if email is not None:
+            payload["email"] = email
+        report_response = self._post(url, data=payload, json=True)
+        return report_response
+
+    def report_job_error(
+        self, job_id: str, dataset_id: str, message: str = "", email: Optional[str] = None
+    ) -> Response:
+        report_response = self.report_job_error_raw(job_id, dataset_id, message=message, email=email)
+        api_asserts.assert_status_code_is_ok(report_response)
+        return report_response.json()
+
+    def run_detect_errors(self, history_id: str, exit_code: int, stdout: str = "", stderr: str = "") -> dict:
+        inputs = {
+            "stdoutmsg": stdout,
+            "stderrmsg": stderr,
+            "exit_code": exit_code,
+        }
+        response = self.run_tool("detect_errors", inputs, history_id)
+        self.wait_for_history(history_id, assert_ok=False)
+        return response
+
     def run_exit_code_from_file(self, history_id: str, hdca_id: str) -> dict:
         exit_code_inputs = {
             "input": {"batch": True, "values": [{"src": "hdca", "id": hdca_id}]},

test/integration/test_error_report.py

@@ -0,0 +1,114 @@
+"""Integration tests for user error reporting."""
+
+import json
+import os
+import string
+
+from galaxy_test.base.populators import DatasetPopulator
+from galaxy_test.driver import integration_util
+
+JSON_ERROR_REPORTS = """
+- type: json
+  verbose: true
+  user_submission: true
+  directory: ${reports_directory}
+"""
+
+MOCK_EMAIL_ERROR_REPORTS = """
+- type: email
+  verbose: true
+  user_submission: true
+"""
+
+
+class TestErrorReportIntegration(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    reports_directory: str
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        reports_directory = cls._test_driver.mkdtemp()
+        cls.reports_directory = reports_directory
+        template = string.Template(JSON_ERROR_REPORTS)
+        reports_yaml = template.safe_substitute({"reports_directory": reports_directory})
+        reports_conf = os.path.join(reports_directory, "error_report.yml")
+        with open(reports_conf, "w") as f:
+            f.write(reports_yaml)
+        config["error_report_file"] = reports_conf
+
+    def test_basic_tool_error(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.run_detect_errors(history_id, 6, "my stdout", "my stderr")
+            job_id = response["jobs"][0]["id"]
+            dataset_result = response["outputs"][0]
+            self.dataset_populator.report_job_error(job_id, dataset_result["id"])
+            assert len(os.listdir(self.reports_directory)) == 2
+            error_json = self.read_error_report(job_id)
+            error_dict = json.loads(error_json)
+            assert error_dict["exit_code"] == 6
+
+    def test_tool_error_custom_message_and_email(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.run_detect_errors(history_id, 6, "my stdout", "my stderr")
+            job_id = response["jobs"][0]["id"]
+            dataset_result = response["outputs"][0]
+            self.dataset_populator.report_job_error(
+                job_id, dataset_result["id"], "some new details", "[email protected]"
+            )
+            error_json = self.read_error_report(job_id)
+            error_dict = json.loads(error_json)
+            assert error_dict["exit_code"] == 6
+            assert error_dict["message"] == "some new details"
+            assert error_dict["email"] == "[email protected]"
+
+    def read_error_report(self, job_id: str):
+        app = self._app
+        job_id_decoded = app.security.decode_id(job_id)
+        with open(os.path.join(self.reports_directory, str(job_id_decoded))) as f:
+            return f.read()
+
+
+class TestErrorEmailReportIntegration(integration_util.IntegrationTestCase):
+    dataset_populator: DatasetPopulator
+    reports_directory: str
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super().setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        reports_directory = cls._test_driver.mkdtemp()
+        cls.reports_directory = reports_directory
+        template = string.Template(MOCK_EMAIL_ERROR_REPORTS)
+        reports_yaml = template.safe_substitute({"reports_directory": reports_directory})
+        reports_conf = os.path.join(reports_directory, "error_report.yml")
+        with open(reports_conf, "w") as f:
+            f.write(reports_yaml)
+        config["error_report_file"] = reports_conf
+        config["smtp_server"] = f"mock_emails_to_path://{reports_directory}/email.json"
+        config["error_email_to"] = "[email protected]"
+
+    def test_tool_error_custom_message_and_email(self):
+        with self.dataset_populator.test_history() as history_id:
+            response = self.dataset_populator.run_detect_errors(history_id, 6, "my stdout", "my stderr")
+            job_id = response["jobs"][0]["id"]
+            dataset_result = response["outputs"][0]
+            self.dataset_populator.report_job_error(
+                job_id, dataset_result["id"], "some new details", "[email protected]"
+            )
+            error_json = self.read_most_recent_error_report()
+            error_dict = json.loads(error_json)
+            assert error_dict["to"] == "[email protected], [email protected]"
+            assert error_dict["subject"] == "Galaxy tool error report from [email protected] (detect_errors)"
+            assert "<h1>Galaxy Tool Error Report</h1>" in error_dict["html"]
+
+    def read_most_recent_error_report(self):
+        with open(os.path.join(self.reports_directory, "email.json")) as f:
+            return f.read()

test/unit/app/tools/test_error_reporting.py

@@ -0,0 +1,145 @@
+import json
+import shutil
+import tempfile
+from pathlib import Path
+
+from galaxy import model
+from galaxy.app_unittest_utils.tools_support import UsesApp
+from galaxy.model.base import transaction
+from galaxy.tools.errors import EmailErrorReporter
+from galaxy.util.unittest import TestCase
+
+# The email the user created their account with.
+TEST_USER_EMAIL = "[email protected]"
+# The email the user supplied when submitting the error
+TEST_USER_SUPPLIED_EMAIL = "[email protected]"
+TEST_SERVER_EMAIL_FROM = "[email protected]"
+TEST_SERVER_ERROR_EMAIL_TO = "[email protected]"  # setup in mock config
+
+
+class TestErrorReporter(TestCase, UsesApp):
+
+    def setUp(self):
+        self.setup_app()
+        self.app.config.email_from = TEST_SERVER_EMAIL_FROM
+        self.tmp_path = Path(tempfile.mkdtemp())
+        self.email_path = self.tmp_path / "email.json"
+        smtp_server = f"mock_emails_to_path://{self.email_path}"
+        self.app.config.smtp_server = smtp_server  # type: ignore[attr-defined]
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp_path)
+
+    def test_basic(self):
+        user, hda = self._setup_model_objects()
+
+        email_path = self.email_path
+        assert not email_path.exists()
+        error_report = EmailErrorReporter(hda, self.app)
+        error_report.send_report(user, email=TEST_USER_SUPPLIED_EMAIL, message="My custom message")
+        assert email_path.exists()
+        text = email_path.read_text()
+        email_json = json.loads(text)
+        assert email_json["from"] == TEST_SERVER_EMAIL_FROM
+        assert email_json["to"] == f"{TEST_SERVER_ERROR_EMAIL_TO}, {TEST_USER_SUPPLIED_EMAIL}"
+        assert f"Galaxy tool error report from {TEST_USER_SUPPLIED_EMAIL}" == email_json["subject"]
+        assert "cat1" in email_json["body"]
+        assert "cat1" in email_json["html"]
+        assert TEST_USER_EMAIL == email_json["reply_to"]
+
+    def test_hda_security(self, tmp_path):
+        user, hda = self._setup_model_objects()
+        error_report = EmailErrorReporter(hda, self.app)
+        security_agent = self.app.security_agent
+        private_role = security_agent.create_private_user_role(user)
+        access_action = security_agent.permitted_actions.DATASET_ACCESS.action
+        manage_action = security_agent.permitted_actions.DATASET_MANAGE_PERMISSIONS.action
+        permissions = {access_action: [private_role], manage_action: [private_role]}
+        security_agent.set_all_dataset_permissions(hda.dataset, permissions)
+
+        other_user = model.User(email="[email protected]", password="mockpass2")
+        self._commit_objects([other_user])
+        security_agent = self.app.security_agent
+        email_path = self.email_path
+        assert not email_path.exists()
+        error_report.send_report(other_user, email=TEST_USER_SUPPLIED_EMAIL, message="My custom message")
+        # Without permissions, the email still gets sent but the supplied email is ignored
+        # I'm not saying this is the right behavior but it is what the code does at the time of test
+        # writing -John
+        assert email_path.exists()
+        text = email_path.read_text()
+        email_json = json.loads(text)
+        assert "[email protected]" not in email_json["to"]
+
+    def test_html_sanitization(self, tmp_path):
+        user, hda = self._setup_model_objects()
+        email_path = self.email_path
+        assert not email_path.exists()
+        error_report = EmailErrorReporter(hda, self.app)
+        error_report.send_report(
+            user, email=TEST_USER_SUPPLIED_EMAIL, message='My custom <a href="http://sneaky.com/">message</a>'
+        )
+        assert email_path.exists()
+        text = email_path.read_text()
+        email_json = json.loads(text)
+        html = email_json["html"]
+        assert "&lt;a href=&#34;http://sneaky.com/&#34;&gt;message&lt;/a&gt;" in html
+
+    def test_redact_user_details_in_bugreport(self, tmp_path):
+        user, hda = self._setup_model_objects()
+
+        email_path = self.email_path
+        assert not email_path.exists()
+        error_report = EmailErrorReporter(hda, self.app)
+        error_report.send_report(
+            user, email=TEST_USER_SUPPLIED_EMAIL, message="My custom message", redact_user_details_in_bugreport=True
+        )
+        assert email_path.exists()
+        text = email_path.read_text()
+        email_json = json.loads(text)
+        assert "The user redacted (user: 1) provided the following information:" in email_json["body"]
+        assert (
+            """The user <span style="font-family: monospace;">redacted (user: 1)</span> provided the following information:"""
+            in email_json["html"]
+        )
+
+    def test_no_redact_user_details_in_bugreport(self, tmp_path):
+        user, hda = self._setup_model_objects()
+
+        email_path = self.email_path
+        assert not email_path.exists()
+        error_report = EmailErrorReporter(hda, self.app)
+        error_report.send_report(
+            user, email=TEST_USER_SUPPLIED_EMAIL, message="My custom message", redact_user_details_in_bugreport=False
+        )
+        assert email_path.exists()
+        text = email_path.read_text()
+        email_json = json.loads(text)
+        assert (
+            f"The user '{TEST_USER_EMAIL}' (providing preferred contact email '{TEST_USER_SUPPLIED_EMAIL}') provided the following information:"
+            in email_json["body"]
+        )
+        assert (
+            f"""The user <span style="font-family: monospace;">&#39;{TEST_USER_EMAIL}&#39; (providing preferred contact email &#39;{TEST_USER_SUPPLIED_EMAIL}&#39;)</span> provided the following information:"""
+            in email_json["html"]
+        )
+
+    def _setup_model_objects(self):
+        user = model.User(email=TEST_USER_EMAIL, password="mockpass")
+        job = model.Job()
+        job.tool_id = "cat1"
+        job.history = model.History()
+        job.user = user
+        job.history.user = user
+        hda = model.HistoryDatasetAssociation(history=job.history)
+        hda.dataset = model.Dataset()
+        hda.dataset.state = "ok"
+        job.add_output_dataset("out1", hda)
+        self._commit_objects([job, hda, user])
+        return user, hda
+
+    def _commit_objects(self, objects):
+        session = self.app.model.context
+        session.add_all(objects)
+        with transaction(session):
+            session.commit()