feat: Fragment APIs (e.g. for video)

src/c2pa/c2pa.py

@@ -76,6 +76,7 @@
     'c2pa_context_new',
     'c2pa_reader_from_context',
     'c2pa_reader_with_stream',
+    'c2pa_reader_with_fragment',
     'c2pa_builder_from_context',
     'c2pa_builder_with_definition',
     'c2pa_builder_with_archive',
@@ -694,6 +695,12 @@ def _setup_function(func, argtypes, restype=None):
      ctypes.POINTER(C2paStream)],
     ctypes.POINTER(C2paReader)
 )
+_setup_function(
+    _lib.c2pa_reader_with_fragment,
+    [ctypes.POINTER(C2paReader), ctypes.c_char_p,
+     ctypes.POINTER(C2paStream), ctypes.POINTER(C2paStream)],
+    ctypes.POINTER(C2paReader)
+)
 _setup_function(
     _lib.c2pa_builder_from_context,
     [ctypes.POINTER(C2paContext)],
@@ -2092,7 +2099,8 @@ class Reader(ManagedResource):
         'file_error': "Error cleaning up file: {}",
         'reader_cleanup_error': "Error cleaning up reader: {}",
         'encoding_error': "Invalid UTF-8 characters in input: {}",
-        'closed_error': "Reader is closed"
+        'closed_error': "Reader is closed",
+        'fragment_error': "Failed to process fragment: {}"
     }
 
     @classmethod
@@ -2466,6 +2474,58 @@ def _get_cached_manifest_data(self) -> Optional[dict]:
 
         return self._manifest_data_cache
 
+    def with_fragment(self, format: str, stream,
+                      fragment_stream) -> "Reader":
+        """Process a BMFF fragment stream with this reader.
+
+        Used for fragmented BMFF media (DASH/HLS streaming) where
+        content is split into init segments and fragment files.
+
+        Args:
+            format: MIME type of the media (e.g., "video/mp4")
+            stream: Stream-like object with the main/init segment data
+            fragment_stream: Stream-like object with the fragment data
+
+        Returns:
+            This reader instance, for method chaining.
+
+        Raises:
+            C2paError: If there was an error processing the fragment
+        """
+        self._ensure_valid_state()
+
+        supported = Reader.get_supported_mime_types()
+        format_bytes = _validate_and_encode_format(
+            format, supported, "Reader"
+        )
+
+        with Stream(stream) as main_obj, Stream(fragment_stream) as frag_obj:
+            new_ptr = _lib.c2pa_reader_with_fragment(
+                self._handle,
+                format_bytes,
+                main_obj._stream,
+                frag_obj._stream,
+            )
+
+            if not new_ptr:
+                self._handle = None
+                error = _parse_operation_result_for_error(
+                    _lib.c2pa_error()
+                )
+                if error:
+                    raise C2paError(error)
+                raise C2paError(
+                    Reader._ERROR_MESSAGES[
+                        'fragment_error'
+                    ].format("Unknown error"))
+            self._handle = new_ptr
+
+        # Invalidate caches: fragment may change manifest data
+        self._manifest_json_str_cache = None
+        self._manifest_data_cache = None
+
+        return self
+
     def close(self):
         """Release the reader resources."""
         self._manifest_json_str_cache = None

tests/test_unit_tests.py

@@ -256,7 +256,7 @@ def test_stream_read_detailed_and_parse(self):
             title = manifest_store["manifests"][manifest_store["active_manifest"]]["claim"]["dc:title"]
             self.assertEqual(title, DEFAULT_TEST_FILE_NAME)
 
-    def test_stream_read_string_stream(self):
+    def test_stream_read_string_stream_path_only(self):
         with Reader(self.testPath) as reader:
             json_data = reader.json()
             self.assertIn(DEFAULT_TEST_FILE_NAME, json_data)
@@ -4666,7 +4666,7 @@ def test_builder_add_ingredient_from_file_path(self):
 
             builder.close()
 
-    def test_builder_add_ingredient_from_file_path(self):
+    def test_builder_add_ingredient_from_file_path_not_found(self):
         """Test Builder class add_ingredient_from_file_path method."""
 
         # Suppress the specific deprecation warning for this test, as this is a legacy method
@@ -4925,56 +4925,6 @@ def test_sign_file_callback_signer(self):
         finally:
             shutil.rmtree(temp_dir)
 
-    def test_sign_file_callback_signer(self):
-        """Test signing a file using the sign_file method."""
-
-        temp_dir = tempfile.mkdtemp()
-
-        try:
-            output_path = os.path.join(temp_dir, "signed_output.jpg")
-
-            # Use the sign_file method
-            builder = Builder(self.manifestDefinition)
-
-            # Create signer with callback using create_signer function
-            signer = create_signer(
-                callback=self.callback_signer_es256,
-                alg=SigningAlg.ES256,
-                certs=self.certs.decode('utf-8'),
-                tsa_url="http://timestamp.digicert.com"
-            )
-
-            manifest_bytes = builder.sign_file(
-                source_path=self.testPath,
-                dest_path=output_path,
-                signer=signer
-            )
-
-            # Verify the output file was created
-            self.assertTrue(os.path.exists(output_path))
-
-            # Verify results
-            self.assertIsInstance(manifest_bytes, bytes)
-            self.assertGreater(len(manifest_bytes), 0)
-
-            # Read the signed file and verify the manifest
-            with open(output_path, "rb") as file, Reader("image/jpeg", file) as reader:
-                json_data = reader.json()
-                # Needs trust configuration to be set up to validate as Trusted
-                # self.assertNotIn("validation_status", json_data)
-
-                # Parse the JSON and verify the signature algorithm
-                manifest_data = json.loads(json_data)
-                active_manifest_id = manifest_data["active_manifest"]
-                active_manifest = manifest_data["manifests"][active_manifest_id]
-
-                self.assertIn("signature_info", active_manifest)
-                signature_info = active_manifest["signature_info"]
-                self.assertEqual(signature_info["alg"], self.callback_signer_alg)
-
-        finally:
-            shutil.rmtree(temp_dir)
-
     def test_sign_file_callback_signer_managed_single(self):
         """Test signing a file using the sign_file method with context managers."""
 
@@ -5490,6 +5440,42 @@ def test_reader_format_and_path_with_ctx(self):
         reader.close()
         context.close()
 
+    def test_with_fragment_on_closed_reader_raises(self):
+        context = Context()
+        reader = Reader(DEFAULT_TEST_FILE, context=context)
+        reader.close()
+        with self.assertRaises(Error):
+            reader.with_fragment(
+                "video/mp4",
+                io.BytesIO(b"\x00" * 100),
+                io.BytesIO(b"\x00" * 100),
+            )
+        context.close()
+
+    def test_with_fragment_unsupported_format_raises(self):
+        context = Context()
+        reader = Reader(DEFAULT_TEST_FILE, context=context)
+        with self.assertRaises(Error):
+            reader.with_fragment(
+                "text/plain",
+                io.BytesIO(b"\x00" * 100),
+                io.BytesIO(b"\x00" * 100),
+            )
+        reader.close()
+        context.close()
+
+    def test_with_fragment_with_dash_fixtures(self):
+        context = Context()
+        init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4")
+        with open(init_path, "rb") as init_fragment:
+            reader = Reader("video/mp4", init_fragment, context=context)
+        frag_path = os.path.join(FIXTURES_DIR, "dash1.m4s")
+        with open(init_path, "rb") as init_fragment, \
+             open(frag_path, "rb") as next_fragment:
+            reader.with_fragment("video/mp4", init_fragment, next_fragment)
+        reader.close()
+        context.close()
+
 
 class TestBuilderWithContext(TestContextAPIs):