Add test results finisher

services/lock_manager.py

@@ -16,6 +16,7 @@
 class LockType(Enum):
     BUNDLE_ANALYSIS_PROCESSING = "bundle_analysis_processing"
     BUNDLE_ANALYSIS_NOTIFY = "bundle_analysis_notify"
+    NOTIFICATION = "notify"
     # TODO: port existing task locking to use `LockManager`
 
 

services/test_results.py

@@ -3,15 +3,17 @@
 from typing import Mapping, Sequence
 
 from shared.torngit.exceptions import TorngitClientError
-from test_results_parser import Outcome, Testrun
+from sqlalchemy import desc
+from test_results_parser import Outcome
 
 from database.enums import ReportType
-from database.models import Commit, CommitReport, RepositoryFlag, Upload
+from database.models import Commit, CommitReport, RepositoryFlag, TestInstance, Upload
 from services.report import BaseReportService
 from services.repository import (
     fetch_and_update_pull_request_information_from_commit,
     get_repo_provider_service,
 )
+from services.urls import get_pull_url
 
 log = logging.getLogger(__name__)
 
@@ -94,3 +96,157 @@
             "utf-8"
         )
     ).hexdigest()
+
+
+class TestResultsNotifier:
+    def __init__(self, commit: Commit, commit_yaml, test_instances):
+        self.commit = commit
+        self.commit_yaml = commit_yaml
+        self.test_instances = test_instances
+
+    async def notify(self):
+        commit_report = self.commit.commit_report(report_type=ReportType.TEST_RESULTS)
+        if not commit_report:
+            log.warning(
+                "No test results commit report found for this commit",
+                extra=dict(
+                    commitid=self.commit.commitid,
+                    report_key=commit_report.external_id,
+                ),
+            )
+
+        repo_service = get_repo_provider_service(self.commit.repository, self.commit)
+
+        pull = await fetch_and_update_pull_request_information_from_commit(
+            repo_service, self.commit, self.commit_yaml
+        )
+        pullid = pull.database_pull.pullid
+        if pull is None:
+            log.info(
+                "Not notifying since there is no pull request associated with this commit",
+                extra=dict(
+                    commitid=self.commit.commitid,
+                    report_key=commit_report.external_id,
+                    pullid=pullid,
+                ),
+            )
+
+        pull_url = get_pull_url(pull.database_pull)
+
+        message = self.build_message(pull_url, self.test_instances)
+
+        try:
+            comment_id = pull.database_pull.commentid
+            if comment_id:
+                await repo_service.edit_comment(pullid, comment_id, message)
+            else:
+                res = await repo_service.post_comment(pullid, message)
+                pull.database_pull.commentid = res["id"]
+            return True
+        except TorngitClientError:
+            log.error(
+                "Error creating/updating PR comment",
+                extra=dict(
+                    commitid=self.commit.commitid,
+                    report_key=commit_report.external_id,
+                    pullid=pullid,
+                ),
+            )
+            return False
+
+    def build_message(self, url, test_instances):
+        message = []
+
+        message += [
+            f"##  [Codecov]({url}) Report",
+            "",
+            "**Test Failures Detected**: Due to failing tests, we cannot provide coverage reports at this time.",
+            "",
+            "### :x: Failed Test Results: ",
+        ]
+        failed_tests = 0
+        passed_tests = 0
+        skipped_tests = 0
+
+        failures = dict()
+
+        for test_instance in test_instances:
+            if test_instance.outcome == str(
+                Outcome.Failure
+            ) or test_instance.outcome == str(Outcome.Error):
+                failed_tests += 1
+                job_code = test_instance.upload.job_code
+                flag_names = sorted(test_instance.upload.flag_names)
+                suffix = ""
+                if job_code or flag_names:
+                    suffix = f"[{''.join(flag_names) or ''} {job_code or ''}]"
+                failures[
+                    f"{test_instance.test.testsuite}::{test_instance.test.name}{suffix}"
+                ] = test_instance.failure_message
+            elif test_instance.outcome == str(Outcome.Skip):
+                skipped_tests += 1
+            elif test_instance.outcome == str(Outcome.Pass):
+                passed_tests += 1
+
+        results = f"Completed {len(test_instances)} tests with **`{failed_tests} failed`**, {passed_tests} passed and {skipped_tests} skipped."
+
+        message.append(results)
+
+        details = [
+            "<details><summary>View the full list of failed tests</summary>",
+            "",
+            "| **File path** | **Failure message** |",
+            "| :-- | :-- |",
+        ]
+
+        message += details
+
+        failure_table = [
+            "| {0} | <pre>{1}</pre> |".format(
+                self.insert_breaks(failed_test_name),
+                failure_message.replace("\n", "<br>"),
+            )
+            for failed_test_name, failure_message in sorted(
+                failures.items(), key=lambda failure: failure[0]
+            )
+        ]
+
+        message += failure_table
+
+        return "\n".join(message)
+
+    def insert_breaks(self, table_value):
+        line_size = 70
+        lines = [
+            table_value[i : i + line_size]
+            for i in range(0, len(table_value), line_size)
+        ]
+        return "<br>".join(lines)
+
+
+def latest_test_instances_for_a_given_commit(db_session, commit_id):
+    """
+    This will result in a SQL query that looks something like this:
+
+    SELECT DISTINCT ON (rt.test_id) rt.id, rt.external_id, rt.created_at, rt.updated_at, rt.test_id, rt.duration_seconds, rt.outcome, rt.upload_id, rt.failure_message
+        FROM reports_testinstance rt JOIN reports_upload ru ON ru.id = rt.upload_id JOIN reports_commitreport rc ON rc.id = ru.report_id
+        WHERE rc.commit_id = <commit_id> ORDER BY rt.test_id, ru.created_at desc
+
+    The goal of this query is to return: "the latest test instance for each unique test based on upload creation time"
+
+    The `DISTINCT ON` test_id with the order by test_id, enforces that we are only fetching one test instance for each test
+
+    The ordering of the upload.create_at desc enforces that we get the latest test instance for that unique test
+    """
+    return (
+        db_session.query(TestInstance)
+        .join(Upload)
+        .join(CommitReport)
+        .filter(
+            CommitReport.commit_id == commit_id,
+        )
+        .order_by(TestInstance.test_id)
+        .order_by(desc(Upload.created_at))
+        .distinct(TestInstance.test_id)
+        .all()
+    )

tasks/notify.py

@@ -10,10 +10,12 @@
 from shared.reports.readonly import ReadOnlyReport
 from shared.torngit.exceptions import TorngitClientError, TorngitServerFailureError
 from shared.yaml import UserYaml
+from sqlalchemy import desc
 from sqlalchemy.orm.session import Session
+from test_results_parser import Outcome
 
 from app import celery_app
-from database.enums import CommitErrorTypes, Decoration
+from database.enums import CommitErrorTypes, Decoration, ReportType
 from database.models import Commit, Pull
 from database.models.core import GITHUB_APP_INSTALLATION_DEFAULT_NAME
 from helpers.checkpoint_logger import from_kwargs as checkpoints_from_kwargs
@@ -26,6 +28,7 @@
 from services.comparison import ComparisonProxy
 from services.comparison.types import Comparison, FullCommit
 from services.decoration import determine_decoration_details
+from services.lock_manager import LockManager, LockType
 from services.notification import NotificationService
 from services.redis import Redis, get_redis_connection
 from services.report import ReportService
@@ -34,6 +37,7 @@
     fetch_and_update_pull_request_information_from_commit,
     get_repo_provider_service,
 )
+from services.test_results import latest_test_instances_for_a_given_commit
 from services.yaml import get_current_yaml, read_yaml_field
 from tasks.base import BaseCodecovTask
 from tasks.upload_processor import UPLOAD_PROCESSING_LOCK_NAME
@@ -68,13 +72,18 @@
                 "notifications": None,
                 "reason": "has_other_notifies_coming",
             }
-        notify_lock_name = f"notify_lock_{repoid}_{commitid}"
+
+        lock_manager = LockManager(
+            repoid=repoid,
+            commitid=commitid,
+            report_type=ReportType.COVERAGE,
+            lock_timeout=max(80, self.hard_time_limit_task),
+        )
+
         try:
             lock_acquired = False
-            with redis_connection.lock(
-                notify_lock_name,
-                timeout=max(80, self.hard_time_limit_task),
-                blocking_timeout=10,
+            with lock_manager.locked(
+                lock_type=LockType.NOTIFICATION, retry_num=self.request.retries
             ):
                 lock_acquired = True
                 return await self.run_async_within_lock(
@@ -131,6 +140,21 @@
             Commit.repoid == repoid, Commit.commitid == commitid
         )
         commit = commits_query.first()
+
+        # check if there were any test failures
+        latest_test_instances = latest_test_instances_for_a_given_commit(
+            db_session, commit.id_
+        )
+
+        if any(
+            [test.outcome == str(Outcome.Failure) for test in latest_test_instances]
+        ):
+            return {
+                "notify_attempted": False,
+                "notifications": None,
+                "reason": "test_failures",
+            }
+
         try:
             repository_service = get_repo_provider_service(commit.repository)
         except RepositoryWithoutValidBotError: