feat(api): add test run results endpoint

Add GET /api/v1/test-runs/{job_id}/results — paginated individual results for
a test-execution run, using the standard {items, page, limit, total} envelope.
Requires view permission on the run's project; not-found and not-accessible
both return 404, and only test-execution runs resolve (other job kinds 404).

Results are resource-shaped: test_type is the raw type code; result_status and
disposition are lowercase snake_case StrEnums shared between request filters
and responses. New api/enums.py maps the title-case DB values to the API
surface. disposition exposes a no_decision value for results with no triage
decision (NULL in the DB); omitting the filter returns active results
(confirmed plus no_decision), excluding dismissed and muted.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: rboni-dk

testgen/api/enums.py

@@ -0,0 +1,52 @@
+"""Lowercase presentation enums for API v1 request filters and responses.
+
+Shared home for StrEnums that map a DB-stored value to the lowercase snake_case
+form exposed through the API. Several DB columns store title-case values
+(``test_results.result_status``, ``*.disposition``); the mapping dicts here are the
+single seam that normalizes them to the API surface — the DB is never changed.
+"""
+
+from enum import StrEnum
+
+from testgen.common.enums import Disposition as DbDisposition
+from testgen.common.models.test_result import TestResultStatus
+
+
+class ResultStatus(StrEnum):
+    """Outcome of a single test result."""
+
+    passed = "passed"
+    failed = "failed"
+    warning = "warning"
+    error = "error"
+    log = "log"
+
+
+class Disposition(StrEnum):
+    """Triage state of a test result. ``no_decision`` is the state of a result no one
+    has triaged yet. Omitting the filter returns active results (``confirmed`` and
+    ``no_decision``); pass an explicit value to filter to a single state."""
+
+    confirmed = "confirmed"
+    dismissed = "dismissed"
+    muted = "muted"
+    no_decision = "no_decision"
+
+
+RESULT_STATUS_TO_DB: dict[ResultStatus, TestResultStatus] = {
+    ResultStatus.passed: TestResultStatus.Passed,
+    ResultStatus.failed: TestResultStatus.Failed,
+    ResultStatus.warning: TestResultStatus.Warning,
+    ResultStatus.error: TestResultStatus.Error,
+    ResultStatus.log: TestResultStatus.Log,
+}
+RESULT_STATUS_FROM_DB: dict[TestResultStatus, ResultStatus] = {v: k for k, v in RESULT_STATUS_TO_DB.items()}
+
+# ``no_decision`` has no stored DB value — it corresponds to a NULL ``disposition``
+# column, handled explicitly at the API boundary (not present in these dicts).
+DISPOSITION_TO_DB: dict[Disposition, DbDisposition] = {
+    Disposition.confirmed: DbDisposition.CONFIRMED,
+    Disposition.dismissed: DbDisposition.DISMISSED,
+    Disposition.muted: DbDisposition.INACTIVE,
+}
+DISPOSITION_FROM_DB: dict[DbDisposition, Disposition] = {v: k for k, v in DISPOSITION_TO_DB.items()}

testgen/api/runs.py

@@ -1,24 +1,35 @@
 """API v1 — test run and profiling run retrieval."""
 
-from fastapi import APIRouter, Depends
-from sqlalchemy import select
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import or_, select
 
 from testgen.api.deps import db_session, resolve_job
+from testgen.api.enums import (
+    DISPOSITION_FROM_DB,
+    DISPOSITION_TO_DB,
+    RESULT_STATUS_FROM_DB,
+    RESULT_STATUS_TO_DB,
+    Disposition,
+    ResultStatus,
+)
 from testgen.api.schemas import (
     ErrorResponse,
     IssueCounts,
     ProfilingRunResponse,
     ProfilingRunResult,
     ResultCounts,
+    TestResultItem,
+    TestResultListResponse,
     TestRunResponse,
     TestRunResult,
 )
+from testgen.common.enums import Disposition as DbDisposition
 from testgen.common.enums import JobKey
 from testgen.common.models import get_current_session
 from testgen.common.models.hygiene_issue import HygieneIssue
 from testgen.common.models.job_execution import JobExecution
 from testgen.common.models.profiling_run import ProfilingRun
-from testgen.common.models.test_result import TestResult
+from testgen.common.models.test_result import TestResult, TestRunResultRow
 from testgen.common.models.test_run import TestRun
 from testgen.common.models.test_suite import TestSuite
 
@@ -63,6 +74,68 @@ def get_test_run(job: JobExecution = resolve_job("view", JobExecution.job_key ==
     )
 
 
+def _to_item(row: TestRunResultRow) -> TestResultItem:
+    """Map a DB-valued result row to the API item, normalizing enum casing."""
+    return TestResultItem(
+        test_definition_id=row.test_definition_id,
+        test_type=row.test_type,
+        schema_name=row.schema_name,
+        table_name=row.table_name,
+        column_names=row.column_names,
+        result_status=RESULT_STATUS_FROM_DB.get(row.status),
+        result_measure=row.result_measure,
+        threshold_value=row.threshold_value,
+        result_message=row.message,
+        test_time=row.test_time,
+        disposition=DISPOSITION_FROM_DB[DbDisposition(row.disposition)] if row.disposition else Disposition.no_decision,
+    )
+
+
+@router.get(
+    "/test-runs/{job_id}/results",
+    response_model=TestResultListResponse,
+)
+def list_test_run_results(
+    job: JobExecution = resolve_job("view", JobExecution.job_key == JobKey.run_tests),  # noqa: B008
+    status: ResultStatus | None = Query(default=None),  # noqa: B008
+    table_name: str | None = Query(default=None),
+    column_name: str | None = Query(default=None),
+    test_type: str | None = Query(default=None),
+    disposition: Disposition | None = Query(default=None),  # noqa: B008
+    page: int = Query(default=1, ge=1),
+    limit: int = Query(default=20, ge=1, le=100),
+):
+    """List individual results for a test run.
+
+    Omitting ``disposition`` returns active results — confirmed and no_decision
+    (excludes dismissed and muted); pass an explicit value to filter to one state.
+    """
+    clauses = []
+    if status:
+        clauses.append(TestResult.status == RESULT_STATUS_TO_DB[status])
+    if table_name:
+        clauses.append(TestResult.table_name == table_name)
+    if column_name:
+        clauses.append(TestResult.column_names == column_name)
+    if test_type:
+        clauses.append(TestResult.test_type == test_type)
+    if disposition is None:
+        # Active: confirmed plus no_decision (NULL). Dismissed/muted excluded.
+        clauses.append(or_(TestResult.disposition.is_(None), TestResult.disposition == DbDisposition.CONFIRMED.value))
+    elif disposition == Disposition.no_decision:
+        clauses.append(TestResult.disposition.is_(None))
+    else:
+        clauses.append(TestResult.disposition == DISPOSITION_TO_DB[disposition].value)
+
+    items, total = TestResult.list_for_run(job.id, *clauses, page=page, limit=limit)
+    return TestResultListResponse(
+        items=[_to_item(row) for row in items],
+        page=page,
+        limit=limit,
+        total=total,
+    )
+
+
 @router.get(
     "/profiling-runs/{job_id}",
     response_model=ProfilingRunResponse,

testgen/api/schemas.py

@@ -5,6 +5,7 @@
 
 from pydantic import BaseModel
 
+from testgen.api.enums import Disposition, ResultStatus
 from testgen.common.enums import JobSource, JobStatus, PublicJobKey
 from testgen.common.test_definition_export_import_service import ImportConfig, ImportPayload, ImportResponse
 
@@ -78,6 +79,31 @@ class TestRunResponse(BaseModel):
     result: TestRunResult | None = None
 
 
+class TestResultItem(BaseModel):
+    """One individual test result within a test run."""
+
+    test_definition_id: UUID
+    test_type: str
+    schema_name: str
+    table_name: str | None = None
+    column_names: str | None = None
+    result_status: ResultStatus | None = None
+    result_measure: str | None = None
+    threshold_value: str | None = None
+    result_message: str | None = None
+    test_time: datetime | None = None
+    disposition: Disposition
+
+
+class TestResultListResponse(BaseModel):
+    """Paginated list of individual test results."""
+
+    items: list[TestResultItem]
+    page: int
+    limit: int
+    total: int
+
+
 # --- Profiling Runs ---
 
 

testgen/common/models/test_result.py

@@ -64,6 +64,23 @@ class TestResultSearchRow:
     result_message: str | None
 
 
+@dataclass
+class TestRunResultRow:
+    """One individual result within a single test run for the API results endpoint."""
+
+    test_definition_id: UUID
+    test_type: str
+    schema_name: str
+    table_name: str | None
+    column_names: str | None
+    status: TestResultStatus | None
+    result_measure: str | None
+    threshold_value: str | None
+    message: str | None
+    test_time: datetime | None
+    disposition: str | None
+
+
 @dataclass
 class TrendBucket:
     """One time-bucket of failure aggregates for ``get_failure_trend``."""
@@ -175,6 +192,38 @@ def select_results(
         query = query.order_by(cls.status, cls.table_name, cls.column_names).offset(offset).limit(limit)
         return get_current_session().scalars(query).all()
 
+    @classmethod
+    def list_for_run(
+        cls,
+        test_run_id: UUID,
+        *clauses,
+        page: int = 1,
+        limit: int = 20,
+    ) -> tuple[list[TestRunResultRow], int]:
+        """Paginated individual results for a single run, scoped by caller-supplied WHERE clauses.
+
+        Monitor suites are always filtered out.
+        """
+        query = (
+            select(
+                cls.test_definition_id.label("test_definition_id"),
+                cls.test_type.label("test_type"),
+                cls.schema_name.label("schema_name"),
+                cls.table_name.label("table_name"),
+                cls.column_names.label("column_names"),
+                cls.status.label("status"),
+                cls.result_measure.label("result_measure"),
+                cls.threshold_value.label("threshold_value"),
+                cls.message.label("message"),
+                cls.test_time.label("test_time"),
+                cls.disposition.label("disposition"),
+            )
+            .join(TestSuite, cls.test_suite_id == TestSuite.id)
+            .where(cls.test_run_id == test_run_id, TestSuite.is_monitor.isnot(True), *clauses)
+            .order_by(cls.status, cls.table_name, cls.column_names, cls.id)
+        )
+        return cls._paginate(query, page=page, limit=limit, data_class=TestRunResultRow)
+
     @classmethod
     def select_failures(
         cls,