feat: add hardware selectors and revision API

Author: Gustavo Flores

backend/kernelCI_app/queries/hardware.py

@@ -179,6 +179,134 @@ def get_hardware_listing_data(
         return cursor.fetchall()
 
 
+def get_hardware_selectors(origin: str) -> list[dict]:
+    params = {"origin": origin}
+
+    query = """
+        WITH qualified_revisions AS (
+            SELECT
+                c.tree_name,
+                c.git_repository_url,
+                c.git_repository_branch,
+                c.git_commit_hash,
+                c.git_commit_name,
+                MAX(c.start_time) AS latest_start_time
+            FROM
+                checkouts c
+                INNER JOIN builds b ON b.checkout_id = c.id
+                INNER JOIN tests ON tests.build_id = b.id
+            WHERE
+                tests.origin = %(origin)s
+                AND tests.environment_misc ->> 'platform' IS NOT NULL
+            GROUP BY
+                c.tree_name,
+                c.git_repository_url,
+                c.git_repository_branch,
+                c.git_commit_hash,
+                c.git_commit_name
+        )
+        SELECT
+            qr.tree_name,
+            qr.git_repository_url,
+            qr.git_repository_branch,
+            qr.git_commit_hash,
+            qr.git_commit_name,
+            qr.latest_start_time,
+            MAX(qr.latest_start_time) OVER (
+                PARTITION BY
+                    qr.tree_name,
+                    qr.git_repository_url,
+                    qr.git_repository_branch
+            ) AS branch_latest_start_time
+        FROM
+            qualified_revisions qr
+        ORDER BY
+            qr.tree_name ASC,
+            branch_latest_start_time DESC,
+            qr.latest_start_time DESC
+    """
+
+    with connection.cursor() as cursor:
+        cursor.execute(query, params)
+        return dict_fetchall(cursor)
+
+
+def get_hardware_listing_data_by_revision(
+    *,
+    origin: str,
+    tree_name: str,
+    git_repository_url: str,
+    git_repository_branch: str,
+    git_commit_hash: str,
+) -> list[tuple]:
+    count_clauses = _get_hardware_listing_count_clauses()
+    params = {
+        "origin": origin,
+        "tree_name": tree_name,
+        "git_repository_url": git_repository_url,
+        "git_repository_branch": git_repository_branch,
+        "git_commit_hash": git_commit_hash,
+    }
+
+    query = f"""
+        WITH relevant_tests AS (
+            SELECT
+                tests.environment_compatible,
+                tests.environment_misc ->> 'platform' AS platform,
+                tests.status,
+                tests.path,
+                tests.id,
+                b.id AS build_id,
+                b.status AS build_status
+            FROM
+                checkouts c
+                INNER JOIN builds b ON b.checkout_id = c.id
+                INNER JOIN tests ON tests.build_id = b.id
+            WHERE
+                c.tree_name = %(tree_name)s
+                AND c.git_repository_url = %(git_repository_url)s
+                AND c.git_repository_branch = %(git_repository_branch)s
+                AND c.git_commit_hash = %(git_commit_hash)s
+                AND tests.origin = %(origin)s
+                AND tests.environment_misc ->> 'platform' IS NOT NULL
+        ),
+        compatible_values AS (
+            SELECT
+                platform,
+                UNNEST(environment_compatible) AS compatible
+            FROM
+                relevant_tests
+            WHERE
+                environment_compatible IS NOT NULL
+        ),
+        compatible_agg AS (
+            SELECT
+                platform,
+                ARRAY_AGG(DISTINCT compatible ORDER BY compatible) AS hardware
+            FROM
+                compatible_values
+            GROUP BY
+                platform
+        )
+        SELECT
+            relevant_tests.platform,
+            compatible_agg.hardware,
+            {count_clauses}
+        FROM
+            relevant_tests
+            LEFT JOIN compatible_agg ON compatible_agg.platform = relevant_tests.platform
+        GROUP BY
+            relevant_tests.platform,
+            compatible_agg.hardware
+        ORDER BY
+            relevant_tests.platform ASC
+    """
+
+    with connection.cursor() as cursor:
+        cursor.execute(query, params)
+        return cursor.fetchall()
+
+
 def get_hardware_listing_data_bulk(
     keys: list[tuple[str, str]],
     start_date: datetime,

backend/kernelCI_app/tests/unitTests/url_patterns_test.py

@@ -133,3 +133,11 @@ def test_tree_commit_hash_patterns(self):
         url = reverse("treeDetailsSummaryView", kwargs={"commit_hash": commit_hash})
         resolved = resolve(url)
         assert resolved.kwargs["commit_hash"] == commit_hash
+
+    def test_hardware_selectors_route(self):
+        url = reverse("hardwareSelectors")
+        resolved = resolve(url)
+
+        assert url == "/api/hardware/selectors/"
+        assert resolved.url_name == "hardwareSelectors"
+        assert "hardware_id" not in resolved.kwargs

backend/kernelCI_app/tests/unitTests/views/hardwareByRevisionView_test.py

@@ -0,0 +1,79 @@
+from http import HTTPStatus
+from unittest.mock import patch
+
+from django.test.testcases import SimpleTestCase
+from rest_framework.test import APIRequestFactory
+
+from kernelCI_app.views.hardwareByRevisionView import HardwareByRevisionView
+
+
+class TestHardwareByRevisionView(SimpleTestCase):
+    def setUp(self):
+        self.factory = APIRequestFactory()
+        self.view = HardwareByRevisionView()
+        self.url = "/hardware-by-revision"
+        self.required_query_params = {
+            "origin": "origin1",
+            "tree_name": "mainline",
+            "git_repository_url": "https://example.com/linux.git",
+            "git_repository_branch": "master",
+            "git_commit_hash": "abc123",
+        }
+
+    @patch(
+        "kernelCI_app.views.hardwareByRevisionView.get_hardware_listing_data_by_revision"
+    )
+    def test_get_hardware_listing_by_revision_success(
+        self, mock_get_hardware_listing_data_by_revision
+    ):
+        mock_get_hardware_listing_data_by_revision.return_value = [
+            ("platform1", ["hardware1"], *range(22)),
+        ]
+
+        request = self.factory.get(self.url, self.required_query_params)
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        self.assertEqual(response.data["hardware"][0]["platform"], "platform1")
+
+    def test_get_hardware_listing_by_revision_missing_query_params_returns_bad_request(
+        self,
+    ):
+        request = self.factory.get(self.url, {"origin": "origin1"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.BAD_REQUEST)
+        self.assertIn("tree_name", response.data)
+        self.assertIn("git_repository_url", response.data)
+        self.assertIn("git_repository_branch", response.data)
+        self.assertIn("git_commit_hash", response.data)
+
+    @patch(
+        "kernelCI_app.views.hardwareByRevisionView.get_hardware_listing_data_by_revision"
+    )
+    def test_get_hardware_listing_by_revision_empty_response(
+        self, mock_get_hardware_listing_data_by_revision
+    ):
+        mock_get_hardware_listing_data_by_revision.return_value = []
+
+        request = self.factory.get(self.url, self.required_query_params)
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        self.assertEqual(response.data, {"hardware": []})
+
+    @patch(
+        "kernelCI_app.views.hardwareByRevisionView.get_hardware_listing_data_by_revision"
+    )
+    def test_get_hardware_listing_by_revision_sanitize_validation_error_returns_internal_server_error(
+        self, mock_get_hardware_listing_data_by_revision
+    ):
+        mock_get_hardware_listing_data_by_revision.return_value = [
+            (None, "hardware1", *range(22)),
+        ]
+
+        request = self.factory.get(self.url, self.required_query_params)
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.INTERNAL_SERVER_ERROR)
+        self.assertIn("platform", response.data)

backend/kernelCI_app/tests/unitTests/views/hardwareSelectorsView_test.py

@@ -0,0 +1,84 @@
+from datetime import datetime
+from http import HTTPStatus
+from unittest.mock import patch
+
+from django.test.testcases import SimpleTestCase
+from rest_framework.test import APIRequestFactory
+
+from kernelCI_app.constants.general import DEFAULT_ORIGIN
+from kernelCI_app.views.hardwareSelectorsView import HardwareSelectorsView
+
+
+class TestHardwareSelectorsView(SimpleTestCase):
+    def setUp(self):
+        self.factory = APIRequestFactory()
+        self.view = HardwareSelectorsView()
+        self.url = "/hardware/selectors"
+
+    @patch("kernelCI_app.views.hardwareSelectorsView.get_hardware_selectors")
+    def test_get_hardware_selectors_success(self, mock_get_hardware_selectors):
+        mock_get_hardware_selectors.return_value = [
+            {
+                "tree_name": "mainline",
+                "git_repository_url": "https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git",
+                "git_repository_branch": "master",
+                "git_commit_hash": "abc123",
+                "git_commit_name": "v6.12-rc1",
+                "latest_start_time": datetime(2026, 1, 1, 10, 0, 0),
+                "branch_latest_start_time": datetime(2026, 1, 1, 10, 0, 0),
+            }
+        ]
+
+        request = self.factory.get(self.url, {"origin": "origin1"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        self.assertEqual(response.data["trees"][0]["tree_name"], "mainline")
+        self.assertEqual(
+            response.data["trees"][0]["branches"][0]["revisions"][0]["git_commit_hash"],
+            "abc123",
+        )
+        mock_get_hardware_selectors.assert_called_once_with(origin="origin1")
+
+    @patch("kernelCI_app.views.hardwareSelectorsView.get_hardware_selectors")
+    def test_get_hardware_selectors_defaults_origin(self, mock_get_hardware_selectors):
+        mock_get_hardware_selectors.return_value = []
+
+        request = self.factory.get(self.url)
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        self.assertEqual(response.data, {"trees": []})
+        mock_get_hardware_selectors.assert_called_once_with(origin=DEFAULT_ORIGIN)
+
+    @patch("kernelCI_app.views.hardwareSelectorsView.get_hardware_selectors")
+    def test_get_hardware_selectors_empty_response(self, mock_get_hardware_selectors):
+        mock_get_hardware_selectors.return_value = []
+
+        request = self.factory.get(self.url, {"origin": "origin1"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.OK)
+        self.assertEqual(response.data, {"trees": []})
+
+    @patch("kernelCI_app.views.hardwareSelectorsView.get_hardware_selectors")
+    def test_get_hardware_selectors_sanitize_validation_error_returns_internal_server_error(
+        self, mock_get_hardware_selectors
+    ):
+        mock_get_hardware_selectors.return_value = [
+            {
+                "tree_name": "mainline",
+                "git_repository_url": "https://example.com/linux.git",
+                "git_repository_branch": "master",
+                "git_commit_hash": None,
+                "git_commit_name": "v6.12-rc1",
+                "latest_start_time": datetime(2026, 1, 1, 10, 0, 0),
+                "branch_latest_start_time": datetime(2026, 1, 1, 10, 0, 0),
+            }
+        ]
+
+        request = self.factory.get(self.url, {"origin": "origin1"})
+        response = self.view.get(request)
+
+        self.assertEqual(response.status_code, HTTPStatus.INTERNAL_SERVER_ERROR)
+        self.assertIn("git_commit_hash", response.data)

backend/kernelCI_app/typeModels/hardwareListingByRevision.py

@@ -0,0 +1,36 @@
+from typing import Annotated
+
+from pydantic import BaseModel, BeforeValidator, Field
+
+from kernelCI_app.constants.general import DEFAULT_ORIGIN
+from kernelCI_app.constants.localization import DocStrings
+
+
+class HardwareListingByRevisionQueryParamsDocumentationOnly(BaseModel):
+    origin: Annotated[
+        str,
+        Field(
+            default=DEFAULT_ORIGIN,
+            description=DocStrings.HARDWARE_LISTING_ORIGIN_DESCRIPTION,
+        ),
+    ]
+    tree_name: str = Field(description=DocStrings.TREE_NAME_PATH_DESCRIPTION)
+    git_repository_url: str = Field(
+        description=DocStrings.TREE_QUERY_GIT_URL_DESCRIPTION
+    )
+    git_repository_branch: str = Field(
+        description=DocStrings.DEFAULT_GIT_BRANCH_DESCRIPTION
+    )
+    git_commit_hash: str = Field(description=DocStrings.COMMIT_HASH_PATH_DESCRIPTION)
+
+
+class HardwareListingByRevisionQueryParams(BaseModel):
+    origin: Annotated[
+        str,
+        Field(default=DEFAULT_ORIGIN),
+        BeforeValidator(lambda o: DEFAULT_ORIGIN if o is None else o),
+    ]
+    tree_name: str
+    git_repository_url: str
+    git_repository_branch: str
+    git_commit_hash: str

backend/kernelCI_app/typeModels/hardwareSelectors.py

@@ -0,0 +1,48 @@
+from datetime import datetime
+from typing import Annotated
+
+from pydantic import BaseModel, BeforeValidator, Field
+
+from kernelCI_app.constants.general import DEFAULT_ORIGIN
+from kernelCI_app.constants.localization import DocStrings
+
+
+class HardwareSelectorRevision(BaseModel):
+    git_commit_hash: str
+    git_commit_name: str | None = None
+    start_time: datetime
+
+
+class HardwareSelectorBranch(BaseModel):
+    git_repository_url: str
+    git_repository_branch: str
+    latest_start_time: datetime
+    revisions: list[HardwareSelectorRevision]
+
+
+class HardwareSelectorTree(BaseModel):
+    tree_name: str
+    latest_start_time: datetime
+    branches: list[HardwareSelectorBranch]
+
+
+class HardwareSelectorsResponse(BaseModel):
+    trees: list[HardwareSelectorTree]
+
+
+class HardwareSelectorsQueryParamsDocumentationOnly(BaseModel):
+    origin: Annotated[
+        str,
+        Field(
+            default=DEFAULT_ORIGIN,
+            description=DocStrings.HARDWARE_LISTING_ORIGIN_DESCRIPTION,
+        ),
+    ]
+
+
+class HardwareSelectorsQueryParams(BaseModel):
+    origin: Annotated[
+        str,
+        Field(default=DEFAULT_ORIGIN),
+        BeforeValidator(lambda o: DEFAULT_ORIGIN if o is None else o),
+    ]