Create report status constants

tests/ci/commit_status_helper.py

@@ -3,7 +3,7 @@
 import csv
 import os
 import time
-from typing import Dict, List, Literal, Optional, Union
+from typing import Dict, List, Optional, Union
 import logging
 
 from github import Github
@@ -16,7 +16,16 @@ from github.Repository import Repository
 from ci_config import CI_CONFIG, REQUIRED_CHECKS, CHECK_DESCRIPTIONS, CheckDescription
 from env_helper import GITHUB_REPOSITORY, GITHUB_RUN_URL
 from pr_info import PRInfo, SKIP_MERGEABLE_CHECK_LABEL
-from report import TestResult, TestResults
+from report import (
+    ERROR,
+    FAILURE,
+    PENDING,
+    StatusType,
+    SUCCESS,
+    TestResult,
+    TestResults,
+    get_worst_status,
+)
 from s3_helper import S3Helper
 from upload_result_helper import upload_results
 
@@ -37,8 +46,8 @@ class RerunHelper:
         # currently we agree even for failed statuses
         for status in self.statuses:
             if self.check_name in status.context and status.state in (
-                "success",
-                "failure",
+                SUCCESS,
+                FAILURE,
             ):
                 return True
         return False
@@ -53,12 +62,12 @@ class RerunHelper:
 def override_status(status: str, check_name: str, invert: bool = False) -> str:
     test_config = CI_CONFIG.test_configs.get(check_name)
     if test_config and test_config.force_tests:
-        return "success"
+        return SUCCESS
 
     if invert:
-        if status == "success":
-            return "error"
-        return "success"
+        if status == SUCCESS:
+            return ERROR
+        return SUCCESS
 
     return status
 
@@ -137,7 +146,7 @@ def set_status_comment(commit: Commit, pr_info: PRInfo) -> None:
         # W/o pr_info to avoid recursion, and yes, one extra create_ci_report
         post_commit_status(
             commit,
-            "pending",
+            PENDING,
             create_ci_report(pr_info, statuses),
             "The report for running CI",
             CI_STATUS_NAME,
@@ -172,11 +181,11 @@ def generate_status_comment(pr_info: PRInfo, statuses: CommitStatuses) -> str:
     """The method generates the comment body, as well it updates the CI report"""
 
     def beauty_state(state: str) -> str:
-        if state == "success":
+        if state == SUCCESS:
             return f"🟢 {state}"
-        if state == "pending":
+        if state == PENDING:
             return f"🟡 {state}"
-        if state in ["error", "failure"]:
+        if state in [ERROR, FAILURE]:
             return f"🔴 {state}"
         return state
 
@@ -235,20 +244,7 @@ def generate_status_comment(pr_info: PRInfo, statuses: CommitStatuses) -> str:
 
 
 def get_worst_state(statuses: CommitStatuses) -> str:
-    worst_status = None
-    states = {"error": 0, "failure": 1, "pending": 2, "success": 3}
-    for status in statuses:
-        if worst_status is None:
-            worst_status = status
-            continue
-        if states[status.state] < states[worst_status.state]:
-            worst_status = status
-        if worst_status.state == "error":
-            break
-
-    if worst_status is None:
-        return ""
-    return worst_status.state
+    return get_worst_status(status.state for status in statuses)
 
 
 def create_ci_report(pr_info: PRInfo, statuses: CommitStatuses) -> str:
@@ -324,7 +320,7 @@ def format_description(description: str) -> str:
 def set_mergeable_check(
     commit: Commit,
     description: str = "",
-    state: Literal["success", "failure"] = "success",
+    state: StatusType = "success",
 ) -> None:
     commit.create_status(
         context=MERGEABLE_NAME,
@@ -363,7 +359,7 @@ def update_mergeable_check(gh: Github, pr_info: PRInfo, check_name: str) -> None
     success = []
     fail = []
     for status in required_checks:
-        if status.state == "success":
+        if status.state == SUCCESS:
             success.append(status.context)
         else:
             fail.append(status.context)
@@ -372,7 +368,7 @@ def update_mergeable_check(gh: Github, pr_info: PRInfo, check_name: str) -> None
         description = "failed: " + ", ".join(fail)
         description = format_description(description)
         if mergeable_status is None or mergeable_status.description != description:
-            set_mergeable_check(commit, description, "failure")
+            set_mergeable_check(commit, description, FAILURE)
         return
 
     description = ", ".join(success)

tests/ci/install_check.py

@@ -29,7 +29,7 @@ from docker_pull_helper import get_image_with_version, DockerImage
 from env_helper import CI, TEMP_PATH as TEMP, REPORTS_PATH
 from get_robot_token import get_best_robot_token
 from pr_info import PRInfo
-from report import TestResults, TestResult
+from report import TestResults, TestResult, FAILURE, FAIL, OK, SUCCESS
 from s3_helper import S3Helper
 from stopwatch import Stopwatch
 from tee_popen import TeePopen
@@ -40,10 +40,6 @@ RPM_IMAGE = "clickhouse/install-rpm-test"
 DEB_IMAGE = "clickhouse/install-deb-test"
 TEMP_PATH = Path(TEMP)
 LOGS_PATH = TEMP_PATH / "tests_logs"
-SUCCESS = "success"
-FAILURE = "failure"
-OK = "OK"
-FAIL = "FAIL"
 
 
 def prepare_test_scripts():

tests/ci/report.py

@@ -2,12 +2,45 @@
 from ast import literal_eval
 from dataclasses import dataclass
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Final, Iterable, List, Literal, Optional, Tuple
 from html import escape
 import csv
 import os
 import datetime
 
+
+ERROR: Final = "error"
+FAILURE: Final = "failure"
+PENDING: Final = "pending"
+SUCCESS: Final = "success"
+
+OK: Final = "OK"
+FAIL: Final = "FAIL"
+
+StatusType = Literal["error", "failure", "pending", "success"]
+# The order of statuses from the worst to the best
+_STATES = {ERROR: 0, FAILURE: 1, PENDING: 2, SUCCESS: 3}
+
+
+def get_worst_status(statuses: Iterable[str]) -> str:
+    worst_status = None
+    for status in statuses:
+        if _STATES.get(status) is None:
+            continue
+        if worst_status is None:
+            worst_status = status
+            continue
+        if _STATES.get(status) < _STATES.get(worst_status):
+            worst_status = status
+
+        if worst_status == ERROR:
+            break
+
+    if worst_status is None:
+        return ""
+    return worst_status
+
+
 ### BEST FRONTEND PRACTICES BELOW
 
 HEAD_HTML_TEMPLATE = """
@@ -291,8 +324,8 @@ def _format_header(
 
 
 def _get_status_style(status: str, colortheme: Optional[ColorTheme] = None) -> str:
-    ok_statuses = ("OK", "success", "PASSED")
-    fail_statuses = ("FAIL", "failure", "error", "FAILED", "Timeout", "NOT_FAILED")
+    ok_statuses = (OK, SUCCESS, "PASSED")
+    fail_statuses = (FAIL, FAILURE, ERROR, "FAILED", "Timeout", "NOT_FAILED")
 
     if colortheme is None:
         colortheme = ReportColorTheme.default
@@ -490,7 +523,7 @@ def create_build_html_report(
             style = _get_status_style(build_result.status)
             row.append(f'<td style="{style}">{build_result.status}</td>')
         else:
-            style = _get_status_style("error")
+            style = _get_status_style(ERROR)
             row.append(f'<td style="{style}">error</td>')
 
         row.append(f'<td><a href="{build_log_url}">link</a></td>')