feat: add not supported status code

In case of 69 exit code from an app, the test should be reported
as not supported and be considered as non-error status.

Author: dabrain34

fluster/fluster.py

@@ -106,6 +106,7 @@ def to_test_suite_context(
     TestVectorResult.FAIL: "❌",
     TestVectorResult.TIMEOUT: "⌛",
     TestVectorResult.ERROR: "☠",
+    TestVectorResult.NOT_SUPPORTED: "○",
 }
 
 TEXT_RESULT = {
@@ -114,6 +115,7 @@ def to_test_suite_context(
     TestVectorResult.FAIL: "KO",
     TestVectorResult.TIMEOUT: "TO",
     TestVectorResult.ERROR: "ER",
+    TestVectorResult.NOT_SUPPORTED: "NS",
 }
 
 
@@ -376,8 +378,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,
@@ -417,6 +419,7 @@ def _generate_csv_summary(ctx: Context, results: Dict[str, List[Tuple[Decoder, T
             TestVectorResult.ERROR: "Error",
             TestVectorResult.FAIL: "Fail",
             TestVectorResult.NOT_RUN: "Not run",
+            TestVectorResult.NOT_SUPPORTED: "Not supported",
         }
         content: Dict[Any, Any] = defaultdict(lambda: defaultdict(dict))
         max_vectors = 0
@@ -463,6 +466,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
@@ -541,14 +555,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 = (
@@ -578,6 +593,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

@@ -16,7 +16,7 @@
 
 from fluster.decoder import Decoder
 from fluster.test_vector import TestVector, TestVectorResult
-from fluster.utils import compare_byte_wise_files, normalize_path
+from fluster.utils import NotSupportedError, compare_byte_wise_files, normalize_path
 
 
 class Test(unittest.TestCase):
@@ -110,6 +110,13 @@ def _test(self) -> None:
         try:
             result = self._execute_decode()
             self.test_vector_result.test_time = perf_counter() - start
+        except NotSupportedError as ex:
+            # Exit code 69: media not supported by decoder
+            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 that might be present in old JSON files
+            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")
@@ -469,8 +475,12 @@ 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:
+            # Check for NOT_SUPPORTED first
+            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:
@@ -486,10 +496,13 @@ 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"
-        )
+
+        # Build status message
+        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:

fluster/utils.py

@@ -34,6 +34,20 @@
 
 TARBALL_EXTS = ("tar.gz", "tgz", "tar.bz2", "tbz2", "tar.xz")
 
+
+class NotSupportedError(Exception):
+    """Exception raised when decoder cannot handle the media (exit code 69).
+
+    Exit code 69 (EX_UNAVAILABLE) is from BSD sysexits.h, meaning "service unavailable".
+    This is widely recognized across Unix-like systems (Linux, macOS, BSD) and indicates
+    that the decoder service is unavailable for the specific media format/profile.
+    """
+
+    def __init__(self, message: str = "Media not supported by decoder"):
+        self.message = message
+        super().__init__(self.message)
+
+
 download_lock = Lock()
 
 
@@ -199,7 +213,14 @@ def run_command(
         print(f'\nRunning command "{" ".join(command)}"')
     try:
         subprocess.run(command, stdout=sout, stderr=serr, check=check, timeout=timeout)
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as ex:
+    except subprocess.CalledProcessError as ex:
+        # Developer experience improvement (facilitates copy/paste)
+        ex.cmd = " ".join(ex.cmd)
+        # Check for exit code 69 (EX_UNAVAILABLE from BSD sysexits.h)
+        if ex.returncode == 69:
+            raise NotSupportedError(f"Command returned exit code 69 (not supported): {ex.cmd}") from ex
+        raise ex
+    except subprocess.TimeoutExpired as ex:
         # Developer experience improvement (facilitates copy/paste)
         ex.cmd = " ".join(ex.cmd)
         raise ex
@@ -224,17 +245,24 @@ def run_command_with_output(
         if verbose and output:
             print(output)
         return output or ""
-    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as ex:
+    except subprocess.CalledProcessError as ex:
         if verbose and ex.output:
-            # Workaround inconsistent Python implementation
-            if isinstance(ex, subprocess.TimeoutExpired):
-                print(ex.output.decode("utf-8"))
-            else:
-                print(ex.output)
+            print(ex.output)
 
-        if isinstance(ex, subprocess.CalledProcessError) and not check:
+        if not check:
             return ex.output or ""
 
+        # Developer experience improvement (facilitates copy/paste)
+        ex.cmd = " ".join(ex.cmd)
+        # Check for exit code 69 (EX_UNAVAILABLE from BSD sysexits.h)
+        if ex.returncode == 69:
+            raise NotSupportedError(f"Command returned exit code 69 (not supported): {ex.cmd}") from ex
+        raise ex
+    except subprocess.TimeoutExpired as ex:
+        if verbose and ex.output:
+            # Workaround inconsistent Python implementation
+            print(ex.output.decode("utf-8"))
+
         # Developer experience improvement (facilitates copy/paste)
         ex.cmd = " ".join(ex.cmd)
         raise ex