Update

[ghstack-poisoned]

Author: GregoryComer

backends/qualcomm/tests/tester.py

@@ -13,8 +13,8 @@
 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,
+    generate_qnn_executorch_compiler_spec,
     get_soc_to_chipset_map,
 )
 from executorch.backends.test.harness import Tester as TesterBase
@@ -36,7 +36,7 @@ def __init__(
         self,
         partitioners: Optional[List[Partitioner]] = None,
         edge_compile_config: Optional[EdgeCompileConfig] = None,
-        soc_model: str = "SM8650"
+        soc_model: str = "SM8650",
     ):
         backend_options = generate_htp_compiler_spec(use_fp16=True)
         self.chipset = get_soc_to_chipset_map()[soc_model]
@@ -47,7 +47,8 @@ def __init__(
 
         super().__init__(
             partitioners=partitioners or [QnnPartitioner(self.compiler_specs)],
-            edge_compile_config=edge_compile_config or EdgeCompileConfig(_check_ir_validity=False),
+            edge_compile_config=edge_compile_config
+            or EdgeCompileConfig(_check_ir_validity=False),
             default_partitioner_cls=QnnPartitioner,
         )
 
@@ -69,7 +70,7 @@ def __init__(
         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()

backends/test/suite/context.py

@@ -1,7 +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, test_base_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

backends/test/suite/flow.py

@@ -64,6 +64,7 @@ def all_flows() -> dict[str, TestFlow]:
 
     try:
         from executorch.backends.test.suite.flows.vulkan import VULKAN_TEST_FLOW
+
         flows += [
             VULKAN_TEST_FLOW,
         ]
@@ -72,6 +73,7 @@ def all_flows() -> dict[str, TestFlow]:
 
     try:
         from executorch.backends.test.suite.flows.qualcomm import QUALCOMM_TEST_FLOW
+
         flows += [
             QUALCOMM_TEST_FLOW,
         ]

backends/test/suite/flows/qualcomm.py

@@ -1,9 +1,10 @@
 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, 
+    name: str,
+    quantize: bool = False,
 ) -> TestFlow:
     return TestFlow(
         name,
@@ -12,4 +13,5 @@ def _create_qualcomm_flow(
         quantize=quantize,
     )
 
+
 QUALCOMM_TEST_FLOW = _create_qualcomm_flow("qualcomm")

backends/test/suite/flows/vulkan.py

@@ -1,9 +1,10 @@
-from executorch.backends.vulkan.test.tester import VulkanTester
 from executorch.backends.test.suite.flow import TestFlow
+from executorch.backends.vulkan.test.tester import VulkanTester
+
 
 def _create_vulkan_flow(
-    name: str, 
-    quantize: bool = False, 
+    name: str,
+    quantize: bool = False,
 ) -> TestFlow:
     return TestFlow(
         name,
@@ -12,4 +13,5 @@ def _create_vulkan_flow(
         quantize=quantize,
     )
 
+
 VULKAN_TEST_FLOW = _create_vulkan_flow("vulkan")

backends/test/suite/operators/__init__.py

@@ -117,7 +117,9 @@ def _create_test_for_backend(
 
     if test_type == TestType.STANDARD:
         test_name = f"{test_func.__name__}_{flow.name}"
-        wrapped_test = _make_wrapped_test(test_func, test_name, test_func.__name__, flow)
+        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:

backends/test/suite/reporting.py

@@ -1,11 +1,10 @@
+import csv
 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."""
@@ -79,13 +78,13 @@ class TestCaseSummary:
     """
     Contains summary results for the execution of a single test case.
     """
-    
+
     backend: str
     """ The name of the target backend. """
 
     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. """
 
@@ -173,8 +172,9 @@ def complete_test_session() -> RunSummary:
 
     return summary
 
+
 def generate_csv_report(summary: RunSummary, output: TextIO):
-    """ Write a run summary report to a file in CSV format. """
+    """Write a run summary report to a file in CSV format."""
 
     field_names = [
         "Test ID",
@@ -183,19 +183,23 @@ def generate_csv_report(summary: RunSummary, output: TextIO):
         "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()
+        (
+            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,
@@ -205,7 +209,5 @@ def generate_csv_report(summary: RunSummary, output: TextIO):
             "Result": record.result.display_name(),
         }
         if record.params is not None:
-            row.update({
-                k.capitalize(): v for k, v in record.params.items()
-            })
+            row.update({k.capitalize(): v for k, v in record.params.items()})
         writer.writerow(row)

backends/test/suite/runner.py

@@ -173,7 +173,10 @@ def parse_args():
         "-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."
+        "-r",
+        "--report",
+        nargs="?",
+        help="A file to write the test report to, in CSV format.",
     )
     return parser.parse_args()
 
@@ -202,7 +205,7 @@ 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}.")

backends/test/suite/tests/test_reporting.py

@@ -1,10 +1,18 @@
-import torch
 import unittest
 
 from csv import DictReader
-from ..reporting import TestResult, TestCaseSummary, RunSummary, TestSessionState, generate_csv_report
 from io import StringIO
 
+import torch
+
+from ..reporting import (
+    generate_csv_report,
+    RunSummary,
+    TestCaseSummary,
+    TestResult,
+    TestSessionState,
+)
+
 # Test data for simulated test results.
 TEST_CASE_SUMMARIES = [
     TestCaseSummary(
@@ -45,16 +53,17 @@
     ),
 ]
 
+
 class Reporting(unittest.TestCase):
     def test_csv_report_simple(self):
         # Verify the format of a simple CSV run report.
         session_state = TestSessionState()
         session_state.test_case_summaries.extend(TEST_CASE_SUMMARIES)
         run_summary = RunSummary.from_session(session_state)
-        
+
         strio = StringIO()
         generate_csv_report(run_summary, strio)
-        
+
         # Attempt to deserialize and validate the CSV report.
         report = DictReader(StringIO(strio.getvalue()))
         records = list(report)
@@ -95,7 +104,3 @@ def test_csv_report_simple(self):
         self.assertEqual(records[3]["Result"], "Fail (Export)")
         self.assertEqual(records[3]["Dtype"], "")
         self.assertEqual(records[3]["Use_dynamic_shapes"], "True")
-
-
-        
-        

backends/vulkan/test/tester.py

@@ -10,9 +10,9 @@
 import executorch.backends.test.harness.stages as BaseStages
 
 import torch
-from executorch.backends.vulkan.partitioner.vulkan_partitioner import VulkanPartitioner
 from executorch.backends.test.harness import Tester as TesterBase
 from executorch.backends.test.harness.stages import StageType
+from executorch.backends.vulkan.partitioner.vulkan_partitioner import VulkanPartitioner
 from executorch.exir import EdgeCompileConfig
 from executorch.exir.backend.partitioner import Partitioner
 
@@ -33,7 +33,8 @@ def __init__(
         super().__init__(
             default_partitioner_cls=VulkanPartitioner,
             partitioners=partitioners,
-            edge_compile_config=edge_compile_config or EdgeCompileConfig(_check_ir_validity=False),
+            edge_compile_config=edge_compile_config
+            or EdgeCompileConfig(_check_ir_validity=False),
         )