Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 41 additions & 41 deletions src/c2pa/c2pa.py
Original file line number Diff line number Diff line change
Expand Up @@ -1464,36 +1464,6 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None:
check=lambda r: r != 0)


def _get_mime_type_from_path(path: Union[str, Path]) -> str:
"""Attempt to guess the MIME type from a file path's extension.
When the extension is missing or unrecognized, this returns an empty
string so the caller hands it to the lib for auto-detection.
A recognized-but-wrong extension still returns a mimetype:
the native layer will attempt to correct it from the bytes when
reading (the real type still needs to be supported by the lib),
and if it can't, an error will happen then.

Args:
path: File path as string or Path object

Returns:
MIME type string, or an empty string
(when it cannot be determined from the extension).
An empty string here means the native lib should attempt auto-detect.
"""
path_obj = Path(path)
file_extension = path_obj.suffix.lower() if path_obj.suffix else ""

if file_extension == ".dng":
# mimetypes guesses the wrong type for dng,
# so we bypass it and set the correct type
return "image/dng"
else:
# Fall back to an empty string for extensionless or unknown files.
# Empty string flags this as a guess-type attempt.
return mimetypes.guess_type(str(path))[0] or ""


class ContextProvider(ABC):
"""Abstract base class for types that provide a C2PA context.

Expand Down Expand Up @@ -2154,6 +2124,46 @@ def initialized(self) -> bool:
"""
return self._initialized

@staticmethod
def is_read_stream(obj) -> bool:
Comment thread
tmathern marked this conversation as resolved.
"""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)


def _get_mime_type_from_path(path: Union[str, Path]) -> str:
Comment thread
tmathern marked this conversation as resolved.
"""Attempt to guess the MIME type from a file path's extension.
When the extension is missing or unrecognized, this returns an empty
string so the caller hands it to the lib for auto-detection.
A recognized-but-wrong extension still returns a mimetype:
the native layer will attempt to correct it from the bytes when
reading (the real type still needs to be supported by the lib),
and if it can't, an error will happen then.

Args:
path: File path as string or Path object

Returns:
MIME type string, or an empty string
(when it cannot be determined from the extension).
An empty string here means the native lib should attempt auto-detect.
"""
path_obj = Path(path)
file_extension = path_obj.suffix.lower() if path_obj.suffix else ""

if file_extension == ".dng":
# mimetypes guesses the wrong type for dng,
# so we bypass it and set the correct type
return "image/dng"
else:
# Fall back to an empty string for extensionless or unknown files.
# Empty string flags this as a guess-type attempt.
return mimetypes.guess_type(str(path))[0] or ""


def _get_supported_mime_types(ffi_func, cache):
"""Shared helper to retrieve supported MIME types from the native library.
Expand Down Expand Up @@ -2269,16 +2279,6 @@ def _format_ffi_arg(fmt: Optional[bytes]) -> bytes:
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.

Expand Down Expand Up @@ -2478,7 +2478,7 @@ 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):
if stream is None and Stream.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
Expand Down
24 changes: 12 additions & 12 deletions tests/test_unit_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,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, ManagedResource, 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 c2pa.c2pa import Stream, LifecycleState, ManagedResource, load_settings, create_signer, create_signer_from_info, ed25519_sign, format_embeddable, _get_mime_type_from_path, _encode_format, _format_ffi_arg
import c2pa.c2pa as c2pa_module
from pathlib import Path

Expand Down Expand Up @@ -105,17 +105,17 @@ def _tmp_path(self):
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))
self.assertFalse(Stream.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))
self.assertFalse(Stream.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()))
self.assertFalse(Stream.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.
Expand All @@ -128,11 +128,11 @@ def seek(self, *a):

def tell(self):
return 0
self.assertFalse(_is_read_stream(ReadOnly()))
self.assertFalse(Stream.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")))
self.assertTrue(Stream.is_read_stream(io.BytesIO(b"x")))
self.assertTrue(Stream.is_read_stream(io.StringIO("x")))

def test_file_handles_are_streams(self):
path = self._tmp_path()
Expand All @@ -145,13 +145,13 @@ def test_file_handles_are_streams(self):
):
fh = opener()
try:
self.assertTrue(_is_read_stream(fh), type(fh).__name__)
self.assertTrue(Stream.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))
self.assertTrue(Stream.is_read_stream(fh))

def test_compressed_file_objects_are_streams(self):
with tempfile.TemporaryDirectory() as tmp:
Expand All @@ -164,7 +164,7 @@ def test_compressed_file_objects_are_streams(self):
with opener(p, "wb") as fh:
fh.write(b"data")
with opener(p, "rb") as fh:
self.assertTrue(_is_read_stream(fh), ext)
self.assertTrue(Stream.is_read_stream(fh), ext)

def test_placeholder_typed_object_with_all_methods_is_a_stream(self):
class Duck:
Expand All @@ -182,7 +182,7 @@ def tell(self):

def flush(self):
pass
self.assertTrue(_is_read_stream(Duck()))
self.assertTrue(Stream.is_read_stream(Duck()))


class TestFormatValidation(unittest.TestCase):
Expand Down Expand Up @@ -562,7 +562,7 @@ def test_try_create_bare_stream_keyword(self):
reader.close()

def test_bare_stream_closed_handle_raises(self):
# _is_read_stream only checks method presence
# Stream.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()
Expand Down
Loading