feat: extend process_pending to populate TreeTestsRollup table (#1831)

* feat: enhance test aggregation with tree_tests rollup processing

* test: add unit tests for process_pending_helpers functions

Author: gustavobtflores

backend/kernelCI_app/constants/process_pending.py

@@ -0,0 +1,9 @@
+ROLLUP_STATUS_FIELDS = {
+    "PASS": "pass_tests",
+    "FAIL": "fail_tests",
+    "SKIP": "skip_tests",
+    "ERROR": "error_tests",
+    "MISS": "miss_tests",
+    "DONE": "done_tests",
+    "NULL": "null_tests",
+}

backend/kernelCI_app/management/commands/helpers/aggregation_helpers.py

@@ -38,6 +38,7 @@ def convert_build(b: Builds) -> PendingBuilds:
 
 
 def convert_test(t: Tests) -> PendingTest:
+    misc = t.misc or {}
     return PendingTest(
         test_id=t.id,
         origin=t.origin,
@@ -46,6 +47,10 @@ def convert_test(t: Tests) -> PendingTest:
         build_id=t.build_id,
         status=simplify_status(t.status),
         is_boot=is_boot(t.path) if t.path else False,
+        path=t.path,
+        start_time=t.start_time,
+        lab=misc.get("runtime"),
+        full_status=t.status,
     )
 
 
@@ -159,6 +164,10 @@ def aggregate_tests(
             test.build_id,
             test.status,
             test.is_boot,
+            test.path,
+            test.start_time,
+            test.lab,
+            test.full_status,
         )
         for test in pending_tests
     ]
@@ -167,13 +176,18 @@ def aggregate_tests(
         query = """
             INSERT INTO pending_test (
                 test_id, origin, platform, compatible,
-                build_id, status, is_boot
-            ) VALUES (%s, %s, %s, %s, %s, %s, %s)
+                build_id, status, is_boot,
+                path, start_time, lab, full_status
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
             ON CONFLICT (test_id)
             DO UPDATE SET
                 platform = COALESCE(pending_test.platform, EXCLUDED.platform),
                 compatible = COALESCE(pending_test.compatible, EXCLUDED.compatible),
-                status = COALESCE(pending_test.status, EXCLUDED.status)
+                status = COALESCE(pending_test.status, EXCLUDED.status),
+                path = COALESCE(pending_test.path, EXCLUDED.path),
+                start_time = COALESCE(pending_test.start_time, EXCLUDED.start_time),
+                lab = COALESCE(pending_test.lab, EXCLUDED.lab),
+                full_status = COALESCE(pending_test.full_status, EXCLUDED.full_status)
         """
 
         with connections["default"].cursor() as cursor:

backend/kernelCI_app/management/commands/helpers/process_pending_helpers.py

@@ -0,0 +1,160 @@
+from typing import NamedTuple, Optional, Sequence, TypedDict
+from kernelCI_app.helpers.logger import logger
+from kernelCI_app.constants.general import UNKNOWN_STRING
+from kernelCI_app.constants.process_pending import ROLLUP_STATUS_FIELDS
+from kernelCI_app.models import Builds, Checkouts, PendingTest, StatusChoices
+
+
+EMPTY_PATH_GROUP = "-"
+
+
+class RollupKey(NamedTuple):
+    origin: str
+    tree_name: str
+    git_repository_branch: str
+    git_repository_url: str
+    git_commit_hash: str
+    path_group: str
+    config: str
+    arch: str
+    compiler: str
+    hardware_key: str
+    platform: Optional[str]
+    lab: Optional[str]
+    test_origin: str
+    issue_id: Optional[str]
+    issue_version: Optional[int]
+    issue_uncategorized: bool
+    is_boot: bool
+
+
+class RollupEntryData(TypedDict):
+    checkout: Checkouts
+    path_group: str
+    config: str
+    arch: str
+    compiler: str
+    hardware_key: str
+    platform: Optional[str]
+    lab: Optional[str]
+    origin: str
+    issue_id: Optional[str]
+    issue_version: Optional[int]
+    issue_uncategorized: bool
+    is_boot: bool
+    status: StatusChoices
+
+
+def extract_path_group(path: str) -> str:
+    """Extract the path group from a path."""
+    return path.split(".", 1)[0] if path else EMPTY_PATH_GROUP
+
+
+def accumulate_rollup_entry(
+    rollup_data: dict[tuple, dict],
+    entry: RollupEntryData,
+) -> None:
+    """Accumulate a single test entry into rollup_data in-place."""
+    checkout = entry["checkout"]
+    rollup_key = RollupKey(
+        origin=checkout.origin,
+        tree_name=checkout.tree_name,
+        git_repository_branch=checkout.git_repository_branch,
+        git_repository_url=checkout.git_repository_url,
+        git_commit_hash=checkout.git_commit_hash,
+        path_group=entry["path_group"],
+        config=entry["config"],
+        arch=entry["arch"],
+        compiler=entry["compiler"],
+        hardware_key=entry["hardware_key"],
+        platform=entry["platform"],
+        lab=entry["lab"],
+        test_origin=entry["origin"],
+        issue_id=entry["issue_id"],
+        issue_version=entry["issue_version"],
+        issue_uncategorized=entry["issue_uncategorized"],
+        is_boot=entry["is_boot"],
+    )
+
+    record = rollup_data.setdefault(
+        rollup_key,
+        {
+            "pass_tests": 0,
+            "fail_tests": 0,
+            "skip_tests": 0,
+            "error_tests": 0,
+            "miss_tests": 0,
+            "done_tests": 0,
+            "null_tests": 0,
+            "total_tests": 0,
+        },
+    )
+
+    counter = ROLLUP_STATUS_FIELDS.get(entry["status"], "null_tests")
+    record[counter] += 1
+    record["total_tests"] += 1
+
+
+def aggregate_tests_rollup(
+    ready_tests: Sequence[PendingTest],
+    test_builds_by_id: dict[str, Builds],
+    issues_map: dict[str, dict],
+) -> dict[tuple, dict]:
+    """
+    Build rollup data from pending tests.
+    Returns rollup data without touching the database.
+    """
+    rollup_data: dict[tuple, dict] = {}
+
+    for test in ready_tests:
+        # shouldn't happen, but being defensive here
+        try:
+            build = test_builds_by_id[test.build_id]
+        except KeyError:
+            logger.warning(
+                f"Found test {test.test_id} with no build {test.build_id} on aggregate_tests_rollup"
+            )
+            continue
+
+        checkout = build.checkout
+        path = test.path or ""
+        path_group = extract_path_group(path)
+
+        hardware_key = UNKNOWN_STRING
+        if test.compatible:
+            hardware_key = test.compatible[0]
+        elif test.platform:
+            hardware_key = test.platform
+
+        issue_info = issues_map.get(test.test_id, {})
+        issue_id = issue_info.get("issue_id")
+        issue_version = issue_info.get("issue_version")
+        issue_uncategorized = (
+            issue_id is None and test.full_status == StatusChoices.FAIL
+        )
+
+        arch = build.architecture or UNKNOWN_STRING
+        compiler = build.compiler or UNKNOWN_STRING
+        config = build.config_name or UNKNOWN_STRING
+
+        accumulate_rollup_entry(
+            rollup_data,
+            {
+                "checkout": checkout,
+                "path_group": path_group,
+                "config": config,
+                "arch": arch,
+                "compiler": compiler,
+                "hardware_key": hardware_key,
+                "platform": test.platform,
+                "lab": test.lab,
+                "origin": test.origin,
+                "issue_id": issue_id,
+                "issue_version": issue_version,
+                "issue_uncategorized": issue_uncategorized,
+                "is_boot": test.is_boot,
+                "status": test.full_status,
+            },
+        )
+
+    return rollup_data

backend/kernelCI_app/management/commands/process_pending_aggregations.py

@@ -5,6 +5,9 @@
 import time
 from datetime import datetime
 from typing import Literal, Optional, Sequence, TypedDict, Union
+from kernelCI_app.management.commands.helpers.process_pending_helpers import (
+    aggregate_tests_rollup,
+)
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.db import connection, transaction
@@ -16,6 +19,7 @@
 from kernelCI_app.models import (
     Builds,
     Checkouts,
+    Incidents,
     PendingTest,
     PendingBuilds,
     ProcessedListingItems,
@@ -492,6 +496,25 @@ def aggregate_hardware_status(
     return hardware_status_data, new_processed_entries
 
 
+def _fetch_test_issues(test_ids: list[str]) -> dict[str, dict]:
+    """Bulk-fetch the first incident per test_id, returning {test_id: {issue_id, issue_version}}."""
+    issues_map: dict[str, dict] = {}
+    incidents = Incidents.objects.filter(
+        test_id__in=test_ids,
+    ).values("test_id", "issue_id", "issue_version")
+
+    for inc in incidents:
+        issues_map.setdefault(
+            inc["test_id"],
+            {
+                "issue_id": inc["issue_id"],
+                "issue_version": inc["issue_version"],
+            },
+        )
+
+    return issues_map
+
+
 class Command(BaseCommand):
     help = """
         Process pending tests for hardware status aggregation,
@@ -855,8 +878,17 @@ def _get_ready_tests(
             .only(
                 "id",
                 "status",
+                "architecture",
+                "compiler",
+                "config_name",
+                "misc",
                 "checkout__id",
                 "checkout__start_time",
+                "checkout__origin",
+                "checkout__tree_name",
+                "checkout__git_repository_url",
+                "checkout__git_repository_branch",
+                "checkout__git_commit_hash",
             )
             .in_bulk(pending_test_build_ids)
         )
@@ -887,6 +919,75 @@ def _get_ready_tests(
             pending_test_count,
         )
 
+    def _process_tests_rollup(self, rollup_data: dict[tuple, dict]) -> None:
+        if not rollup_data:
+            return
+
+        values = [
+            (
+                *key,
+                data["pass_tests"],
+                data["fail_tests"],
+                data["skip_tests"],
+                data["error_tests"],
+                data["miss_tests"],
+                data["done_tests"],
+                data["null_tests"],
+                data["total_tests"],
+            )
+            for key, data in rollup_data.items()
+        ]
+
+        t0 = time.time()
+        with connection.cursor() as cursor:
+            cursor.executemany(
+                """
+                INSERT INTO tree_tests_rollup (
+                    origin, tree_name, git_repository_branch, git_repository_url,
+                    git_commit_hash,
+                    path_group, build_config_name, build_architecture, build_compiler,
+                    hardware_key, test_platform, test_lab, test_origin,
+                    issue_id, issue_version, issue_uncategorized, is_boot,
+                    pass_tests, fail_tests, skip_tests,
+                    error_tests, miss_tests, done_tests,
+                    null_tests, total_tests
+                )
+                VALUES (
+                    %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                    %s, %s, %s, %s, %s, %s, %s, %s, %s,
+                    %s, %s, %s, %s, %s, %s, %s
+                )
+                ON CONFLICT ON CONSTRAINT tree_tests_rollup_unique DO UPDATE SET
+                    pass_tests = tree_tests_rollup.pass_tests + EXCLUDED.pass_tests,
+                    fail_tests = tree_tests_rollup.fail_tests + EXCLUDED.fail_tests,
+                    skip_tests = tree_tests_rollup.skip_tests + EXCLUDED.skip_tests,
+                    error_tests = tree_tests_rollup.error_tests + EXCLUDED.error_tests,
+                    miss_tests = tree_tests_rollup.miss_tests + EXCLUDED.miss_tests,
+                    done_tests = tree_tests_rollup.done_tests + EXCLUDED.done_tests,
+                    null_tests = tree_tests_rollup.null_tests + EXCLUDED.null_tests,
+                    total_tests = tree_tests_rollup.total_tests + EXCLUDED.total_tests
+                """,
+                values,
+            )
+        out(
+            f"Upserted {len(values)} tree_tests_rollup records "
+            f"in {time.time() - t0:.3f}s"
+        )
+        AGGREGATION_RECORDS_WRITTEN.labels(table="tree_tests_rollup").inc(len(values))
+
+    def _process_tests_rollup_batch(
+        self,
+        ready_tests: Sequence[PendingTest],
+        test_builds_by_id: dict[str, Builds],
+    ) -> None:
+        if not ready_tests:
+            return
+
+        test_ids = [t.test_id for t in ready_tests]
+        issues_map = _fetch_test_issues(test_ids)
+        rollup_data = aggregate_tests_rollup(ready_tests, test_builds_by_id, issues_map)
+        self._process_tests_rollup(rollup_data)
+
     def _process_hardware_batch(
         self,
         ready_tests: Sequence[PendingTest],
@@ -943,6 +1044,7 @@ def process_pending_batch(self, batch_size: int) -> int:
 
                 if ready_tests:
                     self._process_hardware_batch(ready_tests, test_builds_by_id)
+                    self._process_tests_rollup_batch(ready_tests, test_builds_by_id)
 
                 (
                     ready_builds,