diff --git a/pyproject.toml b/pyproject.toml index 0c6253ce..83919d4a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "c2pa-python" -version = "0.37.1" +version = "0.37.2" requires-python = ">=3.10" description = "Python bindings for the C2PA Content Authenticity Initiative (CAI) library" readme = { file = "README.md", content-type = "text/markdown" } diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index dec2921f..6e4ca133 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -1570,6 +1570,9 @@ class Stream: # Maximum value for a 32-bit signed integer (2^31 - 1) _MAX_STREAM_ID = 2**31 - 1 + # Methods an object must expose to be wrapped as Stream. + _REQUIRED_STREAM_METHODS = ("read", "write", "seek", "tell", "flush") + # Class-level error messages to avoid multiple creation _ERROR_MESSAGES = { 'stream_error': "Error cleaning up stream: {}", @@ -1616,15 +1619,14 @@ def __init__(self, file_like_stream): self._stream_id = f"{id(self)}-{stream_counter}" # Rest of the existing initialization code... - required_methods = ['read', 'write', 'seek', 'tell', 'flush'] missing_methods = [ - method for method in required_methods if not hasattr( + method for method in Stream._REQUIRED_STREAM_METHODS if not hasattr( file_like_stream, method)] if missing_methods: raise TypeError( "Object must be a stream-like object with methods: {}. " "Missing: {}".format( - ", ".join(required_methods), + ", ".join(Stream._REQUIRED_STREAM_METHODS), ", ".join(missing_methods), ) ) @@ -1976,7 +1978,7 @@ def _encode_format( format_str: Optional[str], class_name: str, allow_autodetect: bool = True, -) -> bytes: +) -> Optional[bytes]: """Normalize a MIME type/format string and encode it to UTF-8 bytes. The binding does not validate the format itself: the native library is the @@ -1998,17 +2000,19 @@ def _encode_format( C2paError.NotSupported. Returns: - UTF-8 encoded format bytes (empty bytes for auto-detection). + The lowercased UTF-8 format bytes, or None to request auto-detection + (no format given). Use _format_ffi_arg to turn the result into the + empty-bytes flag the native library expects. Raises: C2paError.NotSupported: If the format is missing and ``allow_autodetect`` is False. C2paError.Encoding: If the format contains invalid UTF-8 characters. """ - key = (format_str or "").strip() + key = (format_str or "").strip().lower() if not key: if allow_autodetect: - return b"" + return None raise C2paError.NotSupported( f"{class_name} requires an explicit format (MIME type)" ) @@ -2019,6 +2023,26 @@ def _encode_format( f"Invalid UTF-8 characters in input: {e}") +def _format_ffi_arg(fmt: Optional[bytes]) -> bytes: + """Convert an encoded format to the native FFI argument. + + The native library treats empty bytes as the "detect the type from the + bytes" flag, so a missing format (None) maps to b"" rather than a NULL + ``c_char_p``. + """ + return fmt if fmt is not None else b"" + + +def _is_read_stream(obj) -> bool: + """Return True if obj is a stream-like object this SDK can use. + Note: only method presence to identify streams are checked, + not that they work (a broken stream can fail later). + """ + if obj is None or isinstance(obj, (str, Path)): + return False + return all(hasattr(obj, method) for method in Stream._REQUIRED_STREAM_METHODS) + + class Reader(ManagedResource): """High-level wrapper for C2PA Reader operations. @@ -2085,7 +2109,16 @@ def _is_mime_type_supported(cls, mime_type: str) -> bool: @overload def try_create( cls, - format_or_path: Union[str, Path, None], + stream: Any, + manifest_data: Optional[Any] = None, + context: Optional['ContextProvider'] = None, + ) -> Optional["Reader"]: ... + + @classmethod + @overload + def try_create( + cls, + format_or_path: Union[str, Path, None] = None, stream: Optional[Any] = None, manifest_data: Optional[Any] = None, ) -> Optional["Reader"]: ... @@ -2103,7 +2136,7 @@ def try_create( @classmethod def try_create( cls, - format_or_path: Union[str, Path, None], + format_or_path: Union[str, Path, None] = None, stream: Optional[Any] = None, manifest_data: Optional[Any] = None, context: Optional['ContextProvider'] = None, @@ -2117,9 +2150,13 @@ def try_create( want to check if an asset contains C2PA data without handling exceptions for the expected case of no manifest. + Pass a stream as the only argument (``try_create(stream)``) to read it + with an auto-detected format. + Args: - format_or_path: The format or path to read from. - None or an empty string requests auto-detection (best effort). + format_or_path: The format or path to read from, or a stream to + read with an auto-detected format. None or an empty string + requests auto-detection (best effort). stream: Optional stream to read from (Python stream-like object) manifest_data: Optional manifest data in bytes context: Optional ContextProvider for settings @@ -2142,7 +2179,15 @@ def try_create( @overload def __init__( self, - format_or_path: Union[str, Path, None], + stream: Any, + manifest_data: Optional[Any] = None, + context: Optional['ContextProvider'] = None, + ) -> None: ... + + @overload + def __init__( + self, + format_or_path: Union[str, Path, None] = None, stream: Optional[Any] = None, manifest_data: Optional[Any] = None, ) -> None: ... @@ -2158,7 +2203,7 @@ def __init__( def __init__( self, - format_or_path: Union[str, Path, None], + format_or_path: Union[str, Path, None] = None, stream: Optional[Any] = None, manifest_data: Optional[Any] = None, context: Optional['ContextProvider'] = None, @@ -2178,9 +2223,13 @@ def __init__( sidecar. If the bytes are not a recognized asset and no format resolves the type, a C2paError is raised. + Pass a stream as the only argument (``Reader(stream)``) to read it with + an auto-detected format. + Args: - format_or_path: The format (MIME type) or path to read from. - None or an empty string requests detection from the bytes. + format_or_path: The format (MIME type) or path to read from, or a + stream to read with an auto-detected format. None or an empty + string requests detection from the bytes. stream: Optional stream to read from (Python stream-like object) manifest_data: Optional manifest data in bytes context: Optional context implementing ContextProvider with settings @@ -2209,6 +2258,15 @@ def __init__( self._context = context + # Only stream, no format: Reader(fh) must auto-detect on the stream. + if stream is None and _is_read_stream(format_or_path): + stream = format_or_path + format_or_path = None + # A context supplies settings, not the asset, so a path or stream is + # still required to read from. + if format_or_path is None and stream is None: + raise C2paError("Reader requires a path or a stream") + if context is not None: self._init_from_context( context, format_or_path, stream, @@ -2216,46 +2274,53 @@ def __init__( ) return + format_bytes = self._resolve_format_bytes(format_or_path, stream) + if stream is None: # Create a stream from the file path in format_or_path. - # Empty format = native lib will try to guess format. - path = str(format_or_path) - mime_type = _get_mime_type_from_path(path) - - format_bytes = _encode_format(mime_type, "Reader") - self._init_from_file(path, format_bytes) + # A None format lets the native lib guess from the bytes. + self._init_from_file(str(format_or_path), format_bytes) elif isinstance(stream, str): # stream is a file path, format_or_path is the format - format_bytes = _encode_format( - None if format_or_path is None else str(format_or_path), - "Reader") self._init_from_file( stream, format_bytes, manifest_data) else: # format_or_path is a format string, stream is a stream object - format_bytes = _encode_format( - None if format_or_path is None else str(format_or_path), - "Reader") - with Stream(stream) as stream_obj: self._create_reader( format_bytes, stream_obj, manifest_data) self._lifecycle_state = LifecycleState.ACTIVE + @staticmethod + def _resolve_format_bytes(format_or_path, stream) -> Optional[bytes]: + """Resolve the encoded format for a (format_or_path, stream) pair. + + When only a path is given (``stream is None``), the format is derived + from the file extension. Otherwise ``format_or_path`` is the format + string (or None). + Returns None to request auto-detection. + """ + if stream is None: + return _encode_format( + _get_mime_type_from_path(str(format_or_path)), "Reader") + return _encode_format( + None if format_or_path is None else str(format_or_path), "Reader") + def _create_reader(self, format_bytes, stream_obj, manifest_data=None): """Create a Reader from a Stream. Args: - format_bytes: UTF-8 encoded format/MIME type + format_bytes: Encoded format/MIME type, or None for auto-detection stream_obj: A Stream instance manifest_data: Optional manifest bytes """ + format_arg = _format_ffi_arg(format_bytes) if manifest_data is None: self._handle = _lib.c2pa_reader_from_stream( - format_bytes, stream_obj._stream) + format_arg, stream_obj._stream) else: if not isinstance(manifest_data, bytes): raise TypeError(Reader._ERROR_MESSAGES['manifest_error']) @@ -2264,7 +2329,7 @@ def _create_reader(self, format_bytes, stream_obj, len(manifest_data)).from_buffer_copy(manifest_data) self._handle = ( _lib.c2pa_reader_from_manifest_data_and_stream( - format_bytes, + format_arg, stream_obj._stream, manifest_array, len(manifest_data), @@ -2309,23 +2374,16 @@ def _init_from_context(self, context, format_or_path, raise TypeError(Reader._ERROR_MESSAGES['manifest_error']) # Determine format and open stream + format_bytes = self._resolve_format_bytes(format_or_path, stream) + format_arg = _format_ffi_arg(format_bytes) if stream is None: - # Empty format = native lib will try to guess format. - path = str(format_or_path) - mime_type = _get_mime_type_from_path(path) - format_bytes = _encode_format(mime_type, "Reader") - self._backing_file = open(path, 'rb') + # A None format lets the native lib guess from the bytes. + self._backing_file = open(str(format_or_path), 'rb') self._own_stream = Stream(self._backing_file) elif isinstance(stream, str): - format_bytes = _encode_format( - None if format_or_path is None else str(format_or_path), - "Reader") self._backing_file = open(stream, 'rb') self._own_stream = Stream(self._backing_file) else: - format_bytes = _encode_format( - None if format_or_path is None else str(format_or_path), - "Reader") self._own_stream = Stream(stream) try: @@ -2354,7 +2412,7 @@ def _init_from_context(self, context, format_or_path, new_ptr = ( _lib.c2pa_reader_with_manifest_data_and_stream( reader_ptr, - format_bytes, + format_arg, self._own_stream._stream, manifest_array, len(manifest_data), @@ -2363,7 +2421,7 @@ def _init_from_context(self, context, format_or_path, else: # Consume reader with stream new_ptr = _lib.c2pa_reader_with_stream( - reader_ptr, format_bytes, + reader_ptr, format_arg, self._own_stream._stream, ) @@ -2453,12 +2511,12 @@ def with_fragment(self, format: Optional[str], stream, """ self._ensure_valid_state() - format_bytes = _encode_format(format, "Reader") + format_arg = _format_ffi_arg(_encode_format(format, "Reader")) with Stream(stream) as main_obj, Stream(fragment_stream) as frag_obj: new_ptr = _lib.c2pa_reader_with_fragment( self._handle, - format_bytes, + format_arg, main_obj._stream, frag_obj._stream, ) @@ -3472,15 +3530,16 @@ def _sign_internal( if not hasattr(signer, '_handle') or not signer._handle: raise C2paError("Invalid or closed signer") - format_bytes = _encode_format( - format, "Builder", allow_autodetect=False) + # allow_autodetect=False, so this never returns None (raises instead). + format_arg = _format_ffi_arg( + _encode_format(format, "Builder", allow_autodetect=False)) manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() try: if signer is not None: result = _lib.c2pa_builder_sign( self._handle, - format_bytes, + format_arg, source_stream._stream, dest_stream._stream, signer._handle, @@ -3489,7 +3548,7 @@ def _sign_internal( else: result = _lib.c2pa_builder_sign_context( self._handle, - format_bytes, + format_arg, source_stream._stream, dest_stream._stream, ctypes.byref(manifest_bytes_ptr), diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 4f226c38..733952b6 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -13,6 +13,9 @@ import os import io +import gzip +import bz2 +import lzma import json import unittest import ctypes @@ -30,7 +33,7 @@ from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version, C2paBuilderIntent, C2paDigitalSourceType from c2pa import Settings, Context, ContextBuilder, ContextProvider -from c2pa.c2pa import Stream, LifecycleState, load_settings, create_signer, create_signer_from_info, ed25519_sign, format_embeddable, _get_mime_type_from_path, _encode_format +from c2pa.c2pa import Stream, LifecycleState, load_settings, create_signer, create_signer_from_info, ed25519_sign, format_embeddable, _get_mime_type_from_path, _encode_format, _format_ffi_arg, _is_read_stream from pathlib import Path @@ -88,6 +91,96 @@ def test_sdk_version(self): self.assertIn(parse_native_version(), sdk_version()) +class TestIsReadableStream(unittest.TestCase): + """ A stream is anything Stream will accept (read/write/seek/tell/flush). + """ + + def _tmp_path(self): + return os.path.join(FIXTURES_DIR, DEFAULT_TEST_FILE_NAME) + + def test_paths_and_formats_are_not_streams(self): + for value in (None, "", "image/jpeg", "path/to/file.jpg", + Path("x.jpg")): + self.assertFalse(_is_read_stream(value), repr(value)) + + def test_non_stream_objects_are_not_streams(self): + for value in (b"bytes", bytearray(b"x"), 42, object()): + self.assertFalse(_is_read_stream(value), repr(value)) + + def test_object_missing_methods_is_not_a_stream(self): + class OnlyRead: + def read(self): + return b"" + self.assertFalse(_is_read_stream(OnlyRead())) + + def test_read_only_object_is_not_a_stream(self): + # Has read/seek/tell but no write/flush, so not a Stream for us. + class ReadOnly: + def read(self): + return b"" + + def seek(self, *a): + return 0 + + def tell(self): + return 0 + self.assertFalse(_is_read_stream(ReadOnly())) + + def test_in_memory_streams_are_streams(self): + self.assertTrue(_is_read_stream(io.BytesIO(b"x"))) + self.assertTrue(_is_read_stream(io.StringIO("x"))) + + def test_file_handles_are_streams(self): + path = self._tmp_path() + for opener in ( + lambda: open(path, "rb"), + lambda: open(path, "rb", buffering=0), + lambda: io.BufferedReader(io.FileIO(path, "rb")), + lambda: open(path, "r+b"), + lambda: open(path, "r"), + ): + fh = opener() + try: + self.assertTrue(_is_read_stream(fh), type(fh).__name__) + finally: + fh.close() + + def test_spooled_temporary_file_is_a_stream(self): + with tempfile.SpooledTemporaryFile() as fh: + self.assertTrue(_is_read_stream(fh)) + + def test_compressed_file_objects_are_streams(self): + with tempfile.TemporaryDirectory() as tmp: + for ext, opener in ( + ("gz", gzip.open), + ("bz2", bz2.open), + ("xz", lzma.open), + ): + p = os.path.join(tmp, f"f.{ext}") + with opener(p, "wb") as fh: + fh.write(b"data") + with opener(p, "rb") as fh: + self.assertTrue(_is_read_stream(fh), ext) + + def test_placeholder_typed_object_with_all_methods_is_a_stream(self): + class Duck: + def read(self, *a): + return b"" + + def write(self, *a): + return 0 + + def seek(self, *a): + return 0 + + def tell(self): + return 0 + + def flush(self): + pass + self.assertTrue(_is_read_stream(Duck())) + + class TestFormatValidation(unittest.TestCase): """Unit tests for _encode_format. @@ -102,16 +195,18 @@ def test_strips_whitespace_from_query(self): _encode_format(value, "Reader"), b"image/jpeg") - def test_case_is_preserved(self): + def test_case_normalized_to_lowercase(self): + # MIME types are case-insensitive, so the format is lowercased before + # being handed to the native library. for value in ("IMAGE/JPEG", "Image/Jpeg"): self.assertEqual( _encode_format(value, "Reader"), - value.encode("utf-8")) + b"image/jpeg") - def test_encodes_stripped_key(self): + def test_encodes_stripped_and_lowercased_key(self): self.assertEqual( _encode_format(" IMAGE/JPEG ", "Reader"), - b"IMAGE/JPEG") + b"image/jpeg") def test_production_supported_list_is_normalized(self): # _get_supported_mime_types strips and lowercases every entry. @@ -119,17 +214,37 @@ def test_production_supported_list_is_normalized(self): self.assertEqual(entry, entry.strip().lower()) def test_none_autodetects(self): - self.assertEqual(_encode_format(None, "Reader"), b"") + # A missing format is None (auto-detect), not the empty-bytes sentinel. + self.assertIsNone(_encode_format(None, "Reader")) def test_blank_autodetects(self): for value in ("", " ", "\t\n"): - self.assertEqual(_encode_format(value, "Reader"), b"") + self.assertIsNone(_encode_format(value, "Reader")) def test_missing_format_rejected_when_autodetect_disabled(self): for value in (None, "", " "): with self.assertRaises(Error.NotSupported): _encode_format(value, "Builder", allow_autodetect=False) + def test_format_ffi_arg_maps_none_to_empty_bytes(self): + # The native library's "detect from bytes" flag is empty bytes, not a + # NULL c_char_p, so None must map to b"". + self.assertEqual(_format_ffi_arg(None), b"") + self.assertEqual(_format_ffi_arg(b"image/jpeg"), b"image/jpeg") + + def test_resolve_format_bytes(self): + # Path form derives the format from the extension. + self.assertEqual( + Reader._resolve_format_bytes("photo.jpg", None), b"image/jpeg") + # Extensionless path -> auto-detect. + self.assertIsNone( + Reader._resolve_format_bytes("no_extension", None)) + # Stream given: format_or_path is the format (or None for detect). + self.assertEqual( + Reader._resolve_format_bytes("image/jpeg", object()), b"image/jpeg") + self.assertIsNone( + Reader._resolve_format_bytes(None, object())) + def test_unknown_format_passes_through(self): # No supported-list check: the format is encoded and handed to the # native library, which is the authority on validity. @@ -138,10 +253,9 @@ def test_unknown_format_passes_through(self): b"application/x-nope") def test_literal_none_string_passes_through(self): - # The string "None" is a plain string; - # it is trimmed and encoded like any other, - # not rejected in the binding. - self.assertEqual(_encode_format("None", "Reader"), b"None") + # The string "None" is a plain string; it is trimmed, lowercased, and + # encoded like any other, not rejected in the binding. + self.assertEqual(_encode_format("None", "Reader"), b"none") class TestReader(unittest.TestCase): @@ -386,6 +500,7 @@ def test_none_format_matches_empty_string_on_stream(self): empty_json = reader.json() self.assertEqual(none_json, empty_json) self.assertNotIn("None", none_json[:64]) + self.assertIn(DEFAULT_TEST_FILE_NAME, none_json) def test_padded_format_accepted_on_stream(self): with open(self.testPath, "rb") as file: @@ -402,6 +517,121 @@ def test_try_create_none_format_matches_empty_string(self): if none_reader is not None and empty_reader is not None: self.assertEqual(none_reader.json(), empty_reader.json()) + def test_bare_stream_matches_empty_and_none_format(self): + with open(self.testPath, "rb") as file: + with Reader(file) as reader: + bare_json = reader.json() + with open(self.testPath, "rb") as file: + with Reader("", file) as reader: + empty_json = reader.json() + with open(self.testPath, "rb") as file: + with Reader(None, file) as reader: + none_json = reader.json() + self.assertEqual(bare_json, empty_json) + self.assertEqual(bare_json, none_json) + + def test_bare_stream_keyword_form(self): + with open(self.testPath, "rb") as file: + with Reader(stream=file) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_bare_stream_in_memory(self): + with open(self.testPath, "rb") as file: + data = file.read() + with Reader(io.BytesIO(data)) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_try_create_bare_stream(self): + with open(self.testPath, "rb") as file: + reader = Reader.try_create(file) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + + def test_try_create_bare_stream_keyword(self): + with open(self.testPath, "rb") as file: + reader = Reader.try_create(stream=file) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + + def test_bare_stream_closed_handle_raises(self): + # _is_read_stream only checks method presence + # so a closed handle is accepted as a stream and fails at read time. + file = open(self.testPath, "rb") + file.close() + with self.assertRaises(Error): + Reader(file) + + def test_bare_stream_with_manifest_data_keeps_manifest(self): + # The bare-stream shift must not swallow manifest_data. + with open(self.testPath, "rb") as file: + with Reader(file, manifest_data=None) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_path_string_still_opens_by_path(self): + with Reader(self.testPath) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_path_object_still_opens_by_path(self): + with Reader(Path(self.testPath)) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_format_plus_stream_still_means_format_and_stream(self): + with open(self.testPath, "rb") as file: + with Reader("image/jpeg", file) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + with open(self.testPath, "rb") as file: + with Reader(" image/jpeg ", file) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_format_plus_path_string_still_opens_by_path(self): + with Reader("image/jpeg", self.testPath) as reader: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + + def test_no_args_raises_clear_error(self): + with self.assertRaises(Error) as ctx: + Reader() + self.assertIn("stream", str(ctx.exception).lower()) + + def test_none_first_arg_no_stream_raises_clear_error(self): + with self.assertRaises(Error) as ctx: + Reader(None) + self.assertIn("stream", str(ctx.exception).lower()) + + def test_text_stream_matches_existing_explicit_form(self): + with open(self.testPath, "rb") as file: + data = file.read() + text = data.decode("latin-1") + + def outcome(call): + try: + r = call() + out = ("ok", r.json()[:0]) + r.close() + return out + except Exception as e: + return ("err", type(e).__name__) + bare = outcome(lambda: Reader(io.StringIO(text))) + explicit = outcome(lambda: Reader("image/jpeg", io.StringIO(text))) + self.assertEqual(bare, explicit) + + def test_read_only_object_not_treated_as_stream(self): + # read/seek/tell but no write/flush + class ReadOnly: + def read(self, *a): + return b"" + + def seek(self, *a): + return 0 + + def tell(self): + return 0 + with self.assertRaises(Error): + Reader(ReadOnly()) + def test_unrecognized_extension_defers_to_detection(self): with tempfile.TemporaryDirectory() as tmp: asset = os.path.join(tmp, "C.test") @@ -591,17 +821,17 @@ def test_get_mime_type_from_path_accepts_path_object(self): _get_mime_type_from_path(Path("dir/photo.jpg")), "image/jpeg") def test_encode_format_blank_autodetects(self): - # Reader default: a blank format requests auto-detection. - self.assertEqual(_encode_format("", "Reader"), b"") - self.assertEqual(_encode_format(" ", "Reader"), b"") + # A blank format requests auto-detection. + self.assertIsNone(_encode_format("", "Reader")) + self.assertIsNone(_encode_format(" ", "Reader")) def test_encode_format_returns_bytes(self): self.assertEqual( _encode_format("image/jpeg", "Reader"), b"image/jpeg") - def test_encode_format_case_preserved(self): + def test_encode_format_case_normalized(self): self.assertEqual( - _encode_format("IMAGE/JPEG", "Reader"), b"IMAGE/JPEG") + _encode_format("IMAGE/JPEG", "Reader"), b"image/jpeg") def test_encode_format_unknown_passes_through(self): # The native library decides validity. @@ -2762,6 +2992,75 @@ def test_remote_sign_using_returned_bytes(self): manifest_data = reader.json() self.assertIn("Python Test", manifest_data) + def _detached_manifest_and_asset(self): + with open(self.testPath, "rb") as file: + builder = Builder(self.manifestDefinition) + builder.set_no_embed() + with io.BytesIO() as output_buffer: + manifest = builder.sign( + self.signer, "image/jpeg", file, output_buffer) + return manifest, output_buffer.getvalue() + + def test_bare_stream_with_real_manifest_data_is_used(self): + manifest, asset = self._detached_manifest_and_asset() + with Reader(io.BytesIO(asset), manifest_data=manifest) as reader: + self.assertIn("Python Test", reader.json()) + + def test_manifest_data_keyword_with_real_bytes(self): + # manifest_data passed by keyword (not positionally) with real bytes. + manifest, asset = self._detached_manifest_and_asset() + with Reader( + "image/jpeg", io.BytesIO(asset), + manifest_data=manifest) as reader: + self.assertIn("Python Test", reader.json()) + + def test_try_create_with_manifest_data(self): + manifest, asset = self._detached_manifest_and_asset() + reader = Reader.try_create( + "image/jpeg", io.BytesIO(asset), manifest) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn("Python Test", reader.json()) + reader.close() + + def test_try_create_with_manifest_data_keyword(self): + # try_create with manifest_data passed by keyword. + manifest, asset = self._detached_manifest_and_asset() + reader = Reader.try_create( + "image/jpeg", io.BytesIO(asset), manifest_data=manifest) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn("Python Test", reader.json()) + reader.close() + + def test_bare_stream_with_context_and_manifest(self): + # Stream and real manifest and context together. + manifest, asset = self._detached_manifest_and_asset() + context = Context() + try: + with Reader(io.BytesIO(asset), manifest_data=manifest, + context=context) as reader: + self.assertIn("Python Test", reader.json()) + finally: + context.close() + + def test_try_create_format_plus_path_string(self): + with tempfile.TemporaryDirectory() as tmp: + asset = os.path.join(tmp, "signed.jpg") + with open(self.testPath, "rb") as file, open(asset, "wb") as out: + Builder(self.manifestDefinition).sign( + self.signer, "image/jpeg", file, out) + reader = Reader.try_create("image/jpeg", asset) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn("Python Test", reader.json()) + reader.close() + + def test_non_context_wrong_type_manifest_raises_typeerror(self): + with open(self.testPath, "rb") as file: + with self.assertRaises(TypeError): + Reader("image/jpeg", file, "not-bytes") + def test_remote_sign_using_returned_bytes_V2(self): with open(self.testPath, "rb") as file: builder = Builder(self.manifestDefinitionV2) @@ -3061,6 +3360,41 @@ def test_builder_add_multiple_ingredients_and_resources(self): builder.close() + def test_sign_with_any_format_spelling_keeps_thumbnail(self): + # The format reaches thumbnail generation, + # which matches a MIME type case sensitively, + # so an unnormalized spelling silently produces no thumbnail. + for fmt in ("image/jpeg", "IMAGE/JPEG", "Image/Jpeg", "jpg", "JPG"): + with self.subTest(format=fmt): + builder = Builder(self.manifestDefinition) + dest = io.BytesIO() + with open(self.testPath, "rb") as source: + builder.sign(self.signer, fmt, source, dest) + + dest.seek(0) + reader = Reader("image/jpeg", dest) + manifest_store = json.loads(reader.json()) + active = manifest_store["manifests"][ + manifest_store["active_manifest"]] + + self.assertIn( + "thumbnail", active, + f"signing with {fmt} dropped the claim thumbnail") + thumbnail = active["thumbnail"] + self.assertEqual(thumbnail["format"], "image/jpeg") + + # The reference must resolve to a real JPEG, not just be present. + thumbnail_bytes = io.BytesIO() + reader.resource_to_stream( + thumbnail["identifier"], thumbnail_bytes) + data = thumbnail_bytes.getvalue() + self.assertGreater( + len(data), 1000, + f"signing with {fmt} produced an empty claim thumbnail") + self.assertEqual( + data[:3], b"\xff\xd8\xff", + f"signing with {fmt} produced a thumbnail that is not a JPEG") + def test_builder_add_multiple_ingredients_and_resources_interleaved(self): builder = Builder.from_json(self.manifestDefinition) assert builder._handle is not None @@ -6349,6 +6683,59 @@ def test_reader_format_and_path_with_ctx(self): reader.close() context.close() + def test_reader_bare_stream_with_context(self): + # The bare-stream shift runs before the context branch, so + # Reader(fh, context=ctx) reads via context with auto-detect. + context = Context() + with open(DEFAULT_TEST_FILE, "rb") as file_handle: + reader = Reader(file_handle, context=context) + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + context.close() + + def test_reader_path_object_with_context(self): + context = Context() + reader = Reader(Path(DEFAULT_TEST_FILE), context=context) + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + context.close() + + def test_try_create_format_stream_with_context(self): + context = Context() + with open(DEFAULT_TEST_FILE, "rb") as file_handle: + reader = Reader.try_create( + "image/jpeg", file_handle, context=context) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + context.close() + + def test_try_create_bare_stream_with_context(self): + context = Context() + with open(DEFAULT_TEST_FILE, "rb") as file_handle: + reader = Reader.try_create(file_handle, context=context) + self.assertIsNotNone(reader) + if reader is not None: + self.assertIn(DEFAULT_TEST_FILE_NAME, reader.json()) + reader.close() + context.close() + + def test_context_only_no_source_raises(self): + # A context supplies settings, not the asset, + # so no path/stream is an error rather than open("None"). + context = Context() + with self.assertRaises(Error) as ctx: + Reader(context=context) + self.assertIn("stream", str(ctx.exception).lower()) + context.close() + + def test_reader_with_closed_context_raises(self): + context = Context() + context.close() + with self.assertRaises(Error): + Reader(DEFAULT_TEST_FILE, context=context) + def test_with_fragment_on_closed_reader_raises(self): context = Context() reader = Reader(DEFAULT_TEST_FILE, context=context)