feat: add not supported status code

fluster/decoder.py

@@ -24,6 +24,14 @@
 from fluster.utils import normalize_binary_cmd
 
 
+class NotSupportedError(Exception):
+    """Decoder cannot handle the media."""
+
+    def __init__(self, message: str = "Media not supported by decoder"):
+        self.message = message
+        super().__init__(self.message)
+
+
 class Decoder(ABC):
     """Base class for decoders"""
 

fluster/decoders/vk_video_decoder.py

@@ -14,12 +14,16 @@
 #
 # You should have received a copy of the GNU Lesser General Public
 # License along with this library. If not, see <https://www.gnu.org/licenses/>.
+import subprocess
 from typing import Any, Dict, Optional
 
 from fluster.codec import Codec, OutputFormat
-from fluster.decoder import Decoder, register_decoder
+from fluster.decoder import Decoder, NotSupportedError, register_decoder
 from fluster.utils import file_checksum, run_command
 
+# BSD sysexits.h EX_UNAVAILABLE — media not supported.
+EX_UNAVAILABLE = 69
+
 
 class VKVSDecoder(Decoder):
     """NVidia vk_video_samples decoder implementation"""
@@ -48,22 +52,27 @@ def decode(
             Codec.AV1: "av1",
             Codec.VP9: "vp9",
         }
-        run_command(
-            [
-                self.binary,
-                "-i",
-                input_filepath,
-                "-o",
-                output_filepath,
-                "--codec",
-                codec_mapping[self.codec],
-                "--noPresent",
-                "--enablePostProcessFilter",
-                "0",
-            ],
-            timeout=timeout,
-            verbose=verbose,
-        )
+        try:
+            run_command(
+                [
+                    self.binary,
+                    "-i",
+                    input_filepath,
+                    "-o",
+                    output_filepath,
+                    "--codec",
+                    codec_mapping[self.codec],
+                    "--noPresent",
+                    "--enablePostProcessFilter",
+                    "0",
+                ],
+                timeout=timeout,
+                verbose=verbose,
+            )
+        except subprocess.CalledProcessError as ex:
+            if ex.returncode == EX_UNAVAILABLE:
+                raise NotSupportedError(f"Media not supported: {ex.cmd}") from ex
+            raise
         return file_checksum(output_filepath)
 
 

fluster/fluster.py

@@ -109,6 +109,7 @@ def to_test_suite_context(
     TestVectorResult.FAIL: "❌",
     TestVectorResult.TIMEOUT: "⌛",
     TestVectorResult.ERROR: "☠",
+    TestVectorResult.NOT_SUPPORTED: "○",
 }
 
 TEXT_RESULT = {
@@ -117,6 +118,7 @@ def to_test_suite_context(
     TestVectorResult.FAIL: "KO",
     TestVectorResult.TIMEOUT: "TO",
     TestVectorResult.ERROR: "ER",
+    TestVectorResult.NOT_SUPPORTED: "NS",
 }
 
 RESULT_MAP = {
@@ -126,6 +128,7 @@ def to_test_suite_context(
     TestVectorResult.ERROR: "Error",
     TestVectorResult.FAIL: "Fail",
     TestVectorResult.NOT_RUN: "Not run",
+    TestVectorResult.NOT_SUPPORTED: "Not supported",
 }
 
 
@@ -401,8 +404,8 @@ def _parse_suite_results(
 
                 for vector in suite_decoder_res[1].test_vectors.values():
                     jcase = junitp.TestCase(vector.name)
-                    if vector.test_result == TestVectorResult.NOT_RUN:
-                        jcase.result = [junitp.Skipped()]
+                    if vector.test_result in [TestVectorResult.NOT_RUN, TestVectorResult.NOT_SUPPORTED]:
+                        jcase.result = [junitp.Skipped(message=vector.test_result.value)]
                     elif vector.test_result not in [
                         TestVectorResult.SUCCESS,
                         TestVectorResult.REFERENCE,
@@ -579,6 +582,7 @@ def _generate_json_summary(self, ctx: Context, results: Dict[str, List[Tuple[Dec
                     "decoder_name": decoder.name,
                     "total_vectors": len(test_suite.test_vectors),
                     "success_vectors": test_suite.test_vectors_success,
+                    "not_supported_vectors": test_suite.test_vectors_not_supported,
                     "total_time": round(test_suite.time_taken - timeouts, 3),
                     "vectors": {},
                 }
@@ -669,6 +673,17 @@ def _global_stats(
             output += "\n|TOTAL|"
             for test_suite in test_suites:
                 output += f"{test_suite.test_vectors_success}/{len(test_suite.test_vectors)}|"
+            output += "\n|NOT SUPPORTED|"
+            for test_suite in test_suites:
+                output += f"{test_suite.test_vectors_not_supported}/{len(test_suite.test_vectors)}|"
+            output += "\n|FAIL/ERROR|"
+            for test_suite in test_suites:
+                failed = (
+                    len(test_suite.test_vectors)
+                    - test_suite.test_vectors_success
+                    - test_suite.test_vectors_not_supported
+                )
+                output += f"{failed}/{len(test_suite.test_vectors)}|"
             output += "\n|TOTAL TIME|"
             for test_suite in test_suites:
                 # Substract from the total time that took running a test suite on a decoder
@@ -737,14 +752,15 @@ def _generate_global_summary(results: Dict[str, List[Tuple[Decoder, TestSuite]]]
                         all_decoders.append(decoder)
                         decoder_names.add(decoder.name)
 
-            decoder_totals = {dec.name: {"success": 0, "total": 0} for dec in all_decoders}
+            decoder_totals = {dec.name: {"success": 0, "total": 0, "not_supported": 0} for dec in all_decoders}
             decoder_times = {dec.name: 0.0 for dec in all_decoders}
             global_profile_stats: Dict[str, Dict[str, Dict[str, int]]] = {dec.name: {} for dec in all_decoders}
 
             for test_suite_results in results.values():
                 for decoder, test_suite in test_suite_results:
                     totals = decoder_totals[decoder.name]
                     totals["success"] += test_suite.test_vectors_success
+                    totals["not_supported"] += test_suite.test_vectors_not_supported
                     totals["total"] += len(test_suite.test_vectors)
 
                     timeouts = self._calculate_timeout_adjustment(ctx, test_suite)
@@ -762,6 +778,16 @@ def _generate_global_summary(results: Dict[str, List[Tuple[Decoder, TestSuite]]]
             output += "\n|TOTAL|" + "".join(
                 f"{decoder_totals[dec.name]['success']}/{decoder_totals[dec.name]['total']}|" for dec in all_decoders
             )
+            output += "\n|NOT SUPPORTED|" + "".join(
+                f"{decoder_totals[dec.name]['not_supported']}/{decoder_totals[dec.name]['total']}|"
+                for dec in all_decoders
+            )
+            fail_error_parts = []
+            for dec in all_decoders:
+                totals = decoder_totals[dec.name]
+                failed = totals["total"] - totals["success"] - totals["not_supported"]
+                fail_error_parts.append(f"{failed}/{totals['total']}|")
+            output += "\n|FAIL/ERROR|" + "".join(fail_error_parts)
             output += "\n|TOTAL TIME|" + "".join(f"{decoder_times[dec.name]:.3f}s|" for dec in all_decoders)
 
             all_profiles: Set[str] = set()

fluster/test.py

@@ -14,7 +14,7 @@
 from time import perf_counter
 from typing import Any
 
-from fluster.decoder import Decoder
+from fluster.decoder import Decoder, NotSupportedError
 from fluster.test_vector import TestVector, TestVectorResult
 from fluster.utils import compare_wav_files, compare_yuv_files, normalize_path
 
@@ -111,6 +111,12 @@ def _test(self) -> None:
         try:
             result = self._execute_decode()
             self.test_vector_result.test_time = perf_counter() - start
+        except NotSupportedError as ex:
+            self.test_vector_result.test_result = TestVectorResult.NOT_SUPPORTED
+            self.test_vector_result.test_time = perf_counter() - start
+            if self.verbose:
+                print(f"  {self.test_vector.name}: {ex.message}")
+            return
         except TimeoutExpired:
             self.test_vector_result.test_result = TestVectorResult.TIMEOUT
             self.test_vector_result.test_time = perf_counter() - start

fluster/test_suite.py

@@ -157,6 +157,7 @@ def __init__(
         self.filename = filename
         self.resources_dir = resources_dir
         self.test_vectors_success = 0
+        self.test_vectors_not_supported = 0
         self.time_taken = 0.0
 
     def clone(self) -> "TestSuite":
@@ -174,6 +175,10 @@ def from_json_file(cls: Type["TestSuite"], filename: str, resources_dir: str) ->
             data["codec"] = Codec(data["codec"])
             if "test_method" in data:
                 data["test_method"] = TestMethod(data["test_method"])
+            # Remove runtime-only fields if present in malformed JSON
+            data.pop("test_vectors_success", None)
+            data.pop("test_vectors_not_supported", None)
+            data.pop("time_taken", None)
             return cls(filename, resources_dir, **data)
 
     def to_json_file(self, filename: str) -> None:
@@ -183,6 +188,7 @@ def to_json_file(self, filename: str) -> None:
             data.pop("resources_dir")
             data.pop("filename")
             data.pop("test_vectors_success")
+            data.pop("test_vectors_not_supported")
             data.pop("time_taken")
             if self.failing_test_vectors is None:
                 data.pop("failing_test_vectors")
@@ -459,8 +465,11 @@ def _callback(test_result: TestVector) -> None:
         self.time_taken = perf_counter() - start
         print("\n")
         self.test_vectors_success = 0
+        self.test_vectors_not_supported = 0
         for test_vector_res in test_vector_results:
-            if test_vector_res.errors:
+            if test_vector_res.test_result == TestVectorResult.NOT_SUPPORTED:
+                self.test_vectors_not_supported += 1
+            elif test_vector_res.errors:
                 if self.negative_test:
                     self.test_vectors_success += 1
                 else:
@@ -476,10 +485,12 @@ def _callback(test_result: TestVector) -> None:
             # Collect the test vector results and failures since they come
             # from a different process
             self.test_vectors[test_vector_res.name] = test_vector_res
-        print(
-            f"Ran {self.test_vectors_success}/{len(tests)} tests successfully \
-              in {self.time_taken:.3f} secs"
-        )
+
+        status_parts = [f"{self.test_vectors_success}/{len(tests)} tests successfully"]
+        if self.test_vectors_not_supported > 0:
+            status_parts.append(f"{self.test_vectors_not_supported} not supported")
+        status_parts.append(f"in {self.time_taken:.3f} secs")
+        print(f"Ran {', '.join(status_parts)}")
 
     def run(self, ctx: Context) -> Optional["TestSuite"]:
         """

fluster/test_vector.py

@@ -30,6 +30,7 @@ class TestVectorResult(Enum):
     TIMEOUT = "Timeout"
     ERROR = "Error"
     REFERENCE = "Reference run"  # used in reference runs to indicate the decoder for this test vector was succesful
+    NOT_SUPPORTED = "Not Supported"  # used to indicate the decoder cannot handle this media
 
 
 class TestVector: