Feat: add test series field

backend/kernelCI_app/constants/localization.py

@@ -18,6 +18,7 @@ class ClientStrings:
     TREE_REPORT_MIN_MAX_AGE = (
         "Minimum age can't be greater than or equal to the maximum age"
     )
+    TEST_SERIES_NOT_FOUND = "Test series not found"
     TEST_STATUS_HISTORY_NOT_FOUND = "Test status history not found"
     TEST_ISSUES_NOT_FOUND = "No issues were found for this test"
     TEST_NOT_FOUND = "Test not found"
@@ -114,6 +115,9 @@ class DocStrings:
     TREE_QUERY_ORIGIN_DESCRIPTION = "Origin of the tree"
     TREE_QUERY_GIT_URL_DESCRIPTION = "Git repository URL of the tree"
 
+    SERIES_ARCHITECTURE_DESCRIPTION = "Architecture filter to retrieve tests"
+    SERIES_COMPILER_DESCRIPTION = "Compiler filter to retrieve tests"
+
     STATUS_HISTORY_PATH_DESCRIPTION = "Test path filter"
     STATUS_HISTORY_ORIGIN_DESCRIPTION = "Origin filter to retrieve tests"
     STATUS_HISTORY_GIT_URL_DESCRIPTION = "Git repository URL to retrieve tests"

backend/kernelCI_app/helpers/test.py

@@ -0,0 +1,40 @@
+from kernelCI_app.typeModels.databases import FAIL_STATUS, PASS_STATUS
+from kernelCI_app.typeModels.testDetails import PossibleRegressionType
+
+
+# TODO: create unit tests for this method
+def process_test_status_history(
+    *, status_history: list[dict]
+) -> PossibleRegressionType:
+    history_task: PossibleRegressionType
+    first_test_flag = True
+    status_changed = False
+
+    for test in status_history:
+        test_status = test["status"]
+        if first_test_flag:
+            if test_status == PASS_STATUS:
+                history_task = "pass"
+                starting_status = PASS_STATUS
+                opposite_status = FAIL_STATUS
+            elif test_status == FAIL_STATUS:
+                history_task = "fail"
+                starting_status = FAIL_STATUS
+                opposite_status = PASS_STATUS
+            else:
+                return "unstable"
+            first_test_flag = False
+            continue
+
+        is_inconclusive = test_status != PASS_STATUS and test_status != FAIL_STATUS
+
+        if test_status == opposite_status:
+            status_changed = True
+            if history_task == "pass":
+                history_task = "fixed"
+            elif history_task == "fail":
+                history_task = "regression"
+        if (status_changed and test_status == starting_status) or is_inconclusive:
+            return "unstable"
+
+    return history_task

backend/kernelCI_app/migrations/0004_add_test_series_field.py

@@ -0,0 +1,31 @@
+# Generated by Django 5.1.12 on 2025-09-11 18:10
+
+import django.db.models.functions.text
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("kernelCI_app", "0003_add_build_series_field"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="tests",
+            name="series",
+            field=models.GeneratedField(
+                db_persist=True,
+                expression=django.db.models.functions.text.MD5(
+                    django.db.models.functions.text.Concat(
+                        models.F("path"), models.F("environment_misc__platform")
+                    )
+                ),
+                output_field=models.TextField(blank=True, null=True),
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="tests",
+            index=models.Index(fields=["series"], name="tests_series_idx"),
+        ),
+    ]

backend/kernelCI_app/models.py

@@ -139,9 +139,22 @@ class UnitPrefix(models.TextChoices):
     )
     number_unit = models.TextField(blank=True, null=True)
     input_files = models.JSONField(blank=True, null=True)
+    series = models.GeneratedField(
+        expression=MD5(
+            Concat(
+                F("path"),
+                F("environment_misc__platform"),
+            )
+        ),
+        output_field=models.TextField(blank=True, null=True),
+        db_persist=True,
+    )
 
     class Meta:
         db_table = "tests"
+        indexes = [
+            models.Index(fields=["series"], name="tests_series_idx"),
+        ]
 
 
 class Incidents(models.Model):

backend/kernelCI_app/queries/test.py

@@ -1,6 +1,6 @@
 from typing import Optional
 
-from django.db import connection
+from django.db import connection, connections
 from kernelCI_app.helpers.database import dict_fetchall
 from kernelCI_app.models import Tests
 from kernelCI_app.typeModels.databases import (
@@ -92,3 +92,37 @@ def get_test_status_history(
     else:
         query = query.filter(start_time__lte=test_start_time)
         return query.order_by("-start_time")[:group_size]
+
+
+def get_series_data(
+    *, test_series: str, build_series: str, limit: str, origin: str
+) -> list[dict]:
+    params = {
+        "test_series": test_series,
+        "build_series": build_series,
+        "limit": limit,
+        "test_origin": origin,
+    }
+
+    # TODO: check if we should filter by checkout origin or test origin
+    query = """
+            SELECT
+                t.id,
+                t.start_time,
+                t.status,
+                c.git_commit_hash AS build__checkout__git_commit_hash
+            FROM
+                tests t
+            LEFT JOIN builds b ON t.build_id = b.id
+            LEFT JOIN checkouts c ON b.checkout_id = c.id
+            WHERE
+                t.series = %(test_series)s
+                AND b.series = %(build_series)s
+                AND t.origin = %(test_origin)s
+            ORDER BY t.start_time DESC
+            LIMIT %(limit)s
+        """
+
+    with connections["default"].cursor() as cursor:
+        cursor.execute(query, params)
+        return dict_fetchall(cursor)

backend/kernelCI_app/routers/databaseRouter.py

@@ -9,7 +9,7 @@ class DatabaseRouter:
 
     def allow_migrate(self, db, app_label, model_name=None, **hints):
         if db == "default":
-            return settings.USE_DASHBOARD_DB
+            return settings.USE_DASHBOARD_DB and app_label != "kernelCI_cache"
         if model_name in ["notificationscheckout", "notificationsissue"]:
             return db == "notifications"
         if model_name in ["checkoutscache"]:

backend/kernelCI_app/tests/integrationTests/testSeries_test.py

@@ -0,0 +1,132 @@
+from kernelCI_app.typeModels.testDetails import TestStatusSeriesRequest
+from kernelCI_app.tests.utils.client.testClient import TestClient
+from kernelCI_app.tests.utils.asserts import (
+    assert_status_code_and_error_response,
+    assert_has_fields_in_response_content,
+)
+from kernelCI_app.tests.utils.fields.tests import (
+    test_series_response_expected_fields,
+    status_history_item_fields,
+)
+from kernelCI_app.utils import string_to_json
+import pytest
+from http import HTTPStatus
+
+
+client = TestClient()
+
+
+@pytest.mark.parametrize(
+    "params, status_code, has_error_body",
+    [
+        (
+            # maestro:68c3e1c661dc41d269171fe1
+            TestStatusSeriesRequest(
+                path="ltp",
+                platform="imx6q-sabrelite",
+                config_name="multi_v7_defconfig",
+                architecture="arm",
+                compiler="gcc-12",
+            ),
+            HTTPStatus.OK,
+            False,
+        ),
+        (
+            TestStatusSeriesRequest(
+                path="non-existent-path",
+            ),
+            HTTPStatus.OK,
+            True,
+        ),
+        (
+            # maestro:68ba7c39b9811ea53d062635
+            TestStatusSeriesRequest(
+                path="kunit.exec.time_test_cases.time64_to_tm_test_date_range",
+                platform="kubernetes",
+                architecture="x86_64",
+            ),
+            HTTPStatus.OK,
+            False,
+        ),
+    ],
+)
+def test_get_test_series(
+    pytestconfig,
+    params: TestStatusSeriesRequest,
+    status_code,
+    has_error_body,
+):
+    response = client.get_test_series(query=params)
+    content = string_to_json(response.content.decode())
+    if "warning" in content:
+        pytest.skip("Skipping test because the migrations have not been run yet")
+
+    assert_status_code_and_error_response(
+        response=response,
+        content=content,
+        status_code=status_code,
+        should_error=has_error_body,
+    )
+
+    if not has_error_body:
+        assert_has_fields_in_response_content(
+            fields=test_series_response_expected_fields, response_content=content
+        )
+
+        # Verify status_history contains expected fields
+        status_history = content.get("status_history", [])
+        assert len(status_history) > 0, "status_history should not be empty"
+
+        for status_history_item in status_history:
+            assert_has_fields_in_response_content(
+                fields=status_history_item_fields, response_content=status_history_item
+            )
+            if not pytestconfig.getoption("--run-all"):
+                break
+
+        test_series = content.get("test_series")
+        build_series = content.get("build_series")
+        assert (
+            isinstance(test_series, str) and len(test_series) == 32
+        ), f"test_series should be a 32-character MD5 hash, got: {test_series}"
+        assert (
+            isinstance(build_series, str) and len(build_series) == 32
+        ), f"build_series should be a 32-character MD5 hash, got: {build_series}"
+
+
+def test_get_test_series_platform_quoting():
+    """Test that platform parameter handles quoting correctly"""
+    params_unquoted = TestStatusSeriesRequest(
+        path="fluster.debian.v4l2.gstreamer_av1.validate-fluster-results",
+        platform="mt8195-cherry-tomato-r2",
+        group_size=5,
+    )
+
+    params_quoted = TestStatusSeriesRequest(
+        path="fluster.debian.v4l2.gstreamer_av1.validate-fluster-results",
+        platform='"mt8195-cherry-tomato-r2"',
+        group_size=5,
+    )
+
+    response_unquoted = client.get_test_series(query=params_unquoted)
+    response_quoted = client.get_test_series(query=params_quoted)
+
+    content_unquoted = string_to_json(response_unquoted.content.decode())
+    content_quoted = string_to_json(response_quoted.content.decode())
+
+    # Both should return the same test_series hash since the platform
+    # should be normalized to the quoted version
+
+    if "warning" in content_unquoted or "warning" in content_quoted:
+        pytest.skip("Skipping test because the migrations have not been run yet")
+
+    assert_has_fields_in_response_content(
+        fields=test_series_response_expected_fields, response_content=content_unquoted
+    )
+    assert_has_fields_in_response_content(
+        fields=test_series_response_expected_fields, response_content=content_unquoted
+    )
+
+    assert (
+        content_unquoted["test_series"] == content_quoted["test_series"]
+    ), "Quoted and unquoted platforms should result in the same test_series hash"

backend/kernelCI_app/tests/utils/client/testClient.py

@@ -2,7 +2,10 @@
 from django.urls import reverse
 
 from kernelCI_app.tests.utils.client.baseClient import BaseClient
-from kernelCI_app.typeModels.testDetails import TestStatusHistoryRequest
+from kernelCI_app.typeModels.testDetails import (
+    TestStatusHistoryRequest,
+    TestStatusSeriesRequest,
+)
 
 
 class TestClient(BaseClient):
@@ -24,3 +27,12 @@ def get_test_status_history(
         path = reverse("testStatusHistory")
         url = self.get_endpoint(path=path, query=query.model_dump())
         return requests.get(url)
+
+    def get_test_series(
+        self,
+        *,
+        query: TestStatusSeriesRequest,
+    ) -> requests.Response:
+        path = reverse("TestStatusSeries")
+        url = self.get_endpoint(path=path, query=query.model_dump())
+        return requests.get(url)

backend/kernelCI_app/tests/utils/fields/tests.py

@@ -45,3 +45,10 @@
     "start_time",
     "git_commit_hash",
 ]
+
+test_series_response_expected_fields = [
+    "status_history",
+    "regression_type",
+    "test_series",
+    "build_series",
+]