diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 38fba0ce..ad1864ce 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -1264,7 +1264,9 @@ class Reader: '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", + 'not_initialized_error': "Reader is not properly initialized", + 'no_manifest_error': "No manifest store to read" } @classmethod @@ -1350,6 +1352,7 @@ def __init__(self, self._closed = False self._initialized = False + self._no_manifest_to_read = False self._reader = None self._own_stream = None @@ -1397,6 +1400,12 @@ def __init__(self, error = _parse_operation_result_for_error( _lib.c2pa_error()) if error: + # Check if this is a ManifestNotFound error + if error.startswith("ManifestNotFound"): + # Set reader to no manifest state + self._no_manifest_to_read = True + self._initialized = True + return raise C2paError(error) raise C2paError( Reader._ERROR_MESSAGES['reader_error'].format( @@ -1460,6 +1469,12 @@ def __init__(self, error = _parse_operation_result_for_error( _lib.c2pa_error()) if error: + # Check if this is a ManifestNotFound error + if error.startswith("ManifestNotFound"): + # Set reader to no manifest state + self._no_manifest_to_read = True + self._initialized = True + return raise C2paError(error) raise C2paError( Reader._ERROR_MESSAGES['reader_error'].format( @@ -1514,6 +1529,12 @@ def __init__(self, error = _parse_operation_result_for_error( _lib.c2pa_error()) if error: + # Check if this is a ManifestNotFound error + if error.startswith("ManifestNotFound"): + # Set reader to no manifest state + self._no_manifest_to_read = True + self._initialized = True + return raise C2paError(error) raise C2paError( Reader._ERROR_MESSAGES['reader_error'].format( @@ -1539,16 +1560,18 @@ def __del__(self): self._cleanup_resources() def _ensure_valid_state(self): - """Ensure the reader is in a valid state for operations. + """Ensure the Reader is in a valid state for operations. Raises: - C2paError: If the reader is closed, not initialized, or invalid + C2paError: If the Reader is closed, not initialized, invalid, or + has no manifest to read. """ + # self._no_manifest_to_read is a valid state, albeit an empty one if self._closed: raise C2paError(Reader._ERROR_MESSAGES['closed_error']) if not self._initialized: - raise C2paError("Reader is not properly initialized") - if not self._reader: + raise C2paError(Reader._ERROR_MESSAGES['not_initialized_error']) + if not self._reader and not self._no_manifest_to_read: raise C2paError(Reader._ERROR_MESSAGES['closed_error']) def _cleanup_resources(self): @@ -1609,14 +1632,22 @@ def _get_cached_manifest_data(self) -> Optional[dict]: Returns: A dictionary containing the parsed manifest data, or None if - JSON parsing fails + JSON parsing fails or reader has no manifest to read Raises: C2paError: If there was an error getting the JSON """ + if self._no_manifest_to_read: + return None + + self._ensure_valid_state() + if self._manifest_data_cache is None: if self._manifest_json_str_cache is None: - self._manifest_json_str_cache = self.json() + manifest_json = self.json() + if manifest_json is None: + return None + self._manifest_json_str_cache = manifest_json try: self._manifest_data_cache = json.loads( @@ -1656,15 +1687,20 @@ def close(self): self._manifest_data_cache = None self._closed = True - def json(self) -> str: + def json(self) -> Optional[str]: """Get the manifest store as a JSON string. Returns: - The manifest store as a JSON string + The manifest store as a JSON string, or None if no manifest is + found Raises: - C2paError: If there was an error getting the JSON + C2paError: If there was an error getting the JSON (other than + ManifestNotFound) """ + # Check if reader is in no-manifest state first + if self._no_manifest_to_read: + return None self._ensure_valid_state() @@ -1677,7 +1713,11 @@ def json(self) -> str: if result is None: error = _parse_operation_result_for_error(_lib.c2pa_error()) if error: - raise C2paError(error) + # Check if this is a ManifestNotFound error + if error.startswith("ManifestNotFound"): + # Set reader to no manifest state and return None + self._no_manifest_to_read = True + return None raise C2paError("Error during manifest parsing in Reader") # Cache the result and return it @@ -1698,30 +1738,27 @@ def get_active_manifest(self) -> Optional[dict]: Raises: KeyError: If the active_manifest key is missing from the JSON """ - try: - # Get cached manifest data - manifest_data = self._get_cached_manifest_data() - if manifest_data is None: - # raise C2paError("Failed to parse manifest JSON") - return None + # Get cached manifest data + manifest_data = self._get_cached_manifest_data() + if manifest_data is None: + # raise C2paError("Failed to parse manifest JSON") + return None - # Get the active manfiest id/label - if "active_manifest" not in manifest_data: - raise KeyError("No 'active_manifest' key found") + # Get the active manfiest id/label + if "active_manifest" not in manifest_data: + raise KeyError("No 'active_manifest' key found") - active_manifest_id = manifest_data["active_manifest"] + active_manifest_id = manifest_data["active_manifest"] - # Retrieve the active manifest data using manifest id/label - if "manifests" not in manifest_data: - raise KeyError("No 'manifests' key found in manifest data") + # Retrieve the active manifest data using manifest id/label + if "manifests" not in manifest_data: + raise KeyError("No 'manifests' key found in manifest data") - manifests = manifest_data["manifests"] - if active_manifest_id not in manifests: - raise KeyError("Active manifest not found in manifest store") + manifests = manifest_data["manifests"] + if active_manifest_id not in manifests: + raise KeyError("Active manifest not found in manifest store") - return manifests[active_manifest_id] - except C2paError.ManifestNotFound: - return None + return manifests[active_manifest_id] def get_manifest_from_label(self, label: str) -> Optional[dict]: """Get a specific manifest from the manifest store by its label. @@ -1740,23 +1777,20 @@ def get_manifest_from_label(self, label: str) -> Optional[dict]: Raises: KeyError: If the manifests key is missing from the JSON """ - try: - # Get cached manifest data - manifest_data = self._get_cached_manifest_data() - if manifest_data is None: - # raise C2paError("Failed to parse manifest JSON") - return None + # Get cached manifest data + manifest_data = self._get_cached_manifest_data() + if manifest_data is None: + # raise C2paError("Failed to parse manifest JSON") + return None - if "manifests" not in manifest_data: - raise KeyError("No 'manifests' key found in manifest data") + if "manifests" not in manifest_data: + raise KeyError("No 'manifests' key found in manifest data") - manifests = manifest_data["manifests"] - if label not in manifests: - raise KeyError(f"Manifest {label} not found in manifest store") + manifests = manifest_data["manifests"] + if label not in manifests: + raise KeyError(f"Manifest {label} not found in manifest store") - return manifests[label] - except C2paError.ManifestNotFound: - return None + return manifests[label] def get_validation_state(self) -> Optional[str]: """Get the validation state of the manifest store. @@ -1770,16 +1804,13 @@ def get_validation_state(self) -> Optional[str]: or None if the validation_state field is not present or if no manifest is found or if there was an error parsing the JSON. """ - try: - # Get cached manifest data - manifest_data = self._get_cached_manifest_data() - if manifest_data is None: - return None - - return manifest_data.get("validation_state") - except C2paError.ManifestNotFound: + # Get cached manifest data + manifest_data = self._get_cached_manifest_data() + if manifest_data is None: return None + return manifest_data.get("validation_state") + def get_validation_results(self) -> Optional[dict]: """Get the validation results of the manifest store. @@ -1793,17 +1824,14 @@ def get_validation_results(self) -> Optional[dict]: field is not present or if no manifest is found or if there was an error parsing the JSON. """ - try: - # Get cached manifest data - manifest_data = self._get_cached_manifest_data() - if manifest_data is None: - return None - - return manifest_data.get("validation_results") - except C2paError.ManifestNotFound: + # Get cached manifest data + manifest_data = self._get_cached_manifest_data() + if manifest_data is None: return None - def resource_to_stream(self, uri: str, stream: Any) -> int: + return manifest_data.get("validation_results") + + def resource_to_stream(self, uri: str, stream: Any) -> Optional[int]: """Write a resource to a stream. Args: @@ -1811,11 +1839,14 @@ def resource_to_stream(self, uri: str, stream: Any) -> int: stream: The stream to write to (any Python stream-like object) Returns: - The number of bytes written + The number of bytes written, or None if no manifest to read Raises: C2paError: If there was an error writing the resource to stream """ + if self._no_manifest_to_read: + return None + self._ensure_valid_state() uri_str = uri.encode('utf-8') @@ -1833,16 +1864,20 @@ def resource_to_stream(self, uri: str, stream: Any) -> int: return result - def is_embedded(self) -> bool: + def is_embedded(self) -> Optional[bool]: """Check if the reader was created from an embedded manifest. This method determines whether the C2PA manifest is embedded directly in the asset file or stored remotely. Returns: True if the reader was created from an embedded manifest, - False if it was created from a remote manifest + False if it was created from a remote manifest, + None if no manifest to read Raises: C2paError: If there was an error checking the embedded status """ + if self._no_manifest_to_read: + return None + self._ensure_valid_state() result = _lib.c2pa_reader_is_embedded(self._reader) @@ -1856,10 +1891,14 @@ def get_remote_url(self) -> Optional[str]: is embedded in the asset, this will return None. Returns: The remote URL as a string if the manifest was obtained remotely, - None if the manifest is embedded or no remote URL is available + None if the manifest is embedded, no remote URL is available, + or no manifest to read Raises: C2paError: If there was an error getting the remote URL """ + if self._no_manifest_to_read: + return None + self._ensure_valid_state() result = _lib.c2pa_reader_remote_url(self._reader) @@ -2194,6 +2233,7 @@ class Builder: 'cleanup_error': "Error during cleanup: {}", 'builder_cleanup': "Error cleaning up builder: {}", 'closed_error': "Builder is closed", + 'not_initialized_error': "Builder is not properly initialized", 'manifest_error': "Invalid manifest data: must be string or dict", 'url_error': "Error setting remote URL: {}", 'resource_error': "Error adding resource: {}", @@ -2376,7 +2416,7 @@ def _ensure_valid_state(self): if self._closed: raise C2paError(Builder._ERROR_MESSAGES['closed_error']) if not self._initialized: - raise C2paError("Builder is not properly initialized") + raise C2paError(Builder._ERROR_MESSAGES['not_initialized_error']) if not self._builder: raise C2paError(Builder._ERROR_MESSAGES['closed_error']) diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index cf9b3b6c..a4e199d1 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -64,6 +64,18 @@ def test_stream_read(self): json_data = reader.json() self.assertIn(DEFAULT_TEST_FILE_NAME, json_data) + def test_stream_read_no_manifest(self): + # INGREDIENT_TEST_FILE does not have a manifest + with open(INGREDIENT_TEST_FILE, "rb") as file: + reader = Reader("image/jpeg", file) + # Reader should be created but in no-manifest state + self.assertIsNotNone(reader) + # Check that json() raises appropriate error for no-manifest files + self.assertIsNone(reader.json()) + + # Check that get_active_manifest() returns None for no-manifest files + self.assertIsNone(reader.get_active_manifest()) + def test_get_active_manifest(self): with open(self.testPath, "rb") as file: reader = Reader("image/jpeg", file) @@ -73,6 +85,11 @@ def test_get_active_manifest(self): expected_label = "contentauth:urn:uuid:c85a2b90-f1a0-4aa4-b17f-f938b475804e" self.assertEqual(active_manifest["label"], expected_label) + def test_get_active_manifest_no_manifest(self): + with open(INGREDIENT_TEST_FILE, "rb") as file: + reader = Reader("image/jpeg", file) + self.assertIsNone(reader.get_active_manifest()) + def test_get_manifest_from_label(self): with open(self.testPath, "rb") as file: reader = Reader("image/jpeg", file) @@ -86,6 +103,15 @@ def test_get_manifest_from_label(self): active_manifest = reader.get_active_manifest() self.assertEqual(manifest, active_manifest) + def test_get_manifest_from_label_no_manifest(self): + with open(INGREDIENT_TEST_FILE, "rb") as file: + reader = Reader("image/jpeg", file) + + # Test getting manifest by the specific label + label = "contentauth:urn:uuid:c85a2b90-f1a0-4aa4-b17f-f938b475804e" + manifest = reader.get_manifest_from_label(label) + self.assertIsNone(manifest) + def test_stream_get_non_active_manifest_by_label(self): video_path = os.path.join(FIXTURES_DIR, "video1.mp4") with open(video_path, "rb") as file: @@ -172,6 +198,11 @@ def test_stream_read_filepath_as_stream_and_parse(self): title = manifest_store["manifests"][manifest_store["active_manifest"]]["title"] self.assertEqual(title, DEFAULT_TEST_FILE_NAME) + def test_read_no_manifest_context_manager(self): + with Reader("image/jpeg", INGREDIENT_TEST_FILE) as reader: + self.assertIsNone(reader.json()) + self.assertIsNone(reader.get_active_manifest()) + def test_reader_double_close(self): with open(self.testPath, "rb") as file: reader = Reader("image/jpeg", file) @@ -475,6 +506,7 @@ def test_reader_partial_initialization_states(self): reader._reader = None reader._own_stream = None reader._backing_file = None + reader._no_manifest_to_read = False with self.assertRaises(Error): reader._ensure_valid_state() @@ -1169,8 +1201,8 @@ def test_remote_sign(self): output.seek(0) # When set_no_embed() is used, no manifest should be embedded in the file # So reading from the file should fail - with self.assertRaises(Error): - Reader("image/jpeg", output) + reader = Reader("image/jpeg", output) + self.assertIsNone(reader.json()) output.close() def test_remote_sign_using_returned_bytes(self):