[Backend Tester] Add CSV report generation

backends/qualcomm/tests/TARGETS

@@ -37,3 +37,13 @@ python_library(
         "//executorch/backends/qualcomm/debugger:utils",
     ],
 )
+
+python_library(
+    name = "tester",
+    srcs = [
+        "tester.py",
+    ],
+    deps = [
+        ":test_qnn_delegate"
+    ]
+)

backends/qualcomm/tests/tester.py

@@ -0,0 +1,87 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+from typing import Any, List, Optional, Tuple
+
+import executorch
+import executorch.backends.test.harness.stages as BaseStages
+
+import torch
+from executorch.backends.qualcomm._passes.qnn_pass_manager import QnnPassManager
+from executorch.backends.qualcomm.partition.qnn_partitioner import QnnPartitioner
+from executorch.backends.qualcomm.utils.utils import (
+    generate_qnn_executorch_compiler_spec,
+    generate_htp_compiler_spec,
+    get_soc_to_chipset_map,
+)
+from executorch.backends.test.harness import Tester as TesterBase
+from executorch.backends.test.harness.stages import StageType
+from executorch.exir import EdgeCompileConfig, to_edge_transform_and_lower
+from executorch.exir.backend.partitioner import Partitioner
+from torch.export import ExportedProgram
+
+
+class Partition(BaseStages.Partition):
+    def __init__(self, partitioner: Optional[Partitioner] = None):
+        super().__init__(
+            partitioner=partitioner or QnnPartitioner,
+        )
+
+
+class ToEdgeTransformAndLower(BaseStages.ToEdgeTransformAndLower):
+    def __init__(
+        self,
+        partitioners: Optional[List[Partitioner]] = None,
+        edge_compile_config: Optional[EdgeCompileConfig] = None,
+        soc_model: str = "SM8650"
+    ):
+        backend_options = generate_htp_compiler_spec(use_fp16=True)
+        self.chipset = get_soc_to_chipset_map()[soc_model]
+        self.compiler_specs = generate_qnn_executorch_compiler_spec(
+            soc_model=self.chipset,
+            backend_options=backend_options,
+        )
+
+        super().__init__(
+            partitioners=partitioners or [QnnPartitioner(self.compiler_specs)],
+            edge_compile_config=edge_compile_config or EdgeCompileConfig(_check_ir_validity=False),
+            default_partitioner_cls=QnnPartitioner,
+        )
+
+    def run(self, artifact: ExportedProgram, inputs=None) -> None:
+        ep = QnnPassManager().transform_for_export_pipeline(artifact)
+        transform_passes = QnnPassManager().get_to_edge_transform_passes(ep)
+
+        self.edge_dialect_program = to_edge_transform_and_lower(
+            ep,
+            transform_passes=transform_passes,
+            partitioner=self.partitioners,
+            compile_config=self.edge_compile_conf,
+        )
+
+
+class QualcommTester(TesterBase):
+    def __init__(
+        self,
+        module: torch.nn.Module,
+        example_inputs: Tuple[torch.Tensor],
+        dynamic_shapes: Optional[Tuple[Any]] = None,
+    ): 
+        # Specialize for Qualcomm
+        stage_classes = (
+            executorch.backends.test.harness.Tester.default_stage_classes()
+            | {
+                StageType.PARTITION: Partition,
+                StageType.TO_EDGE_TRANSFORM_AND_LOWER: ToEdgeTransformAndLower,
+            }
+        )
+
+        super().__init__(
+            module=module,
+            stage_classes=stage_classes,
+            example_inputs=example_inputs,
+            dynamic_shapes=dynamic_shapes,
+        )

backends/test/suite/context.py

@@ -1,8 +1,9 @@
 # Test run context management. This is used to determine the test context for reporting
 # purposes.
 class TestContext:
-    def __init__(self, test_name: str, flow_name: str, params: dict | None):
+    def __init__(self, test_name: str, test_base_name: str, flow_name: str, params: dict | None):
         self.test_name = test_name
+        self.test_base_name = test_base_name
         self.flow_name = flow_name
         self.params = params
 

backends/test/suite/flow.py

@@ -62,4 +62,20 @@ def all_flows() -> dict[str, TestFlow]:
     except Exception as e:
         logger.info(f"Skipping Core ML flow registration: {e}")
 
+    try:
+        from executorch.backends.test.suite.flows.vulkan import VULKAN_TEST_FLOW
+        flows += [
+            VULKAN_TEST_FLOW,
+        ]
+    except Exception as e:
+        logger.info(f"Skipping Vulkan flow registration: {e}")
+
+    try:
+        from executorch.backends.test.suite.flows.qualcomm import QUALCOMM_TEST_FLOW
+        flows += [
+            QUALCOMM_TEST_FLOW,
+        ]
+    except Exception as e:
+        logger.info(f"Skipping Qualcomm flow registration: {e}")
+
     return {f.name: f for f in flows if f is not None}

backends/test/suite/flows/qualcomm.py

@@ -0,0 +1,15 @@
+from executorch.backends.qualcomm.tests.tester import QualcommTester
+from executorch.backends.test.suite.flow import TestFlow
+
+def _create_qualcomm_flow(
+    name: str, 
+    quantize: bool = False, 
+) -> TestFlow:
+    return TestFlow(
+        name,
+        backend="qualcomm",
+        tester_factory=QualcommTester,
+        quantize=quantize,
+    )
+
+QUALCOMM_TEST_FLOW = _create_qualcomm_flow("qualcomm")

backends/test/suite/flows/vulkan.py

@@ -0,0 +1,15 @@
+from executorch.backends.vulkan.test.tester import VulkanTester
+from executorch.backends.test.suite.flow import TestFlow
+
+def _create_vulkan_flow(
+    name: str, 
+    quantize: bool = False, 
+) -> TestFlow:
+    return TestFlow(
+        name,
+        backend="vulkan",
+        tester_factory=VulkanTester,
+        quantize=quantize,
+    )
+
+VULKAN_TEST_FLOW = _create_vulkan_flow("vulkan")

backends/test/suite/models/__init__.py

@@ -42,19 +42,19 @@ def _create_test(
     dtype: torch.dtype,
     use_dynamic_shapes: bool,
 ):
+    dtype_name = str(dtype)[6:]  # strip "torch."
+    test_name = f"{test_func.__name__}_{flow.name}_{dtype_name}"
+    if use_dynamic_shapes:
+        test_name += "_dynamic_shape"
+
     def wrapped_test(self):
         params = {
             "dtype": dtype,
             "use_dynamic_shapes": use_dynamic_shapes,
         }
-        with TestContext(test_name, flow.name, params):
+        with TestContext(test_name, test_func.__name__, flow.name, params):
             test_func(self, flow, dtype, use_dynamic_shapes)
 
-    dtype_name = str(dtype)[6:]  # strip "torch."
-    test_name = f"{test_func.__name__}_{flow.name}_{dtype_name}"
-    if use_dynamic_shapes:
-        test_name += "_dynamic_shape"
-
     wrapped_test._name = test_func.__name__  # type: ignore
     wrapped_test._flow = flow  # type: ignore
 
@@ -118,6 +118,7 @@ def run_model_test(
         inputs,
         flow,
         context.test_name,
+        context.test_base_name,
         context.params,
         dynamic_shapes=dynamic_shapes,
     )

backends/test/suite/operators/__init__.py

@@ -6,6 +6,7 @@
 
 # pyre-unsafe
 
+import copy
 import os
 import unittest
 
@@ -90,12 +91,13 @@ def _expand_test(cls, test_name: str):
 def _make_wrapped_test(
     test_func: Callable,
     test_name: str,
+    test_base_name: str,
     flow: TestFlow,
     params: dict | None = None,
 ):
     def wrapped_test(self):
-        with TestContext(test_name, flow.name, params):
-            test_kwargs = params or {}
+        with TestContext(test_name, test_base_name, flow.name, params):
+            test_kwargs = copy.copy(params) or {}
             test_kwargs["flow"] = flow
 
             test_func(self, **test_kwargs)
@@ -114,19 +116,20 @@ def _create_test_for_backend(
     test_type = getattr(test_func, "test_type", TestType.STANDARD)
 
     if test_type == TestType.STANDARD:
-        wrapped_test = _make_wrapped_test(test_func, test_func.__name__, flow)
         test_name = f"{test_func.__name__}_{flow.name}"
+        wrapped_test = _make_wrapped_test(test_func, test_name, test_func.__name__, flow)
         setattr(cls, test_name, wrapped_test)
     elif test_type == TestType.DTYPE:
         for dtype in DTYPES:
+            dtype_name = str(dtype)[6:]  # strip "torch."
+            test_name = f"{test_func.__name__}_{dtype_name}_{flow.name}"
             wrapped_test = _make_wrapped_test(
                 test_func,
+                test_name,
                 test_func.__name__,
                 flow,
                 {"dtype": dtype},
             )
-            dtype_name = str(dtype)[6:]  # strip "torch."
-            test_name = f"{test_func.__name__}_{dtype_name}_{flow.name}"
             setattr(cls, test_name, wrapped_test)
     else:
         raise NotImplementedError(f"Unknown test type {test_type}.")
@@ -144,6 +147,7 @@ def _test_op(self, model, inputs, flow: TestFlow):
             inputs,
             flow,
             context.test_name,
+            context.test_base_name,
             context.params,
         )
 

backends/test/suite/reporting.py

@@ -1,7 +1,11 @@
 from collections import Counter
 from dataclasses import dataclass
 from enum import IntEnum
+from functools import reduce
+from re import A
+from typing import TextIO
 
+import csv
 
 class TestResult(IntEnum):
     """Represents the result of a test case run, indicating success or a specific failure reason."""
@@ -75,13 +79,19 @@ class TestCaseSummary:
     """
     Contains summary results for the execution of a single test case.
     """
+
+    backend: str
+    """ The name of the target backend. """
 
-    name: str
-    """ The qualified name of the test, not including the flow suffix. """
-
+    base_name: str
+    """ The base name of the test, not including flow or parameter suffixes. """
+    
     flow: str
     """ The backend-specific flow name. Corresponds to flows registered in backends/test/suite/__init__.py. """
 
+    name: str
+    """ The full name of test, including flow and parameter suffixes. """
+
     params: dict | None
     """ Test-specific parameters, such as dtype. """
 
@@ -162,3 +172,40 @@ def complete_test_session() -> RunSummary:
     _active_session = None
 
     return summary
+
+def generate_csv_report(summary: RunSummary, output: TextIO):
+    """ Write a run summary report to a file in CSV format. """
+
+    field_names = [
+        "Test ID",
+        "Test Case",
+        "Backend",
+        "Flow",
+        "Result",
+    ]
+
+    # Tests can have custom parameters. We'll want to report them here, so we need
+    # a list of all unique parameter names.
+    param_names = reduce(
+        lambda a, b: a.union(b),
+        (set(s.params.keys()) for s in summary.test_case_summaries if s.params is not None),
+        set()
+    )
+    field_names += (s.capitalize() for s in param_names)
+
+    writer = csv.DictWriter(output, field_names)
+    writer.writeheader()
+
+    for record in summary.test_case_summaries:
+        row = {
+            "Test ID": record.name,
+            "Test Case": record.base_name,
+            "Backend": record.backend,
+            "Flow": record.flow,
+            "Result": record.result.display_name(),
+        }
+        if record.params is not None:
+            row.update({
+                k.capitalize(): v for k, v in record.params.items()
+            })
+        writer.writerow(row)

backends/test/suite/runner.py

@@ -13,6 +13,7 @@
 from executorch.backends.test.suite.reporting import (
     begin_test_session,
     complete_test_session,
+    generate_csv_report,
     RunSummary,
     TestCaseSummary,
     TestResult,
@@ -31,6 +32,7 @@ def run_test(  # noqa: C901
     inputs: Any,
     flow: TestFlow,
     test_name: str,
+    test_base_name: str,
     params: dict | None,
     dynamic_shapes: Any | None = None,
 ) -> TestCaseSummary:
@@ -44,8 +46,10 @@ def build_result(
         result: TestResult, error: Exception | None = None
     ) -> TestCaseSummary:
         return TestCaseSummary(
-            name=test_name,
+            backend=flow.backend,
+            base_name=test_base_name,
             flow=flow.name,
+            name=test_name,
             params=params,
             result=result,
             error=error,
@@ -168,6 +172,9 @@ def parse_args():
     parser.add_argument(
         "-f", "--filter", nargs="?", help="A regular expression filter for test names."
     )
+    parser.add_argument(
+        "-r", "--report", nargs="?", help="A file to write the test report to, in CSV format."
+    )
     return parser.parse_args()
 
 
@@ -195,6 +202,11 @@ def runner_main():
 
     summary = complete_test_session()
     print_summary(summary)
+
+    if args.report is not None:
+        with open(args.report, "w") as f:
+            print(f"Writing CSV report to {args.report}.")
+            generate_csv_report(summary, f)
 
 
 if __name__ == "__main__":

backends/test/suite/tests/README.md

@@ -0,0 +1,3 @@
+# Tests
+
+This directory contains meta-tests for the backend test suite. As the test suite contains a non-neglible amount of logic, these tests are useful to ensure that the test suite itself is working correctly.