From 6a79fc589082aac1796735734e054ee09f5e7dc9 Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 00:58:42 +0200 Subject: [PATCH 01/10] feat: verify single-file fMP4 fragments at their byte offset Bind c2pa_reader_with_fragment_at_offset and its probe c2pa_reader_fragment_at_offset_supported as optional native symbols, so libraries without them still import. - has_fragment_at_offset(): True when both symbols exist and the probe reports support. The probe runs on first use, not at import. - Reader.with_fragment_at_offset(format, stream, fragment_stream, fragment_offset) and Reader.from_fragment_at_offset(..., context=None). with_fragment and with_fragment_at_offset share one helper, so the offset variant gets with_fragment's locking, stream retention, cache invalidation and consume/swap handling unchanged. - Unsupported libraries raise C2paError.NotSupported, and offsets that are not an int in 0..2**64-1 (bool included) raise TypeError/ValueError, before any native call; the reader is left untouched. - A wrong cut or offset is reported as assertion.bmffHash.mismatch in the validation results, not raised. Tests: feature detection and validation run against any native library; native tests use the signed DASH fixtures (offset 0 matches with_fragment, other offsets are a mismatch) and are skipped without the native API, or fail under C2PA_FRAGMENT_OFFSET_REQUIRED=1. --- docs/usage.md | 15 ++ src/c2pa/__init__.py | 6 +- src/c2pa/c2pa.py | 199 ++++++++++++++++++++++- tests/test_fragment_at_offset.py | 261 +++++++++++++++++++++++++++++++ 4 files changed, 472 insertions(+), 9 deletions(-) create mode 100644 tests/test_fragment_at_offset.py diff --git a/docs/usage.md b/docs/usage.md index c7754e72..f28d8ff3 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -235,6 +235,21 @@ except Exception as err: print(err) ``` +### Verify one fragment of a single-file fragmented MP4 + +A single-file fragmented MP4 (for example one served as HLS byte ranges) hashes each fragment together with its absolute byte offset in the file. To verify one fragment on its own, pass the bytes before the first C2PA `merkle` box as the init stream, the fragment from its `merkle` box up to the next one, and the fragment's offset in the original file: + +```py +if c2pa.has_fragment_at_offset(): + with Context() as ctx: + with Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(init_bytes), io.BytesIO(fragment_bytes), + fragment_offset, context=ctx) as reader: + print(reader.get_validation_state()) +``` + +A wrong cut or offset is reported as `assertion.bmffHash.mismatch` in the validation results rather than raised. An offset of `0` behaves like `Reader.with_fragment()` for fragments stored as separate files. `has_fragment_at_offset()` is `False` when the native library predates this API; the methods then raise `C2paError.NotSupported`. + ### Add a signed manifest to a stream > [!WARNING] diff --git a/src/c2pa/__init__.py b/src/c2pa/__init__.py index bd581d58..961617af 100644 --- a/src/c2pa/__init__.py +++ b/src/c2pa/__init__.py @@ -32,7 +32,8 @@ ContextBuilder, ContextProvider, sdk_version, - load_settings + load_settings, + has_fragment_at_offset ) # NOQA # Re-export C2paError and its subclasses @@ -51,5 +52,6 @@ 'ContextBuilder', 'ContextProvider', 'sdk_version', - 'load_settings' + 'load_settings', + 'has_fragment_at_offset' ] diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index eca9e952..30525920 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -168,6 +168,20 @@ def _validate_library_exports(lib): _validate_library_exports(_lib) +# Offset-aware single-file fragment verification. Optional, so native +# libraries that predate it still import; see has_fragment_at_offset(). +_FRAGMENT_AT_OFFSET_FUNCTIONS = ( + 'c2pa_reader_with_fragment_at_offset', + 'c2pa_reader_fragment_at_offset_supported', +) +_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE = all( + hasattr(_lib, name) for name in _FRAGMENT_AT_OFFSET_FUNCTIONS +) +_FRAGMENT_OFFSET_MAX = 2**64 - 1 +# Cached probe result; None until has_fragment_at_offset() first runs, so +# importing the module never calls into the native library for it. +_fragment_at_offset_probe = None + class C2paSeekMode(enum.IntEnum): """Seek mode for stream operations.""" @@ -1503,6 +1517,16 @@ def _setup_function(func, argtypes, restype=None): ctypes.POINTER(C2paStream), ctypes.POINTER(C2paStream)], ctypes.POINTER(C2paReader) ) +if _FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE: + _setup_function( + _lib.c2pa_reader_with_fragment_at_offset, + [ctypes.POINTER(C2paReader), ctypes.c_char_p, + ctypes.POINTER(C2paStream), ctypes.POINTER(C2paStream), + ctypes.c_uint64], + ctypes.POINTER(C2paReader) + ) + _setup_function( + _lib.c2pa_reader_fragment_at_offset_supported, [], ctypes.c_bool) _setup_function( _lib.c2pa_builder_from_context, [ctypes.POINTER(C2paContext)], @@ -1941,6 +1965,42 @@ def _to_utf8_bytes(data: Union[str, dict], raise C2paError.Encoding(f"Invalid UTF-8 in {error_context}: {e}") +def has_fragment_at_offset() -> bool: + """Return whether the native library supports offset-aware fragment + verification (:meth:`Reader.with_fragment_at_offset`). + + True only when the library exports both + ``c2pa_reader_with_fragment_at_offset`` and + ``c2pa_reader_fragment_at_offset_supported`` and the latter reports + support. The probe runs on first use, not at import, and is cached. + """ + global _fragment_at_offset_probe + if not _FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE: + return False + if _fragment_at_offset_probe is None: + _fragment_at_offset_probe = bool( + _lib.c2pa_reader_fragment_at_offset_supported()) + return _fragment_at_offset_probe + + +def _require_fragment_at_offset() -> None: + if not has_fragment_at_offset(): + raise C2paError.NotSupported( + "Offset-aware fragment verification is not available in the " + "loaded native library (c2pa_reader_with_fragment_at_offset)") + + +def _check_fragment_offset(fragment_offset) -> int: + """Validate a fragment offset before any native work.""" + if isinstance(fragment_offset, bool) or not isinstance( + fragment_offset, int): + raise TypeError("fragment_offset must be an int") + if not 0 <= fragment_offset <= _FRAGMENT_OFFSET_MAX: + raise ValueError( + f"fragment_offset must be between 0 and {_FRAGMENT_OFFSET_MAX}") + return fragment_offset + + def sdk_version() -> str: """ Returns the underlying c2pa-rs/c2pa-c-ffi version string @@ -3402,7 +3462,135 @@ def with_fragment(self, format: Optional[str], stream, this call refused; yield between reads. """ format_arg = _format_ffi_arg(_encode_format(format, "Reader")) + return self._swap_in_fragment( + format_arg, stream, fragment_stream, + lambda handle, main, frag: _lib.c2pa_reader_with_fragment( + handle, format_arg, main, frag)) + + def with_fragment_at_offset(self, format: Optional[str], stream, + fragment_stream, + fragment_offset: int) -> "Reader": + """Process one fragment of a single-file fragmented BMFF asset, + verified at its absolute byte offset in that asset. + + A single-file fragmented MP4 hashes its fragments together with their + absolute root-box offsets, so a byte-range slice of it (for example an + HLS ``EXT-X-BYTERANGE`` segment) only verifies when the verifier knows + where the slice sat in the file. ``fragment_offset=0`` behaves like + :meth:`with_fragment`. + + With the default exclusions, ``stream`` holds every byte before the + first C2PA ``merkle`` uuid box, and ``fragment_stream`` starts at a + ``merkle`` uuid box and ends right before the next one (or at the end + of the file). A wrong cut or offset does not raise: the reader reports + ``assertion.bmffHash.mismatch``. The offset is supplied by the caller; + it proves the bytes belong to the Merkle leaf they name, not where + they were fetched from. + + Settings and trust come from this reader's Context. + + Args: + format: MIME type of the media (e.g., "video/mp4"), handled as + in :meth:`with_fragment`. + stream: Stream-like object with the initialization bytes. + fragment_stream: Stream-like object with the fragment bytes. + fragment_offset: Absolute byte offset of the fragment's first + byte in the original asset, from 0 to 2**64 - 1. + + Returns: + This reader instance, for method chaining. + + Raises: + C2paError.NotSupported: If the native library lacks the API + (see :func:`has_fragment_at_offset`). The reader is untouched. + TypeError, ValueError: For an invalid ``fragment_offset``. The + reader is untouched. + C2paError: As :meth:`with_fragment`, including the reader being + closed after a native failure. + """ + _require_fragment_at_offset() + offset = _check_fragment_offset(fragment_offset) + format_arg = _format_ffi_arg(_encode_format(format, "Reader")) + return self._swap_in_fragment( + format_arg, stream, fragment_stream, + lambda handle, main, frag: ( + _lib.c2pa_reader_with_fragment_at_offset( + handle, format_arg, main, frag, offset))) + @classmethod + def from_fragment_at_offset( + cls, + format: Optional[str], + init_stream: Any, + fragment_stream: Any, + fragment_offset: int, + context: Optional['ContextProvider'] = None, + ) -> "Reader": + """Create a Reader for one fragment of a single-file fragmented + BMFF asset, verified at its absolute byte offset. + + Unlike ``Reader(format, init_stream, context=...)`` followed by + :meth:`with_fragment_at_offset`, the init stream is not read and + validated on its own first. See :meth:`with_fragment_at_offset` for + the cut rules and result semantics. + + Args: + format: MIME type of the media (e.g., "video/mp4"). + init_stream: Stream-like object with the initialization bytes. + fragment_stream: Stream-like object with the fragment bytes. + fragment_offset: Absolute byte offset of the fragment's first + byte in the original asset, from 0 to 2**64 - 1. + context: Optional ContextProvider supplying settings and trust. + When omitted, a default Context is used. + + Raises: + C2paError.NotSupported: If the native library lacks the API. + TypeError, ValueError: For an invalid argument, before any + native object is created. + C2paError: If the reader cannot be created or the fragment + cannot be processed. + """ + _require_fragment_at_offset() + _check_fragment_offset(fragment_offset) + if context is not None: + if not isinstance(context, ContextProvider): + raise TypeError("context must be a ContextProvider") + if not context.is_valid: + raise C2paError("Context is not valid") + + owned_context = Context() if context is None else None + active_context = context if context is not None else owned_context + try: + reader = object.__new__(cls) + ManagedResource.__init__(reader) + reader._init_attrs() + with _context_guard(active_context): + reader._create_and_activate( + lambda: _lib.c2pa_reader_from_context( + active_context.execution_context), + Reader._ERROR_MESSAGES['reader_error']) + if context is not None: + # Pin a caller-supplied Context for the reader's lifetime. + reader._context = context + try: + reader.with_fragment_at_offset( + format, init_stream, fragment_stream, fragment_offset) + except Exception: + reader.close() + raise + return reader + finally: + # The native reader holds its own reference to the context. + if owned_context is not None: + owned_context.close() + + def _swap_in_fragment(self, format_arg, stream, fragment_stream, + native_call) -> "Reader": + """Run a consuming fragment call and keep its streams. + + ``native_call(handle, main_stream, fragment_stream)`` consumes the + reader handle and returns the replacement (NULL on failure). + """ # A forked child cannot wait on a lock no surviving thread will # release, so it reports the same error _live_op_lock() does. if is_foreign_process(self): @@ -3429,12 +3617,8 @@ def with_fragment_call(handle): # and a reader must never be served them. self._manifest_json_str_cache = None self._manifest_data_cache = None - return _lib.c2pa_reader_with_fragment( - handle, - format_arg, - main_obj._stream, - frag_obj._stream, - ) + return native_call( + handle, main_obj._stream, frag_obj._stream) self._consume_and_swap( with_fragment_call, @@ -4896,5 +5080,6 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes: 'load_settings', 'format_embeddable', 'version', - 'sdk_version' + 'sdk_version', + 'has_fragment_at_offset' ] diff --git a/tests/test_fragment_at_offset.py b/tests/test_fragment_at_offset.py new file mode 100644 index 00000000..2a5d4762 --- /dev/null +++ b/tests/test_fragment_at_offset.py @@ -0,0 +1,261 @@ +# Copyright 2026 Adobe. All rights reserved. +# This file is licensed to you under the Apache License, +# Version 2.0 (http://www.apache.org/licenses/LICENSE-2.0) +# or the MIT license (http://opensource.org/licenses/MIT), +# at your option. + +# Unless required by applicable law or agreed to in writing, +# this software is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR REPRESENTATIONS OF ANY KIND, either express or +# implied. See the LICENSE-MIT and LICENSE-APACHE files for the +# specific language governing permissions and limitations under +# each license. + +"""Tests for Reader.with_fragment_at_offset / from_fragment_at_offset. + +Two groups: + +* Binding behavior that does not need the new native entry point: feature + detection, argument validation, and that a refused call leaves the reader + untouched. These run against any native library. +* Native behavior, against a library that exports + c2pa_reader_with_fragment_at_offset. They are skipped when the loaded + library lacks it, unless C2PA_FRAGMENT_OFFSET_REQUIRED=1, which turns the + skip into a failure for CI lanes that build the native library. + +The native tests use the signed multi-file DASH fixtures: a multi-file +fragment verifies at offset 0 and is a hash mismatch at any other offset. The +offset arithmetic on single-file layouts is covered by the c2pa-rs tests. +""" + +import io +import json +import os +import unittest +from unittest.mock import patch + +import c2pa +import c2pa.c2pa as c2pa_module +from c2pa import C2paError as Error, Context, Reader + +FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures") +INIT_PATH = os.path.join(FIXTURES_DIR, "dashinit.mp4") +FRAGMENT_PATH = os.path.join(FIXTURES_DIR, "dash1.m4s") +DEFAULT_TEST_FILE = os.path.join(FIXTURES_DIR, "C.jpg") +BMFF_MISMATCH = "assertion.bmffHash.mismatch" +REQUIRED = os.environ.get("C2PA_FRAGMENT_OFFSET_REQUIRED") == "1" + + +def _read(path): + with open(path, "rb") as f: + return f.read() + + +def _failure_codes(reader): + store = json.loads(reader.json()) + return [s.get("code") for s in store.get("validation_status") or []] + + +class TestFragmentAtOffsetFeatureDetection(unittest.TestCase): + + def test_exported_from_package(self): + self.assertIs(c2pa.has_fragment_at_offset, + c2pa_module.has_fragment_at_offset) + self.assertIn("has_fragment_at_offset", c2pa.__all__) + + def test_false_without_symbols(self): + with patch.object( + c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", False): + self.assertFalse(c2pa.has_fragment_at_offset()) + + def test_probe_result_is_used_and_cached(self): + with patch.object( + c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", True), \ + patch.object(c2pa_module, "_fragment_at_offset_probe", None), \ + patch.object(c2pa_module, "_lib") as lib: + lib.c2pa_reader_fragment_at_offset_supported.return_value = False + self.assertFalse(c2pa.has_fragment_at_offset()) + self.assertFalse(c2pa.has_fragment_at_offset()) + lib.c2pa_reader_fragment_at_offset_supported.assert_called_once() + + def test_unsupported_raises_before_touching_the_reader(self): + context = Context() + reader = Reader(DEFAULT_TEST_FILE, context=context) + before = reader.json() + with patch.object( + c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", False): + with self.assertRaises(Error.NotSupported): + reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(b"init"), io.BytesIO(b"frag"), 0) + with self.assertRaises(Error.NotSupported): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"init"), io.BytesIO(b"frag"), 0) + self.assertTrue(reader.is_valid) + self.assertEqual(reader.json(), before) + reader.close() + context.close() + + +class TestFragmentAtOffsetValidation(unittest.TestCase): + """Invalid offsets are refused before any native call.""" + + def setUp(self): + self.patch = patch.object( + c2pa_module, "has_fragment_at_offset", return_value=True) + self.patch.start() + self.context = Context() + self.reader = Reader(DEFAULT_TEST_FILE, context=self.context) + self.before = self.reader.json() + + def tearDown(self): + self.reader.close() + self.context.close() + self.patch.stop() + + def assert_untouched(self): + self.assertTrue(self.reader.is_valid) + self.assertEqual(self.reader.json(), self.before) + + def test_rejects_non_int_offsets(self): + for offset in (True, False, 1.0, "1", None, b"\x01"): + with self.subTest(offset=offset): + with self.assertRaises(TypeError): + self.reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), + offset) + with self.assertRaises(TypeError): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), + offset) + self.assert_untouched() + + def test_rejects_out_of_range_offsets(self): + for offset in (-1, 2**64, 2**70): + with self.subTest(offset=offset): + with self.assertRaises(ValueError): + self.reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), + offset) + with self.assertRaises(ValueError): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), + offset) + self.assert_untouched() + + def test_from_rejects_invalid_context(self): + with self.assertRaises(TypeError): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), 0, + context=object()) + closed = Context() + closed.close() + with self.assertRaises(Error): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), 0, + context=closed) + + +@unittest.skipUnless( + c2pa.has_fragment_at_offset() or REQUIRED, + "native library lacks c2pa_reader_with_fragment_at_offset") +class TestFragmentAtOffsetNative(unittest.TestCase): + + @classmethod + def setUpClass(cls): + if not c2pa.has_fragment_at_offset(): + raise AssertionError( + "C2PA_FRAGMENT_OFFSET_REQUIRED=1 but the native library " + "lacks c2pa_reader_with_fragment_at_offset") + cls.init = _read(INIT_PATH) + cls.fragment = _read(FRAGMENT_PATH) + + def setUp(self): + self.context = Context() + + def tearDown(self): + self.context.close() + + def with_fragment_codes(self): + with Reader("video/mp4", io.BytesIO(self.init), + context=self.context) as reader: + reader.with_fragment("video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment)) + return _failure_codes(reader) + + def test_zero_offset_matches_with_fragment(self): + expected = self.with_fragment_codes() + self.assertNotIn(BMFF_MISMATCH, expected) + with Reader("video/mp4", io.BytesIO(self.init), + context=self.context) as reader: + result = reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 0) + self.assertIs(result, reader) + self.assertEqual(_failure_codes(reader), expected) + + def test_from_fragment_at_offset_zero(self): + expected = self.with_fragment_codes() + for context in (self.context, None): + with self.subTest(context=context): + with Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 0, + context=context) as reader: + self.assertEqual(_failure_codes(reader), expected) + + def test_wrong_offset_is_a_mismatch_not_an_error(self): + for offset in (1, len(self.init), 2**32, 2**63): + with self.subTest(offset=offset): + with Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), offset, + context=self.context) as reader: + self.assertIn(BMFF_MISMATCH, _failure_codes(reader)) + + def test_maximum_offset_never_verifies(self): + # Offset arithmetic overflows in the verifier: either an error or a + # failure, never a clean result. + try: + reader = Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 2**64 - 1, context=self.context) + except Error: + return + with reader: + self.assertTrue(_failure_codes(reader)) + + def test_replacing_the_fragment_refreshes_results(self): + with Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 1, + context=self.context) as reader: + self.assertIn(BMFF_MISMATCH, _failure_codes(reader)) + reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 0) + self.assertNotIn(BMFF_MISMATCH, _failure_codes(reader)) + + def test_native_failure_closes_the_reader(self): + reader = Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(self.init), + io.BytesIO(self.fragment), 0, context=self.context) + with self.assertRaises(Error): + reader.with_fragment_at_offset( + "video/mp4", io.BytesIO(b"\x00" * 16), + io.BytesIO(b"\x00" * 16), 0) + self.assertFalse(reader.is_valid) + with self.assertRaises(Error): + reader.json() + reader.close() + + def test_from_fails_cleanly_without_a_manifest(self): + with self.assertRaises(Error): + Reader.from_fragment_at_offset( + "video/mp4", io.BytesIO(b"\x00" * 16), + io.BytesIO(b"\x00" * 16), 0, context=self.context) + # The caller's context stays usable. + self.assertTrue(self.context.is_valid) + + +if __name__ == "__main__": + unittest.main() From c96a864b3f7aaffb40e2b7f69373495eebb2920f Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 04:19:45 +0200 Subject: [PATCH 02/10] fix: validate fragment arguments before native allocation --- src/c2pa/c2pa.py | 39 +++++++++++++++++++++++------ tests/test_fragment_at_offset.py | 42 +++++++++++++++++++++++++++++++- 2 files changed, 72 insertions(+), 9 deletions(-) diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 30525920..5a91d873 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -2945,6 +2945,22 @@ def _encode_format( f"Invalid UTF-8 characters in input: {e}") +def _fragment_offset_format_arg( + format: Optional[str], stream: Any, fragment_stream: Any, +) -> bytes: + """Validate offset-aware fragment inputs before allocating native objects.""" + if format is not None and not isinstance(format, str): + raise TypeError("format must be a str or None") + format_arg = _format_ffi_arg(_encode_format(format, "Reader")) + _check_cstr_arg('format', format_arg) + for name, value in (("stream", stream), ("fragment_stream", fragment_stream)): + if not Stream.is_read_stream(value): + raise TypeError(f"{name} must be a stream-like object") + if stream is fragment_stream: + raise ValueError("stream and fragment_stream must be distinct objects") + return format_arg + + def _format_ffi_arg(fmt: Optional[bytes]) -> bytes: """Convert an encoded format to the native FFI argument. @@ -3482,10 +3498,12 @@ def with_fragment_at_offset(self, format: Optional[str], stream, With the default exclusions, ``stream`` holds every byte before the first C2PA ``merkle`` uuid box, and ``fragment_stream`` starts at a ``merkle`` uuid box and ends right before the next one (or at the end - of the file). A wrong cut or offset does not raise: the reader reports - ``assertion.bmffHash.mismatch``. The offset is supplied by the caller; - it proves the bytes belong to the Merkle leaf they name, not where - they were fetched from. + of the file). With a valid init and a parseable fragment, hash failures + caused by a wrong cut or offset are reported as + ``assertion.bmffHash.mismatch`` rather than raising. Init-loading or + fragment-parsing errors raise ``C2paError``. The offset is supplied by + the caller; it proves the bytes belong to the Merkle leaf they name, + not where they were fetched from. Settings and trust come from this reader's Context. @@ -3503,14 +3521,16 @@ def with_fragment_at_offset(self, format: Optional[str], stream, Raises: C2paError.NotSupported: If the native library lacks the API (see :func:`has_fragment_at_offset`). The reader is untouched. - TypeError, ValueError: For an invalid ``fragment_offset``. The + TypeError, ValueError: For an invalid offset, format type, or + streams. The reader is untouched. + C2paError: For a NUL byte or invalid UTF-8 in the format. The reader is untouched. C2paError: As :meth:`with_fragment`, including the reader being closed after a native failure. """ _require_fragment_at_offset() offset = _check_fragment_offset(fragment_offset) - format_arg = _format_ffi_arg(_encode_format(format, "Reader")) + format_arg = _fragment_offset_format_arg(format, stream, fragment_stream) return self._swap_in_fragment( format_arg, stream, fragment_stream, lambda handle, main, frag: ( @@ -3545,13 +3565,16 @@ def from_fragment_at_offset( Raises: C2paError.NotSupported: If the native library lacks the API. - TypeError, ValueError: For an invalid argument, before any - native object is created. + TypeError, ValueError: For an invalid offset, format type, + streams, or context type, before any native object is created. + C2paError: For a NUL byte or invalid UTF-8 in the format, or an + invalid context, before any native object is created. C2paError: If the reader cannot be created or the fragment cannot be processed. """ _require_fragment_at_offset() _check_fragment_offset(fragment_offset) + _fragment_offset_format_arg(format, init_stream, fragment_stream) if context is not None: if not isinstance(context, ContextProvider): raise TypeError("context must be a ContextProvider") diff --git a/tests/test_fragment_at_offset.py b/tests/test_fragment_at_offset.py index 2a5d4762..9b8db68c 100644 --- a/tests/test_fragment_at_offset.py +++ b/tests/test_fragment_at_offset.py @@ -32,6 +32,7 @@ import json import os import unittest +from pathlib import Path from unittest.mock import patch import c2pa @@ -97,7 +98,7 @@ def test_unsupported_raises_before_touching_the_reader(self): class TestFragmentAtOffsetValidation(unittest.TestCase): - """Invalid offsets are refused before any native call.""" + """Invalid arguments are refused before any native allocation or mutation.""" def setUp(self): self.patch = patch.object( @@ -116,6 +117,45 @@ def assert_untouched(self): self.assertTrue(self.reader.is_valid) self.assertEqual(self.reader.json(), self.before) + def assert_invalid_inputs(self, error, format, init, fragment): + handle = self.reader._handle + streams = (self.reader._own_stream, self.reader._fragment_streams) + with patch.object(c2pa_module, "_lib") as lib: + with self.assertRaises(error): + Reader.from_fragment_at_offset(format, init, fragment, 0) + with self.assertRaises(error): + self.reader.with_fragment_at_offset(format, init, fragment, 0) + # Includes Context, Reader, and Stream allocation entry points. + self.assertEqual(lib.mock_calls, []) + self.assertIs(self.reader._handle, handle) + self.assertIs(self.reader._own_stream, streams[0]) + self.assertIs(self.reader._fragment_streams, streams[1]) + self.assert_untouched() + + def test_rejects_invalid_streams_before_allocation(self): + for invalid in (None, "init.mp4", Path("init.mp4"), b"init", object()): + for position in (0, 1): + with self.subTest(invalid=invalid, position=position): + streams = [io.BytesIO(b"i"), io.BytesIO(b"f")] + streams[position] = invalid + self.assert_invalid_inputs(TypeError, "video/mp4", *streams) + + def test_rejects_aliased_streams_before_allocation(self): + stream = io.BytesIO(b"i") + self.assert_invalid_inputs(ValueError, "video/mp4", stream, stream) + + def test_rejects_invalid_format_types_before_allocation(self): + for format in (False, 0, 1, b"video/mp4", [], {}, object()): + with self.subTest(format=format): + self.assert_invalid_inputs( + TypeError, format, io.BytesIO(b"i"), io.BytesIO(b"f")) + + def test_rejects_invalid_format_encoding_before_allocation(self): + for format in ("video/mp4\x00", "\x00video/mp4", "video/\ud800"): + with self.subTest(format=format): + self.assert_invalid_inputs( + Error, format, io.BytesIO(b"i"), io.BytesIO(b"f")) + def test_rejects_non_int_offsets(self): for offset in (True, False, 1.0, "1", None, b"\x01"): with self.subTest(offset=offset): From 036e9bf159eae4bd1f4834445b62999a8af74aef Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 08:24:13 +0200 Subject: [PATCH 03/10] ci: qualify fragment offsets against pinned native source in fork --- .../fragment-offset-qualification.yml | 78 ++++++ scripts/fragment_offset_qualification.py | 251 ++++++++++++++++++ scripts/test_fragment_offset_qualification.py | 157 +++++++++++ 3 files changed, 486 insertions(+) create mode 100644 .github/workflows/fragment-offset-qualification.yml create mode 100644 scripts/fragment_offset_qualification.py create mode 100644 scripts/test_fragment_offset_qualification.py diff --git a/.github/workflows/fragment-offset-qualification.yml b/.github/workflows/fragment-offset-qualification.yml new file mode 100644 index 00000000..e19a97fa --- /dev/null +++ b/.github/workflows/fragment-offset-qualification.yml @@ -0,0 +1,78 @@ +name: Fork fragment offset qualification + +on: + pull_request: + types: [opened, reopened, synchronize, labeled] + +permissions: + contents: read + +jobs: + qualify: + if: >- + github.repository == 'mstattma/c2pa-python' && + (github.event.pull_request.author_association == 'OWNER' || + github.event.pull_request.author_association == 'MEMBER' || + github.event.pull_request.author_association == 'COLLABORATOR' || + contains(github.event.pull_request.labels.*.name, 'safe to test')) + name: Fragment offset (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + timeout-minutes: 90 + strategy: + fail-fast: false + matrix: + os: [ubuntu-24.04, windows-2022] + env: + CARGO_INCREMENTAL: "0" + CARGO_TERM_COLOR: never + CARGO_HOME: ${{ runner.temp }}/fragment-cargo + PIP_NO_CACHE_DIR: "1" + PYTHONUNBUFFERED: "1" + defaults: + run: + shell: bash + steps: + - name: Checkout exact PR Python + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + ref: ${{ github.event.pull_request.head.sha }} + path: candidate + persist-credentials: false + - name: Checkout exact baseline Python + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + ref: ${{ github.event.pull_request.base.sha }} + path: baseline + persist-credentials: false + - name: Checkout allowlisted native source + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + repository: mstattma/c2pa-rs + ref: 20d7c45972976c32ab24d5d4ac33e7be826ce1be + path: native + persist-credentials: false + - name: Python 3.12 + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5 + with: + python-version: "3.12" + - name: Test qualification helpers (no native imports) + run: python -m unittest discover -s candidate/scripts -p test_fragment_offset_qualification.py -v + - name: Build exact native candidate and record evidence + run: python candidate/scripts/fragment_offset_qualification.py build + - name: Install Python-only test dependencies + run: | + set -euo pipefail + python -m pip install -r candidate/requirements.txt pytest toml requests + python -m pip install --no-deps -e candidate + - name: Stock compatibility, same-binary baseline and PR, focused offsets + env: + GITHUB_TOKEN: ${{ github.token }} + run: python candidate/scripts/fragment_offset_qualification.py qualify + - name: Retain qualification evidence (not a release) + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: fragment-offset-${{ matrix.os }}-${{ github.event.pull_request.head.sha }} + path: evidence/ + retention-days: 14 + if-no-files-found: error diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py new file mode 100644 index 00000000..64ed0cc3 --- /dev/null +++ b/scripts/fragment_offset_qualification.py @@ -0,0 +1,251 @@ +"""Non-publishing fork qualification. Native build mode is GitHub Actions only.""" + +import hashlib +import json +import os +from pathlib import Path +import re +import shutil +import subprocess +import sys +import tempfile +import tomllib +import xml.etree.ElementTree as ET + + +NATIVE_REPOSITORY = "mstattma/c2pa-rs" +NATIVE_SHA = "20d7c45972976c32ab24d5d4ac33e7be826ce1be" +NATIVE_VERSION = "0.92.0-dev" +FEATURES = "rust_native_crypto,add_thumbnails,http,file_io" +RUST = "1.96.0" +SUITES = ("test_unit_tests.py", "test_unit_tests_threaded.py") +ROOT = Path(__file__).resolve().parents[2] +EVIDENCE = ROOT / "evidence" + + +def digest(path): + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def record(name, value): + (EVIDENCE / name).write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8") + + +def command(args, cwd, name, env=None, test=False): + """Only pytest's ordinary failure status is deferred for comparison.""" + with (EVIDENCE / name).open("w", encoding="utf-8") as log: + result = subprocess.run(args, cwd=cwd, env=env, stdout=log, + stderr=subprocess.STDOUT, text=True) + print(f"{name}: exit {result.returncode}", flush=True) + if result.returncode not in ((0, 1) if test else (0,)): + raise subprocess.CalledProcessError(result.returncode, args) + return result.returncode + + +def git(path, *args): + return subprocess.check_output(["git", "-C", str(path), *args], text=True).strip() + + +def validate_native(path): + if os.environ.get("GITHUB_ACTIONS") != "true" or os.environ.get( + "GITHUB_REPOSITORY") != "mstattma/c2pa-python": + raise RuntimeError("Native qualification builds are fork GitHub Actions only") + if not re.fullmatch(r"[0-9a-f]{40}", NATIVE_SHA): + raise ValueError("Native SHA must be full lowercase hex") + if git(path, "remote", "get-url", "origin") != f"https://github.com/{NATIVE_REPOSITORY}": + raise ValueError("Native repository is not allowlisted") + if git(path, "rev-parse", "HEAD") != NATIVE_SHA: + raise ValueError("Native SHA is not allowlisted") + if git(path, "status", "--porcelain"): + raise ValueError("Native checkout must be clean before build") + if git(path, "ls-files", "Cargo.lock") or (path / "Cargo.lock").exists(): + raise ValueError("Expected an untracked, absent native Cargo.lock") + with (path / "Cargo.toml").open("rb") as file: + if tomllib.load(file)["workspace"]["package"]["version"] != NATIVE_VERSION: + raise ValueError("Unexpected native version") + + +def build(): + native = ROOT / "native" + validate_native(native) # Before installing tools or executing any native source. + record("sources.json", { + "native_repository": NATIVE_REPOSITORY, "native_sha": NATIVE_SHA, + "native_version": NATIVE_VERSION, "features": FEATURES, + "default_features": False, "profile": "release", "rust": RUST, + "python_head": git(ROOT / "candidate", "rev-parse", "HEAD"), + "python_base": git(ROOT / "baseline", "rev-parse", "HEAD"), + "run_id": os.environ["GITHUB_RUN_ID"], "python": sys.version, + }) + command(["rustup", "toolchain", "install", RUST, "--profile", "minimal"], + native, "rustup.log") + command(["rustc", f"+{RUST}", "-vV"], native, "rustc.log") + cargo = ["cargo", f"+{RUST}"] + command(cargo + ["--version", "--verbose"], native, "cargo.log") + command(cargo + ["generate-lockfile"], native, "resolve.log") + shutil.copy2(native / "Cargo.lock", EVIDENCE / "Cargo.lock") + lock_digest = digest(native / "Cargo.lock") + record("resolved-lock.json", {"sha256": lock_digest, "generated_in_job": True}) + flags = ["--locked", "-p", "c2pa-c-ffi", "--no-default-features", "--features", FEATURES] + command(cargo + ["tree", *flags, "-e", "features"], native, "features.log") + command(cargo + ["build", "--release", *flags], native, "build.log") + if digest(native / "Cargo.lock") != lock_digest: + raise RuntimeError("Resolved lock changed during locked build") + name = "c2pa_c.dll" if sys.platform == "win32" else "libc2pa_c.so" + library = EVIDENCE / name + shutil.copy2(native / "target" / "release" / name, library) + record("native.json", {"library": str(library), "sha256": digest(library), + "resolved_lock_sha256": lock_digest}) + + +def suite(source, library, expected_hash, version, test_file, label): + # This runs in a fresh process, with PYTHONPATH and the absolute native path set. + import c2pa.c2pa as module + import pytest + + loaded = Path(module._lib._name) + if not loaded.is_absolute() or loaded.resolve() != library: + raise RuntimeError(f"Wrong loaded native library: {loaded}") + if digest(loaded) != expected_hash or Path(module.__file__).resolve() != source / "src/c2pa/c2pa.py": + raise RuntimeError("Native digest or Python module provenance mismatch") + if version not in module.sdk_version(): + raise RuntimeError(f"Unexpected SDK version: {module.sdk_version()}") + record(f"{label}-loaded.json", {"module": module.__file__, "library": str(loaded), + "sha256": digest(loaded), "version": module.sdk_version()}) + + class FailureTypes: + # Pytest's JUnit omits exception types; preserve them explicitly rather + # than considering two different exceptions with the same message equal. + @pytest.hookimpl(hookwrapper=True) + def pytest_runtest_makereport(self, item, call): + report = (yield).get_result() + if report.failed and call.excinfo is not None: + cls = call.excinfo.type + item.user_properties.append(("exception_type", f"{cls.__module__}.{cls.__qualname__}")) + + return pytest.main([f"tests/{test_file}", "-v", "--tb=long", + "-o", "junit_family=xunit2", f"--junitxml={EVIDENCE / (label + '.xml')}"], + plugins=[FailureTypes()]) + + +def normalize(message): + message = re.sub(r"\\+", "/", message) + for folder in (ROOT / "candidate", ROOT / "baseline", ROOT): + message = message.replace(folder.as_posix(), "") + temporary = re.escape(Path(tempfile.gettempdir()).as_posix()) + message = re.sub(temporary + r"/tmp[a-zA-Z0-9_-]{8}(?=[/.\s'\"<>:)]|$)", "", message) + return re.sub(r"0x[0-9a-fA-F]+", "0xADDR", message) + + +def cases(path): + result = {} + for case in ET.parse(path).iter("testcase"): + key = case.attrib["classname"] + "::" + case.attrib["name"] + if key in result: + raise ValueError(f"Duplicate test ID: {key}") + outcome = ("passed", "", "") + for kind in ("failure", "error", "skipped"): + node = case.find(kind) + if node is not None: + exception_type = case.find("./properties/property[@name='exception_type']") + failure_type = node.get("type", "") if exception_type is None else exception_type.get("value", "") + if kind != "skipped" and not failure_type: + raise ValueError(f"Missing exception type for {key}") + outcome = (kind, failure_type, normalize(node.get("message", ""))) + break + result[key] = outcome + if not result: + raise ValueError(f"No tests executed: {path}") + return result + + +def compare(baseline, candidate): + problems = [f"Missing common test: {key}" for key in baseline.keys() - candidate.keys()] + for key, outcome in candidate.items(): + previous = baseline.get(key, ("passed", "", "")) + if outcome[0] != "passed" and outcome != previous: + problems.append(f"New/changed outcome: {key}: {outcome!r}; baseline: {previous!r}") + return sorted(problems) + + +def qualify(): + candidate, baseline = ROOT / "candidate", ROOT / "baseline" + native = json.loads((EVIDENCE / "native.json").read_text(encoding="utf-8")) + library = Path(native["library"]).resolve() + # Keep ordinary release pins untouched; only ephemeral preflight files differ. + for source in (candidate, baseline): + (source / "c2pa-rs-preflight-ref.txt").write_text(f"c2pa-v{NATIVE_VERSION}\n", encoding="utf-8") + command([sys.executable, "scripts/download_artifacts.py", "c2pa-v0.91.0"], + candidate, "stock-download.log") + name = library.name + stock = list((candidate / "artifacts").glob(f"*/{name}")) + if len(stock) != 1: + raise RuntimeError(f"Expected one downloaded stock library, found {stock}") + stock = stock[0].resolve() + shutil.copy2(stock, EVIDENCE / ("stock-" + name)) + results, reports, problems = {}, {}, [] + lanes = (("stock", candidate, stock, digest(stock), "0.91.0"), + ("baseline", baseline, library, native["sha256"], NATIVE_VERSION), + ("candidate", candidate, library, native["sha256"], NATIVE_VERSION)) + for lane, source, binary, sha, version in lanes: + files = SUITES + (("test_fragment_at_offset.py",) if lane != "baseline" else ()) + for test_file in files: + label = lane + "-" + test_file.removesuffix(".py") + env = dict(os.environ, PYTHONPATH=str(source / "src"), + C2PA_LIBRARY_NAME=str(binary), PYTEST_DISABLE_PLUGIN_AUTOLOAD="1") + env.pop("C2PA_PREFLIGHT_RUN", None) + env.pop("C2PA_FRAGMENT_OFFSET_REQUIRED", None) + if lane != "stock": + env["C2PA_PREFLIGHT_RUN"] = "1" + if lane == "candidate": + env["C2PA_FRAGMENT_OFFSET_REQUIRED"] = "1" + results[label] = command([sys.executable, str(Path(__file__).resolve()), "suite", + str(source), str(binary), sha, version, test_file, label], + source, label + ".log", env, test=True) + reports[label] = cases(EVIDENCE / (label + ".xml")) + if (results[label] == 0) != all(v[0] in ("passed", "skipped") for v in reports[label].values()): + raise RuntimeError(f"Process/JUnit status disagreement: {label}") + for file in SUITES: + suffix = "-" + file.removesuffix(".py") + problems.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix])) + focused = reports["candidate-test_fragment_at_offset"] + if len(focused) != 18 or any(v[0] != "passed" for v in focused.values()): + problems.append("Focused candidate offsets require exactly 18 passes, no skips") + # Stock compatibility may skip unsupported native offsets, but no failures. + if any(status for label, status in results.items() if label.startswith("stock-")): + problems.append("Stock 0.91 compatibility failed") + failures = {label: {key: value for key, value in report.items() if value[0] != "passed"} + for label, report in reports.items()} + record("comparison.json", {"exit_codes": results, "nonpasses": failures, "regressions": problems}) + full = all(status == 0 for label, status in results.items() if label.startswith("candidate-")) + summary = (f"## Fragment offset qualification\n\n" + f"Candidate full-suite status: **{'PASS' if full else 'FAIL'}**\n\n" + f"No-new-regressions + stock + mandatory focused gate: **{'FAIL' if problems else 'PASS'}**\n\n" + "Full-suite failures are not waived as full passes. Exact IDs, failure types and " + "normalized messages (including unchanged baseline failures) are in comparison.json; " + "raw JUnit, logs and loaded-library evidence are retained.\n\n" + + "\n".join(f"- {label}: exit {status}" for label, status in results.items()) + "\n\n" + + "\n".join(f"- `{label} / {key}`: {value!r}" + for label, report in failures.items() for key, value in report.items() + if value[0] != "skipped") + "\n") + (EVIDENCE / "summary.md").write_text(summary, encoding="utf-8") + with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as file: + file.write(summary) + print(summary) + return int(bool(problems)) + + +def main(): + EVIDENCE.mkdir(exist_ok=True) + if sys.argv[1] == "build": + build() + return 0 + if sys.argv[1] == "qualify": + return qualify() + if sys.argv[1] == "suite": + source, library, sha, version, test_file, label = sys.argv[2:] + return suite(Path(source), Path(library), sha, version, test_file, label) + raise ValueError("Unknown qualification mode") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py new file mode 100644 index 00000000..68388669 --- /dev/null +++ b/scripts/test_fragment_offset_qualification.py @@ -0,0 +1,157 @@ +"""Pure stdlib tests: never import c2pa or execute Cargo.""" + +import json +import os +from pathlib import Path +import subprocess +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +import fragment_offset_qualification as q + + +class QualificationTests(unittest.TestCase): + def test_compare_requires_every_common_test(self): + self.assertTrue(q.compare({"lost": ("passed", "", "")}, {})) + + def test_compare_accepts_identical_baseline_failure_and_fixes(self): + failure = ("failure", "AssertionError", "expected output") + self.assertEqual(q.compare({"a": failure}, {"a": failure}), []) + self.assertEqual(q.compare({"a": failure}, {"a": ("passed", "", "")}), []) + + def test_compare_rejects_new_or_changed_failure_type_message_and_skip(self): + failure = ("failure", "AssertionError", "expected output") + for outcome in (("failure", "ValueError", failure[2]), + ("failure", failure[1], "different output"), + ("skipped", "", "unsupported")): + with self.subTest(outcome=outcome): + self.assertTrue(q.compare({"a": failure}, {"a": outcome})) + self.assertTrue(q.compare({}, {"new": outcome})) + + def test_normalize_only_paths_and_addresses(self): + with patch.object(q, "ROOT", Path("/work")): + self.assertEqual(q.normalize("/work/baseline/test.py 0x123abc vs 42"), + q.normalize("/work/candidate/test.py 0x456def vs 42")) + self.assertNotEqual(q.normalize("expected 41"), q.normalize("expected 42")) + with patch.object(q.tempfile, "gettempdir", return_value="/temp"): + self.assertEqual(q.normalize("/temp/tmp12_abcd3/file.jpg"), "/file.jpg") + self.assertNotEqual(q.normalize("/temp/real-file.jpg"), "") + + def test_command_propagates_errors_and_only_defers_test_failures(self): + with tempfile.TemporaryDirectory() as directory, patch.object(q, "EVIDENCE", Path(directory)): + with patch.object(q.subprocess, "run", return_value=subprocess.CompletedProcess([], 1)) as run: + with self.assertRaises(subprocess.CalledProcessError): + q.command(["fake"], directory, "build.log") + self.assertEqual(q.command(["fake"], directory, "test.log", test=True), 1) + self.assertEqual(run.call_args.kwargs["cwd"], directory) + with patch.object(q.subprocess, "run", return_value=subprocess.CompletedProcess([], 2)): + with self.assertRaises(subprocess.CalledProcessError): + q.command(["fake"], directory, "test.log", test=True) + + def test_no_local_native_build(self): + with patch.dict(os.environ, {}, clear=True), patch.object(q, "git") as git: + with self.assertRaises(RuntimeError): + q.validate_native(Path("unused")) + git.assert_not_called() + + def test_build_cannot_execute_tools_before_validation(self): + with patch.object(q, "validate_native", side_effect=ValueError("refused")): + with patch.object(q, "command") as command: + with self.assertRaises(ValueError): + q.build() + command.assert_not_called() + + def test_native_allowlist_before_source_execution(self): + env = {"GITHUB_ACTIONS": "true", "GITHUB_REPOSITORY": "mstattma/c2pa-python"} + for origin, sha in (("https://github.com/other/c2pa-rs", q.NATIVE_SHA), + (f"https://github.com/{q.NATIVE_REPOSITORY}", "f" * 40)): + with self.subTest(origin=origin, sha=sha), patch.dict(os.environ, env): + with patch.object(q, "git", side_effect=[origin, sha]): + with self.assertRaises(ValueError): + q.validate_native(Path("unused")) + + def test_junit_ids_types_messages_and_empty_or_duplicate_rejection(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "report.xml" + xml = '' + path.write_text(f"{xml}", encoding="utf-8") + self.assertEqual(q.cases(path), {"tests.T::test_x": ("failure", "AssertionError", "bad 0xADDR")}) + for body in ("", xml + xml): + path.write_text(f"{body}", encoding="utf-8") + with self.assertRaises(ValueError): + q.cases(path) + + def test_junit_requires_and_preserves_explicit_exception_type(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "report.xml" + for properties in ("", ''): + path.write_text('' + + properties + '', encoding="utf-8") + if properties: + self.assertEqual(q.cases(path)["T::test"], ("failure", "builtins.AssertionError", "bad")) + else: + with self.assertRaises(ValueError): + q.cases(path) + + def test_qualification_keeps_full_failure_distinct_and_enforces_gates(self): + for changed, skip_offset in ((False, False), (True, False), (False, True)): + with self.subTest(changed=changed, skip_offset=skip_offset), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + evidence = root / "evidence" + evidence.mkdir() + (root / "baseline").mkdir() + stock_dir = root / "candidate/artifacts/platform" + stock_dir.mkdir(parents=True) + stock = stock_dir / "native.bin" + stock.write_bytes(b"stock") + native = evidence / "native.bin" + native.write_bytes(b"candidate") + (evidence / "native.json").write_text(json.dumps({"library": str(native), "sha256": q.digest(native)}), encoding="utf-8") + seen = [] + + def fake_command(args, cwd, name, env=None, test=False): + if not test: + return 0 + label = args[-1] + seen.append((label, args, env)) + focused = label.endswith("test_fragment_at_offset") + failing = not focused and not label.startswith("stock-") + body = "" + for index in range(18 if focused else 1): + outcome = "" + if failing: + message = "changed" if changed and label.startswith("candidate-") else "baseline failure" + outcome = f'' + if skip_offset and label == "candidate-test_fragment_at_offset" and index == 0: + outcome = '' + body += f'{outcome}' + (evidence / (label + ".xml")).write_text(f"{body}", encoding="utf-8") + return int(failing) + + with patch.object(q, "ROOT", root), patch.object(q, "EVIDENCE", evidence): + with patch.object(q, "command", side_effect=fake_command): + with patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(root / "summary.md")}), patch("builtins.print"): + self.assertEqual(q.qualify(), int(changed or skip_offset)) + summary = (evidence / "summary.md").read_text(encoding="utf-8") + self.assertIn("Candidate full-suite status: **FAIL**", summary) + self.assertEqual(len(seen), 8) + for label, args, env in seen: + if label.startswith("candidate-"): + self.assertEqual(env["C2PA_FRAGMENT_OFFSET_REQUIRED"], "1") + if not label.startswith("stock-"): + self.assertEqual(env["C2PA_PREFLIGHT_RUN"], "1") + self.assertEqual(env["C2PA_LIBRARY_NAME"], str(native)) + self.assertEqual(args[5], q.digest(native)) + + def test_loaded_provenance_rejects_fallback_without_native_imports(self): + fake = SimpleNamespace(_lib=SimpleNamespace(_name="relative-fallback.so")) + package = SimpleNamespace(c2pa=fake) + with patch.dict("sys.modules", {"c2pa": package, "c2pa.c2pa": fake, "pytest": SimpleNamespace()}): + with self.assertRaisesRegex(RuntimeError, "Wrong loaded native"): + q.suite(Path("source"), Path("/expected/library"), "hash", "version", "test.py", "test") + + +if __name__ == "__main__": + unittest.main() From 4cb7f61ee913e6643523156bdf166c55609d7c69 Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 08:26:10 +0200 Subject: [PATCH 04/10] ci: scope ephemeral Cargo home to the build step --- .github/workflows/fragment-offset-qualification.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/fragment-offset-qualification.yml b/.github/workflows/fragment-offset-qualification.yml index e19a97fa..7ffaa4f3 100644 --- a/.github/workflows/fragment-offset-qualification.yml +++ b/.github/workflows/fragment-offset-qualification.yml @@ -25,7 +25,6 @@ jobs: env: CARGO_INCREMENTAL: "0" CARGO_TERM_COLOR: never - CARGO_HOME: ${{ runner.temp }}/fragment-cargo PIP_NO_CACHE_DIR: "1" PYTHONUNBUFFERED: "1" defaults: @@ -58,6 +57,8 @@ jobs: - name: Test qualification helpers (no native imports) run: python -m unittest discover -s candidate/scripts -p test_fragment_offset_qualification.py -v - name: Build exact native candidate and record evidence + env: + CARGO_HOME: ${{ runner.temp }}/fragment-cargo run: python candidate/scripts/fragment_offset_qualification.py build - name: Install Python-only test dependencies run: | From 7a4d69e57ae457dd8429ed723333552bb86b504c Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 08:29:28 +0200 Subject: [PATCH 05/10] ci: canonicalize Windows test paths and retain helper logs --- .github/workflows/fragment-offset-qualification.yml | 5 ++++- scripts/fragment_offset_qualification.py | 2 +- scripts/test_fragment_offset_qualification.py | 5 +++-- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/.github/workflows/fragment-offset-qualification.yml b/.github/workflows/fragment-offset-qualification.yml index 7ffaa4f3..0e53e5e9 100644 --- a/.github/workflows/fragment-offset-qualification.yml +++ b/.github/workflows/fragment-offset-qualification.yml @@ -55,7 +55,10 @@ jobs: with: python-version: "3.12" - name: Test qualification helpers (no native imports) - run: python -m unittest discover -s candidate/scripts -p test_fragment_offset_qualification.py -v + run: | + set -euo pipefail + mkdir -p evidence + python -m unittest discover -s candidate/scripts -p test_fragment_offset_qualification.py -v 2>&1 | tee evidence/helper-tests.log - name: Build exact native candidate and record evidence env: CARGO_HOME: ${{ runner.temp }}/fragment-cargo diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py index 64ed0cc3..b0af200c 100644 --- a/scripts/fragment_offset_qualification.py +++ b/scripts/fragment_offset_qualification.py @@ -131,7 +131,7 @@ def normalize(message): message = re.sub(r"\\+", "/", message) for folder in (ROOT / "candidate", ROOT / "baseline", ROOT): message = message.replace(folder.as_posix(), "") - temporary = re.escape(Path(tempfile.gettempdir()).as_posix()) + temporary = re.escape(Path(tempfile.gettempdir()).resolve().as_posix()) message = re.sub(temporary + r"/tmp[a-zA-Z0-9_-]{8}(?=[/.\s'\"<>:)]|$)", "", message) return re.sub(r"0x[0-9a-fA-F]+", "0xADDR", message) diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py index 68388669..6dc4faf3 100644 --- a/scripts/test_fragment_offset_qualification.py +++ b/scripts/test_fragment_offset_qualification.py @@ -36,7 +36,8 @@ def test_normalize_only_paths_and_addresses(self): q.normalize("/work/candidate/test.py 0x456def vs 42")) self.assertNotEqual(q.normalize("expected 41"), q.normalize("expected 42")) with patch.object(q.tempfile, "gettempdir", return_value="/temp"): - self.assertEqual(q.normalize("/temp/tmp12_abcd3/file.jpg"), "/file.jpg") + temporary = Path("/temp").resolve().as_posix() + self.assertEqual(q.normalize(temporary + "/tmp12_abcd3/file.jpg"), "/file.jpg") self.assertNotEqual(q.normalize("/temp/real-file.jpg"), "") def test_command_propagates_errors_and_only_defers_test_failures(self): @@ -98,7 +99,7 @@ def test_junit_requires_and_preserves_explicit_exception_type(self): def test_qualification_keeps_full_failure_distinct_and_enforces_gates(self): for changed, skip_offset in ((False, False), (True, False), (False, True)): with self.subTest(changed=changed, skip_offset=skip_offset), tempfile.TemporaryDirectory() as directory: - root = Path(directory) + root = Path(directory).resolve() evidence = root / "evidence" evidence.mkdir() (root / "baseline").mkdir() From 72e7b373a819a7b6e2ecc94e5cbbfc35b9f15acc Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 08:45:21 +0200 Subject: [PATCH 06/10] ci: report regression and stock compatibility verdicts separately --- scripts/fragment_offset_qualification.py | 19 +++++++++++++------ scripts/test_fragment_offset_qualification.py | 11 +++++++---- 2 files changed, 20 insertions(+), 10 deletions(-) diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py index b0af200c..9576f40a 100644 --- a/scripts/fragment_offset_qualification.py +++ b/scripts/fragment_offset_qualification.py @@ -182,7 +182,7 @@ def qualify(): raise RuntimeError(f"Expected one downloaded stock library, found {stock}") stock = stock[0].resolve() shutil.copy2(stock, EVIDENCE / ("stock-" + name)) - results, reports, problems = {}, {}, [] + results, reports, regressions = {}, {}, [] lanes = (("stock", candidate, stock, digest(stock), "0.91.0"), ("baseline", baseline, library, native["sha256"], NATIVE_VERSION), ("candidate", candidate, library, native["sha256"], NATIVE_VERSION)) @@ -206,20 +206,27 @@ def qualify(): raise RuntimeError(f"Process/JUnit status disagreement: {label}") for file in SUITES: suffix = "-" + file.removesuffix(".py") - problems.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix])) + regressions.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix])) + problems = list(regressions) focused = reports["candidate-test_fragment_at_offset"] - if len(focused) != 18 or any(v[0] != "passed" for v in focused.values()): + focused_pass = len(focused) == 18 and all(v[0] == "passed" for v in focused.values()) + if not focused_pass: problems.append("Focused candidate offsets require exactly 18 passes, no skips") # Stock compatibility may skip unsupported native offsets, but no failures. - if any(status for label, status in results.items() if label.startswith("stock-")): + stock_pass = all(status == 0 for label, status in results.items() if label.startswith("stock-")) + if not stock_pass: problems.append("Stock 0.91 compatibility failed") failures = {label: {key: value for key, value in report.items() if value[0] != "passed"} for label, report in reports.items()} - record("comparison.json", {"exit_codes": results, "nonpasses": failures, "regressions": problems}) + record("comparison.json", {"exit_codes": results, "nonpasses": failures, + "regressions": regressions, "qualification_failures": problems}) full = all(status == 0 for label, status in results.items() if label.startswith("candidate-")) summary = (f"## Fragment offset qualification\n\n" f"Candidate full-suite status: **{'PASS' if full else 'FAIL'}**\n\n" - f"No-new-regressions + stock + mandatory focused gate: **{'FAIL' if problems else 'PASS'}**\n\n" + f"No-new-regressions: **{'FAIL' if regressions else 'PASS'}**\n\n" + f"Stock 0.91 compatibility: **{'PASS' if stock_pass else 'FAIL'}**\n\n" + f"Mandatory 18 focused offsets (no skips): **{'PASS' if focused_pass else 'FAIL'}**\n\n" + f"Combined qualification gate: **{'FAIL' if problems else 'PASS'}**\n\n" "Full-suite failures are not waived as full passes. Exact IDs, failure types and " "normalized messages (including unchanged baseline failures) are in comparison.json; " "raw JUnit, logs and loaded-library evidence are retained.\n\n" diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py index 6dc4faf3..27e82df8 100644 --- a/scripts/test_fragment_offset_qualification.py +++ b/scripts/test_fragment_offset_qualification.py @@ -97,8 +97,9 @@ def test_junit_requires_and_preserves_explicit_exception_type(self): q.cases(path) def test_qualification_keeps_full_failure_distinct_and_enforces_gates(self): - for changed, skip_offset in ((False, False), (True, False), (False, True)): - with self.subTest(changed=changed, skip_offset=skip_offset), tempfile.TemporaryDirectory() as directory: + for changed, skip_offset, stock_fail in ((False, False, False), (True, False, False), + (False, True, False), (False, False, True)): + with self.subTest(changed=changed, skip_offset=skip_offset, stock_fail=stock_fail), tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() evidence = root / "evidence" evidence.mkdir() @@ -118,7 +119,7 @@ def fake_command(args, cwd, name, env=None, test=False): label = args[-1] seen.append((label, args, env)) focused = label.endswith("test_fragment_at_offset") - failing = not focused and not label.startswith("stock-") + failing = not focused and (not label.startswith("stock-") or stock_fail) body = "" for index in range(18 if focused else 1): outcome = "" @@ -134,9 +135,11 @@ def fake_command(args, cwd, name, env=None, test=False): with patch.object(q, "ROOT", root), patch.object(q, "EVIDENCE", evidence): with patch.object(q, "command", side_effect=fake_command): with patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(root / "summary.md")}), patch("builtins.print"): - self.assertEqual(q.qualify(), int(changed or skip_offset)) + self.assertEqual(q.qualify(), int(changed or skip_offset or stock_fail)) summary = (evidence / "summary.md").read_text(encoding="utf-8") self.assertIn("Candidate full-suite status: **FAIL**", summary) + self.assertIn(f"No-new-regressions: **{'FAIL' if changed else 'PASS'}**", summary) + self.assertIn(f"Stock 0.91 compatibility: **{'FAIL' if stock_fail else 'PASS'}**", summary) self.assertEqual(len(seen), 8) for label, args, env in seen: if label.startswith("candidate-"): From a8d85a4252b705c964b51759e5f72337d20c72ed Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 09:04:00 +0200 Subject: [PATCH 07/10] test: guard the real-fork case on supported platforms --- scripts/fragment_offset_qualification.py | 32 +++++++++++-- scripts/test_fragment_offset_qualification.py | 46 +++++++++++++++++++ tests/test_unit_tests_threaded.py | 1 + 3 files changed, 74 insertions(+), 5 deletions(-) diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py index 9576f40a..2c603065 100644 --- a/scripts/fragment_offset_qualification.py +++ b/scripts/fragment_offset_qualification.py @@ -19,6 +19,13 @@ FEATURES = "rust_native_crypto,add_thumbnails,http,file_io" RUST = "1.96.0" SUITES = ("test_unit_tests.py", "test_unit_tests_threaded.py") +FORK_TEST_ID = "tests.test_unit_tests_threaded.TestForkedChildDoesNotDeadlock::test_parent_copy_unaffected" +FORK_BASELINE_FAILURE = ( + "failure", "builtins.AssertionError", + 'AssertionError: 1 != 0 : Traceback (most recent call last):\n' + ' File "", line 11, in \n' + "AttributeError: module 'os' has no attribute 'fork'", +) ROOT = Path(__file__).resolve().parents[2] EVIDENCE = ROOT / "evidence" @@ -158,10 +165,19 @@ def cases(path): return result -def compare(baseline, candidate): +def compare(baseline, candidate, *, platform_adjustments=None): problems = [f"Missing common test: {key}" for key in baseline.keys() - candidate.keys()] for key, outcome in candidate.items(): previous = baseline.get(key, ("passed", "", "")) + # The base predates the fork guard. Exempt only its exact Windows + # missing-fork failure, never a passing test or an unrelated failure. + if (platform_adjustments is not None and sys.platform == "win32" + and not hasattr(os, "fork") and key == FORK_TEST_ID + and previous == FORK_BASELINE_FAILURE + and outcome == ("skipped", "pytest.skip", "requires os.fork")): + platform_adjustments.append({"test_id": key, "platform": sys.platform, + "baseline": previous, "candidate": outcome}) + continue if outcome[0] != "passed" and outcome != previous: problems.append(f"New/changed outcome: {key}: {outcome!r}; baseline: {previous!r}") return sorted(problems) @@ -182,7 +198,7 @@ def qualify(): raise RuntimeError(f"Expected one downloaded stock library, found {stock}") stock = stock[0].resolve() shutil.copy2(stock, EVIDENCE / ("stock-" + name)) - results, reports, regressions = {}, {}, [] + results, reports, regressions, platform_adjustments = {}, {}, [], [] lanes = (("stock", candidate, stock, digest(stock), "0.91.0"), ("baseline", baseline, library, native["sha256"], NATIVE_VERSION), ("candidate", candidate, library, native["sha256"], NATIVE_VERSION)) @@ -206,7 +222,8 @@ def qualify(): raise RuntimeError(f"Process/JUnit status disagreement: {label}") for file in SUITES: suffix = "-" + file.removesuffix(".py") - regressions.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix])) + regressions.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix], + platform_adjustments=platform_adjustments)) problems = list(regressions) focused = reports["candidate-test_fragment_at_offset"] focused_pass = len(focused) == 18 and all(v[0] == "passed" for v in focused.values()) @@ -219,7 +236,8 @@ def qualify(): failures = {label: {key: value for key, value in report.items() if value[0] != "passed"} for label, report in reports.items()} record("comparison.json", {"exit_codes": results, "nonpasses": failures, - "regressions": regressions, "qualification_failures": problems}) + "regressions": regressions, "qualification_failures": problems, + "platform_adjustments": platform_adjustments}) full = all(status == 0 for label, status in results.items() if label.startswith("candidate-")) summary = (f"## Fragment offset qualification\n\n" f"Candidate full-suite status: **{'PASS' if full else 'FAIL'}**\n\n" @@ -227,13 +245,17 @@ def qualify(): f"Stock 0.91 compatibility: **{'PASS' if stock_pass else 'FAIL'}**\n\n" f"Mandatory 18 focused offsets (no skips): **{'PASS' if focused_pass else 'FAIL'}**\n\n" f"Combined qualification gate: **{'FAIL' if problems else 'PASS'}**\n\n" + f"Explicit platform applicability adjustments: **{len(platform_adjustments)}**\n\n" "Full-suite failures are not waived as full passes. Exact IDs, failure types and " "normalized messages (including unchanged baseline failures) are in comparison.json; " "raw JUnit, logs and loaded-library evidence are retained.\n\n" + "\n".join(f"- {label}: exit {status}" for label, status in results.items()) + "\n\n" + "\n".join(f"- `{label} / {key}`: {value!r}" for label, report in failures.items() for key, value in report.items() - if value[0] != "skipped") + "\n") + if value[0] != "skipped") + "\n" + + "\n".join(f"- Platform applicability: `{entry['test_id']}` on {entry['platform']}: " + f"{entry['baseline']!r} -> {entry['candidate']!r}" + for entry in platform_adjustments) + "\n") (EVIDENCE / "summary.md").write_text(summary, encoding="utf-8") with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as file: file.write(summary) diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py index 27e82df8..959f0045 100644 --- a/scripts/test_fragment_offset_qualification.py +++ b/scripts/test_fragment_offset_qualification.py @@ -1,5 +1,6 @@ """Pure stdlib tests: never import c2pa or execute Cargo.""" +import ast import json import os from pathlib import Path @@ -30,6 +31,49 @@ def test_compare_rejects_new_or_changed_failure_type_message_and_skip(self): self.assertTrue(q.compare({"a": failure}, {"a": outcome})) self.assertTrue(q.compare({}, {"new": outcome})) + def test_windows_fork_applicability_is_exact_and_recorded(self): + skipped = ("skipped", "pytest.skip", "requires os.fork") + adjustments = [] + with patch.object(q, "sys", SimpleNamespace(platform="win32")), patch.object(q, "os", SimpleNamespace()): + self.assertTrue(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE}, {q.FORK_TEST_ID: skipped})) + self.assertEqual(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE}, + {q.FORK_TEST_ID: skipped}, platform_adjustments=adjustments), []) + self.assertEqual(adjustments, [{"test_id": q.FORK_TEST_ID, "platform": "win32", + "baseline": q.FORK_BASELINE_FAILURE, "candidate": skipped}]) + + def test_fork_applicability_does_not_relax_other_comparisons(self): + skipped = ("skipped", "pytest.skip", "requires os.fork") + for platform, os_module, key, previous, outcome in ( + ("linux", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, skipped), + ("win32", SimpleNamespace(fork=object()), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, skipped), + ("win32", SimpleNamespace(), "other::test", q.FORK_BASELINE_FAILURE, skipped), + ("win32", SimpleNamespace(), q.FORK_TEST_ID, ("passed", "", ""), skipped), + ("win32", SimpleNamespace(), q.FORK_TEST_ID, ("failure", "builtins.AssertionError", "different failure"), skipped), + ("win32", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, ("skipped", "pytest.skip", "other reason")), + ("win32", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, ("skipped", "other.type", "requires os.fork"))): + with self.subTest(platform=platform, key=key, previous=previous, outcome=outcome): + adjustments = [] + with patch.object(q, "sys", SimpleNamespace(platform=platform)), patch.object(q, "os", os_module): + self.assertTrue(q.compare({key: previous}, {key: outcome}, platform_adjustments=adjustments)) + self.assertEqual(adjustments, []) + + def test_fork_guard_is_per_test_and_preserves_unix_applicability(self): + source = Path(q.__file__).resolve().parents[1] / "tests/test_unit_tests_threaded.py" + tree = ast.parse(source.read_text(encoding="utf-8")) + cls = next(node for node in tree.body if isinstance(node, ast.ClassDef) + and node.name == "TestForkedChildDoesNotDeadlock") + self.assertEqual(cls.decorator_list, []) + method = next(node for node in cls.body if isinstance(node, ast.FunctionDef) + and node.name == "test_parent_copy_unaffected") + module = ast.Module(body=[method], type_ignores=[]) + for supports_fork in (False, True): + namespace = {"unittest": unittest, "os": SimpleNamespace(**({"fork": object()} if supports_fork else {}))} + exec(compile(module, str(source), "exec"), namespace) + function = namespace[method.name] + self.assertEqual(getattr(function, "__unittest_skip__", False), not supports_fork) + if not supports_fork: + self.assertEqual(function.__unittest_skip_why__, "requires os.fork") + def test_normalize_only_paths_and_addresses(self): with patch.object(q, "ROOT", Path("/work")): self.assertEqual(q.normalize("/work/baseline/test.py 0x123abc vs 42"), @@ -137,6 +181,8 @@ def fake_command(args, cwd, name, env=None, test=False): with patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(root / "summary.md")}), patch("builtins.print"): self.assertEqual(q.qualify(), int(changed or skip_offset or stock_fail)) summary = (evidence / "summary.md").read_text(encoding="utf-8") + comparison = json.loads((evidence / "comparison.json").read_text(encoding="utf-8")) + self.assertEqual(comparison["platform_adjustments"], []) self.assertIn("Candidate full-suite status: **FAIL**", summary) self.assertIn(f"No-new-regressions: **{'FAIL' if changed else 'PASS'}**", summary) self.assertIn(f"Stock 0.91 compatibility: **{'FAIL' if stock_fail else 'PASS'}**", summary) diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 2722ea59..20faf67e 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -315,6 +315,7 @@ def test_is_valid_false_for_inherited_object(self): reader = self._foreign_reader_with_lock_held() self.assertFalse(reader.is_valid) + @unittest.skipUnless(hasattr(os, "fork"), "requires os.fork") def test_parent_copy_unaffected(self): """The child closing its copy must leave the parent's usable. From 68d838f2d0da8b62cf124da1881eba5ef2346e9b Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 09:51:35 +0200 Subject: [PATCH 08/10] ci: pin focused test identities and compare stock baselines --- scripts/fragment_offset_qualification.py | 94 ++++++++--- scripts/test_fragment_offset_qualification.py | 146 +++++++++++++++--- 2 files changed, 198 insertions(+), 42 deletions(-) diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py index 2c603065..23489da0 100644 --- a/scripts/fragment_offset_qualification.py +++ b/scripts/fragment_offset_qualification.py @@ -19,6 +19,31 @@ FEATURES = "rust_native_crypto,add_thumbnails,http,file_io" RUST = "1.96.0" SUITES = ("test_unit_tests.py", "test_unit_tests_threaded.py") +# Reviewed raw JUnit keys: future passing tests may extend, not replace, these IDs. +BINDING_OFFSET_TEST_IDS = frozenset({ + "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_exported_from_package", + "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_false_without_symbols", + "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_probe_result_is_used_and_cached", + "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_unsupported_raises_before_touching_the_reader", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_from_rejects_invalid_context", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_aliased_streams_before_allocation", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_format_encoding_before_allocation", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_format_types_before_allocation", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_streams_before_allocation", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_non_int_offsets", + "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_out_of_range_offsets", +}) +NATIVE_OFFSET_TEST_IDS = frozenset({ + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_from_fails_cleanly_without_a_manifest", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_from_fragment_at_offset_zero", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_maximum_offset_never_verifies", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_native_failure_closes_the_reader", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_replacing_the_fragment_refreshes_results", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_wrong_offset_is_a_mismatch_not_an_error", + "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_zero_offset_matches_with_fragment", +}) +REQUIRED_OFFSET_TEST_IDS = BINDING_OFFSET_TEST_IDS | NATIVE_OFFSET_TEST_IDS +STOCK_NATIVE_SKIP = ("skipped", "pytest.skip", "native library lacks c2pa_reader_with_fragment_at_offset") FORK_TEST_ID = "tests.test_unit_tests_threaded.TestForkedChildDoesNotDeadlock::test_parent_copy_unaffected" FORK_BASELINE_FAILURE = ( "failure", "builtins.AssertionError", @@ -165,17 +190,26 @@ def cases(path): return result -def compare(baseline, candidate, *, platform_adjustments=None): +def focused_problems(report, *, stock=False): + problems = [f"Missing required focused test: {key}" for key in REQUIRED_OFFSET_TEST_IDS - report.keys()] + for key, outcome in report.items(): + expected = STOCK_NATIVE_SKIP if stock and key in NATIVE_OFFSET_TEST_IDS else ("passed", "", "") + if outcome != expected: + problems.append(f"Unexpected focused outcome: {key}: {outcome!r}; expected: {expected!r}") + return sorted(problems) + + +def compare(baseline, candidate, *, platform_adjustments=None, context=None): problems = [f"Missing common test: {key}" for key in baseline.keys() - candidate.keys()] for key, outcome in candidate.items(): previous = baseline.get(key, ("passed", "", "")) # The base predates the fork guard. Exempt only its exact Windows # missing-fork failure, never a passing test or an unrelated failure. - if (platform_adjustments is not None and sys.platform == "win32" + if (platform_adjustments is not None and context in ("stock", "native") and sys.platform == "win32" and not hasattr(os, "fork") and key == FORK_TEST_ID and previous == FORK_BASELINE_FAILURE and outcome == ("skipped", "pytest.skip", "requires os.fork")): - platform_adjustments.append({"test_id": key, "platform": sys.platform, + platform_adjustments.append({"test_id": key, "platform": sys.platform, "context": context, "baseline": previous, "candidate": outcome}) continue if outcome[0] != "passed" and outcome != previous: @@ -199,18 +233,19 @@ def qualify(): stock = stock[0].resolve() shutil.copy2(stock, EVIDENCE / ("stock-" + name)) results, reports, regressions, platform_adjustments = {}, {}, [], [] - lanes = (("stock", candidate, stock, digest(stock), "0.91.0"), + lanes = (("stock-baseline", baseline, stock, digest(stock), "0.91.0"), + ("stock", candidate, stock, digest(stock), "0.91.0"), ("baseline", baseline, library, native["sha256"], NATIVE_VERSION), ("candidate", candidate, library, native["sha256"], NATIVE_VERSION)) for lane, source, binary, sha, version in lanes: - files = SUITES + (("test_fragment_at_offset.py",) if lane != "baseline" else ()) + files = SUITES + (("test_fragment_at_offset.py",) if lane in ("stock", "candidate") else ()) for test_file in files: label = lane + "-" + test_file.removesuffix(".py") env = dict(os.environ, PYTHONPATH=str(source / "src"), C2PA_LIBRARY_NAME=str(binary), PYTEST_DISABLE_PLUGIN_AUTOLOAD="1") env.pop("C2PA_PREFLIGHT_RUN", None) env.pop("C2PA_FRAGMENT_OFFSET_REQUIRED", None) - if lane != "stock": + if lane in ("baseline", "candidate"): env["C2PA_PREFLIGHT_RUN"] = "1" if lane == "candidate": env["C2PA_FRAGMENT_OFFSET_REQUIRED"] = "1" @@ -220,30 +255,53 @@ def qualify(): reports[label] = cases(EVIDENCE / (label + ".xml")) if (results[label] == 0) != all(v[0] in ("passed", "skipped") for v in reports[label].values()): raise RuntimeError(f"Process/JUnit status disagreement: {label}") - for file in SUITES: - suffix = "-" + file.removesuffix(".py") - regressions.extend(compare(reports["baseline" + suffix], reports["candidate" + suffix], - platform_adjustments=platform_adjustments)) + comparisons = {} + for context, base_lane, pr_lane in (("stock", "stock-baseline", "stock"), ("native", "baseline", "candidate")): + comparisons[context] = [] + for file in SUITES: + suffix = "-" + file.removesuffix(".py") + comparisons[context].extend(compare(reports[base_lane + suffix], reports[pr_lane + suffix], + platform_adjustments=platform_adjustments, context=context)) + regressions.extend(f"{context}: {problem}" for problem in comparisons[context]) problems = list(regressions) - focused = reports["candidate-test_fragment_at_offset"] - focused_pass = len(focused) == 18 and all(v[0] == "passed" for v in focused.values()) - if not focused_pass: - problems.append("Focused candidate offsets require exactly 18 passes, no skips") - # Stock compatibility may skip unsupported native offsets, but no failures. - stock_pass = all(status == 0 for label, status in results.items() if label.startswith("stock-")) + focused = {lane: focused_problems(reports[lane + "-test_fragment_at_offset"], stock=lane == "stock") + for lane in ("stock", "candidate")} + focused_pass = not focused["candidate"] + problems.extend(f"{lane} focused: {problem}" for lane, issues in focused.items() for problem in issues) + # The unmodified stock base may fail only the explicitly recognized fork case. + allowed_base_failures = {entry["test_id"]: entry["baseline"] for entry in platform_adjustments + if entry["context"] == "stock"} + stock_base_problems = [] + for file in SUITES: + label = "stock-baseline-" + file.removesuffix(".py") + stock_base_problems.extend(f"Unexpected stock baseline failure: {key}: {outcome!r}" + for key, outcome in reports[label].items() + if outcome[0] in ("failure", "error") and allowed_base_failures.get(key) != outcome) + problems.extend(stock_base_problems) + stock_labels = ["stock-" + file.removesuffix(".py") for file in (*SUITES, "test_fragment_at_offset.py")] + stock_pass = (all(results[label] == 0 for label in stock_labels) + and not comparisons["stock"] and not focused["stock"] and not stock_base_problems) if not stock_pass: problems.append("Stock 0.91 compatibility failed") failures = {label: {key: value for key, value in report.items() if value[0] != "passed"} for label, report in reports.items()} record("comparison.json", {"exit_codes": results, "nonpasses": failures, "regressions": regressions, "qualification_failures": problems, + "comparisons": comparisons, "focused_problems": focused, + "focused_requirements": {"required_ids": sorted(REQUIRED_OFFSET_TEST_IDS), + "stock_native_skip_ids": sorted(NATIVE_OFFSET_TEST_IDS), + "stock_native_skip": STOCK_NATIVE_SKIP}, + "lanes": {lane: {"python_source": str(source), "library": str(binary), + "sha256": sha, "version": version} + for lane, source, binary, sha, version in lanes}, "platform_adjustments": platform_adjustments}) full = all(status == 0 for label, status in results.items() if label.startswith("candidate-")) summary = (f"## Fragment offset qualification\n\n" f"Candidate full-suite status: **{'PASS' if full else 'FAIL'}**\n\n" f"No-new-regressions: **{'FAIL' if regressions else 'PASS'}**\n\n" f"Stock 0.91 compatibility: **{'PASS' if stock_pass else 'FAIL'}**\n\n" - f"Mandatory 18 focused offsets (no skips): **{'PASS' if focused_pass else 'FAIL'}**\n\n" + f"Required 18 focused offset IDs (all pass, no skips): **{'PASS' if focused_pass else 'FAIL'}**\n\n" + f"Stock focused IDs (11 pass, 7 exact unsupported skips): **{'FAIL' if focused['stock'] else 'PASS'}**\n\n" f"Combined qualification gate: **{'FAIL' if problems else 'PASS'}**\n\n" f"Explicit platform applicability adjustments: **{len(platform_adjustments)}**\n\n" "Full-suite failures are not waived as full passes. Exact IDs, failure types and " @@ -253,7 +311,7 @@ def qualify(): + "\n".join(f"- `{label} / {key}`: {value!r}" for label, report in failures.items() for key, value in report.items() if value[0] != "skipped") + "\n" - + "\n".join(f"- Platform applicability: `{entry['test_id']}` on {entry['platform']}: " + + "\n".join(f"- Platform applicability ({entry['context']}): `{entry['test_id']}` on {entry['platform']}: " f"{entry['baseline']!r} -> {entry['candidate']!r}" for entry in platform_adjustments) + "\n") (EVIDENCE / "summary.md").write_text(summary, encoding="utf-8") diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py index 959f0045..c65e8ac8 100644 --- a/scripts/test_fragment_offset_qualification.py +++ b/scripts/test_fragment_offset_qualification.py @@ -9,6 +9,7 @@ from types import SimpleNamespace import unittest from unittest.mock import patch +import xml.etree.ElementTree as ET import fragment_offset_qualification as q @@ -37,9 +38,9 @@ def test_windows_fork_applicability_is_exact_and_recorded(self): with patch.object(q, "sys", SimpleNamespace(platform="win32")), patch.object(q, "os", SimpleNamespace()): self.assertTrue(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE}, {q.FORK_TEST_ID: skipped})) self.assertEqual(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE}, - {q.FORK_TEST_ID: skipped}, platform_adjustments=adjustments), []) + {q.FORK_TEST_ID: skipped}, platform_adjustments=adjustments, context="native"), []) self.assertEqual(adjustments, [{"test_id": q.FORK_TEST_ID, "platform": "win32", - "baseline": q.FORK_BASELINE_FAILURE, "candidate": skipped}]) + "context": "native", "baseline": q.FORK_BASELINE_FAILURE, "candidate": skipped}]) def test_fork_applicability_does_not_relax_other_comparisons(self): skipped = ("skipped", "pytest.skip", "requires os.fork") @@ -54,9 +55,57 @@ def test_fork_applicability_does_not_relax_other_comparisons(self): with self.subTest(platform=platform, key=key, previous=previous, outcome=outcome): adjustments = [] with patch.object(q, "sys", SimpleNamespace(platform=platform)), patch.object(q, "os", os_module): - self.assertTrue(q.compare({key: previous}, {key: outcome}, platform_adjustments=adjustments)) + self.assertTrue(q.compare({key: previous}, {key: outcome}, platform_adjustments=adjustments, context="native")) self.assertEqual(adjustments, []) + def test_required_focused_ids_match_source_without_native_imports(self): + source = Path(q.__file__).resolve().parents[1] / "tests/test_fragment_at_offset.py" + tree = ast.parse(source.read_text(encoding="utf-8")) + ids = {f"tests.test_fragment_at_offset.{cls.name}::{method.name}" + for cls in tree.body if isinstance(cls, ast.ClassDef) + for method in cls.body if isinstance(method, ast.FunctionDef) and method.name.startswith("test_")} + self.assertEqual(len(q.REQUIRED_OFFSET_TEST_IDS), 18) + self.assertEqual(len(q.BINDING_OFFSET_TEST_IDS), 11) + self.assertEqual(len(q.NATIVE_OFFSET_TEST_IDS), 7) + self.assertLessEqual(q.REQUIRED_OFFSET_TEST_IDS, ids) + + def test_focused_required_ids_cannot_be_replaced_at_constant_count(self): + report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", "")) + removed = sorted(q.REQUIRED_OFFSET_TEST_IDS)[0] + del report[removed] + report["unrelated.T::test_other"] = ("passed", "", "") + self.assertEqual(len(report), 18) + self.assertEqual(q.focused_problems(report), [f"Missing required focused test: {removed}"]) + + def test_focused_allows_extra_passes_but_no_extra_failure_or_skip(self): + report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", "")) + self.assertEqual(q.focused_problems(report), []) + report["future.T::test_extra"] = ("passed", "", "") + self.assertEqual(q.focused_problems(report), []) + for key in (sorted(q.REQUIRED_OFFSET_TEST_IDS)[0], "future.T::test_extra"): + for outcome in (("failure", "AssertionError", "bad"), ("skipped", "pytest.skip", "unsupported")): + with self.subTest(key=key, outcome=outcome): + self.assertTrue(q.focused_problems({**report, key: outcome})) + + def test_stock_focused_requires_binding_passes_and_exact_native_skips(self): + report = dict.fromkeys(q.BINDING_OFFSET_TEST_IDS, ("passed", "", "")) + report.update(dict.fromkeys(q.NATIVE_OFFSET_TEST_IDS, q.STOCK_NATIVE_SKIP)) + self.assertEqual(q.focused_problems(report, stock=True), []) + self.assertEqual(q.focused_problems({**report, "future.T::test_extra": ("passed", "", "")}, stock=True), []) + native_id = sorted(q.NATIVE_OFFSET_TEST_IDS)[0] + binding_id = sorted(q.BINDING_OFFSET_TEST_IDS)[0] + for key, outcome in ((binding_id, q.STOCK_NATIVE_SKIP), + (native_id, ("skipped", "pytest.skip", "wrong reason")), + (native_id, ("skipped", "other.type", q.STOCK_NATIVE_SKIP[2])), + (native_id, ("passed", "", "")), + ("future.T::test_extra", q.STOCK_NATIVE_SKIP)): + with self.subTest(key=key, outcome=outcome): + self.assertTrue(q.focused_problems({**report, key: outcome}, stock=True)) + del report[native_id] + report["unrelated.T::test_other"] = ("passed", "", "") + self.assertEqual(len(report), 18) + self.assertTrue(q.focused_problems(report, stock=True)) + def test_fork_guard_is_per_test_and_preserves_unix_applicability(self): source = Path(q.__file__).resolve().parents[1] / "tests/test_unit_tests_threaded.py" tree = ast.parse(source.read_text(encoding="utf-8")) @@ -141,9 +190,10 @@ def test_junit_requires_and_preserves_explicit_exception_type(self): q.cases(path) def test_qualification_keeps_full_failure_distinct_and_enforces_gates(self): - for changed, skip_offset, stock_fail in ((False, False, False), (True, False, False), - (False, True, False), (False, False, True)): - with self.subTest(changed=changed, skip_offset=skip_offset, stock_fail=stock_fail), tempfile.TemporaryDirectory() as directory: + for scenario in ("unchanged", "native-failure", "native-offset-skip", "stock-failure", "stock-new-skip", "stock-missing-common", + "replace-required-id", "stock-binding-skip", "stock-native-wrong-reason", + "stock-baseline-unexpected-failure", "windows-fork"): + with self.subTest(scenario=scenario), tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() evidence = root / "evidence" evidence.mkdir() @@ -163,37 +213,85 @@ def fake_command(args, cwd, name, env=None, test=False): label = args[-1] seen.append((label, args, env)) focused = label.endswith("test_fragment_at_offset") - failing = not focused and (not label.startswith("stock-") or stock_fail) - body = "" - for index in range(18 if focused else 1): - outcome = "" - if failing: - message = "changed" if changed and label.startswith("candidate-") else "baseline failure" - outcome = f'' - if skip_offset and label == "candidate-test_fragment_at_offset" and index == 0: - outcome = '' - body += f'{outcome}' - (evidence / (label + ".xml")).write_text(f"{body}", encoding="utf-8") - return int(failing) + if focused: + report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", "")) + if label.startswith("stock-"): + report.update(dict.fromkeys(q.NATIVE_OFFSET_TEST_IDS, q.STOCK_NATIVE_SKIP)) + native_id = sorted(q.NATIVE_OFFSET_TEST_IDS)[0] + if scenario == "native-offset-skip" and label.startswith("candidate-"): + report[native_id] = q.STOCK_NATIVE_SKIP + if scenario == "replace-required-id" and label.startswith("candidate-"): + del report[native_id] + report["unrelated.T::test_extra"] = ("passed", "", "") + self.assertEqual(len(report), 18) + if scenario == "stock-binding-skip" and label.startswith("stock-"): + report[sorted(q.BINDING_OFFSET_TEST_IDS)[0]] = q.STOCK_NATIVE_SKIP + if scenario == "stock-native-wrong-reason" and label.startswith("stock-"): + report[native_id] = ("skipped", "pytest.skip", "wrong reason") + else: + report = {"T::test_common": ("passed", "", "")} + if label.startswith(("baseline-", "candidate-")): + message = "changed" if scenario == "native-failure" and label.startswith("candidate-") else "baseline failure" + report["T::test_common"] = ("failure", "AssertionError", message) + if label == "stock-test_unit_tests": + if scenario == "stock-failure": + report["T::test_common"] = ("failure", "AssertionError", "stock failure") + if scenario == "stock-new-skip": + report["T::test_common"] = ("skipped", "pytest.skip", "hidden regression") + if scenario == "stock-missing-common": + report = {"T::test_unrelated": ("passed", "", "")} + if scenario == "stock-baseline-unexpected-failure" and label == "stock-baseline-test_unit_tests": + report["T::test_common"] = ("failure", "AssertionError", "unexpected baseline failure") + if scenario == "windows-fork" and label.endswith("test_unit_tests_threaded"): + report[q.FORK_TEST_ID] = (q.FORK_BASELINE_FAILURE if label.startswith(("baseline-", "stock-baseline-")) + else ("skipped", "pytest.skip", "requires os.fork")) + xml = ET.Element("testsuite") + for key, (kind, failure_type, message) in report.items(): + classname, name = key.split("::", 1) + case = ET.SubElement(xml, "testcase", classname=classname, name=name) + if kind != "passed": + ET.SubElement(case, kind, type=failure_type, message=message) + ET.ElementTree(xml).write(evidence / (label + ".xml"), encoding="utf-8") + return int(any(outcome[0] in ("failure", "error") for outcome in report.values())) with patch.object(q, "ROOT", root), patch.object(q, "EVIDENCE", evidence): with patch.object(q, "command", side_effect=fake_command): with patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(root / "summary.md")}), patch("builtins.print"): - self.assertEqual(q.qualify(), int(changed or skip_offset or stock_fail)) + sys_module = SimpleNamespace(platform="win32" if scenario == "windows-fork" else "linux", executable=q.sys.executable) + with patch.object(q, "sys", sys_module), patch.object(q, "os", SimpleNamespace(environ=os.environ)): + self.assertEqual(q.qualify(), int(scenario not in ("unchanged", "windows-fork"))) summary = (evidence / "summary.md").read_text(encoding="utf-8") comparison = json.loads((evidence / "comparison.json").read_text(encoding="utf-8")) - self.assertEqual(comparison["platform_adjustments"], []) + if scenario == "windows-fork": + self.assertEqual({entry["context"] for entry in comparison["platform_adjustments"]}, {"stock", "native"}) + self.assertEqual(len(comparison["platform_adjustments"]), 2) + self.assertEqual(comparison["exit_codes"]["stock-baseline-test_unit_tests_threaded"], 1) + self.assertEqual(comparison["exit_codes"]["stock-test_unit_tests_threaded"], 0) + else: + self.assertEqual(comparison["platform_adjustments"], []) + if scenario in ("stock-new-skip", "stock-missing-common"): + self.assertEqual(comparison["exit_codes"]["stock-test_unit_tests"], 0) + self.assertTrue(comparison["comparisons"]["stock"]) self.assertIn("Candidate full-suite status: **FAIL**", summary) - self.assertIn(f"No-new-regressions: **{'FAIL' if changed else 'PASS'}**", summary) - self.assertIn(f"Stock 0.91 compatibility: **{'FAIL' if stock_fail else 'PASS'}**", summary) - self.assertEqual(len(seen), 8) + self.assertIn(f"No-new-regressions: **{'FAIL' if scenario in ('native-failure', 'stock-failure', 'stock-new-skip', 'stock-missing-common') else 'PASS'}**", summary) + self.assertIn(f"Stock 0.91 compatibility: **{'FAIL' if scenario.startswith('stock-') else 'PASS'}**", summary) + self.assertEqual(len(seen), 10) + self.assertEqual(set(comparison["lanes"]), {"stock-baseline", "stock", "baseline", "candidate"}) for label, args, env in seen: if label.startswith("candidate-"): self.assertEqual(env["C2PA_FRAGMENT_OFFSET_REQUIRED"], "1") - if not label.startswith("stock-"): + if label.startswith("stock-"): + self.assertNotIn("C2PA_PREFLIGHT_RUN", env) + self.assertNotIn("C2PA_FRAGMENT_OFFSET_REQUIRED", env) + self.assertEqual(env["C2PA_LIBRARY_NAME"], str(stock)) + self.assertEqual(args[5], q.digest(stock)) + else: self.assertEqual(env["C2PA_PREFLIGHT_RUN"], "1") self.assertEqual(env["C2PA_LIBRARY_NAME"], str(native)) self.assertEqual(args[5], q.digest(native)) + source = root / ("baseline" if label.startswith(("stock-baseline-", "baseline-")) else "candidate") + self.assertEqual(args[3], str(source)) + self.assertEqual(env["PYTHONPATH"], str(source / "src")) def test_loaded_provenance_rejects_fallback_without_native_imports(self): fake = SimpleNamespace(_lib=SimpleNamespace(_name="relative-fallback.so")) From ebf6fccf5dd9daaad9eb31453275adbfb82facfa Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 10:18:56 +0200 Subject: [PATCH 09/10] test: assert builder semantics across created and gathered ordering --- tests/test_unit_tests.py | 138 +++++++++++++++++++-------------------- 1 file changed, 69 insertions(+), 69 deletions(-) diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index dd3c3ec3..349292ea 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -3745,58 +3745,57 @@ def test_builder_sign_with_ingredients_edit_intent(self): self.assertIsInstance(ingredients_manifest, list) self.assertEqual(len(ingredients_manifest), 2, "Should have exactly two ingredients") - # Verify the first ingredient is the one we added manually with componentOf relationship - first_ingredient = ingredients_manifest[0] - self.assertEqual(first_ingredient["title"], "Test Ingredient") - self.assertEqual(first_ingredient["format"], "image/jpeg") - self.assertIn("instance_id", first_ingredient) - self.assertIn("thumbnail", first_ingredient) - self.assertEqual(first_ingredient["thumbnail"]["format"], "image/jpeg") - self.assertIn("identifier", first_ingredient["thumbnail"]) - self.assertEqual(first_ingredient["relationship"], "componentOf") - self.assertIn("label", first_ingredient) - - # Verify the second ingredient is the auto-created parent with parentOf relationship - second_ingredient = ingredients_manifest[1] + # Created/gathered buckets deliberately reorder ingredients (c2pa-rs #2720, 309b99abe). + self.assertCountEqual( + [ingredient["relationship"] for ingredient in ingredients_manifest], + ["parentOf", "componentOf"], + ) + component_ingredient = next( + ingredient for ingredient in ingredients_manifest + if ingredient["relationship"] == "componentOf" + ) + parent_ingredient = next( + ingredient for ingredient in ingredients_manifest + if ingredient["relationship"] == "parentOf" + ) + self.assertEqual(component_ingredient["title"], "Test Ingredient") + self.assertEqual(component_ingredient["format"], "image/jpeg") + self.assertIn("instance_id", component_ingredient) + self.assertIn("thumbnail", component_ingredient) + self.assertEqual(component_ingredient["thumbnail"]["format"], "image/jpeg") + self.assertIn("identifier", component_ingredient["thumbnail"]) + self.assertEqual(component_ingredient["relationship"], "componentOf") + self.assertIn("label", component_ingredient) + # Parent ingredient may not have a title field, or may have an empty one - self.assertEqual(second_ingredient["format"], "image/jpeg") - self.assertIn("instance_id", second_ingredient) - self.assertIn("thumbnail", second_ingredient) - self.assertEqual(second_ingredient["thumbnail"]["format"], "image/jpeg") - self.assertIn("identifier", second_ingredient["thumbnail"]) - self.assertEqual(second_ingredient["relationship"], "parentOf") - self.assertIn("label", second_ingredient) - - # Count ingredients with parentOf relationship - should be exactly one - parent_ingredients = [ - ing for ing in ingredients_manifest - if ing.get("relationship") == "parentOf" - ] - self.assertEqual(len(parent_ingredients), 1, "Should have exactly one parentOf ingredient") + self.assertEqual(parent_ingredient["format"], "image/jpeg") + self.assertIn("instance_id", parent_ingredient) + self.assertIn("thumbnail", parent_ingredient) + self.assertEqual(parent_ingredient["thumbnail"]["format"], "image/jpeg") + self.assertIn("identifier", parent_ingredient["thumbnail"]) + self.assertEqual(parent_ingredient["relationship"], "parentOf") + self.assertIn("label", parent_ingredient) # Check that assertions exist self.assertIn("assertions", active_manifest) assertions = active_manifest["assertions"] - # Find the actions assertion - actions_assertion = None - for assertion in assertions: - if assertion["label"] in ["c2pa.actions", "c2pa.actions.v2"]: - actions_assertion = assertion - break - - self.assertIsNotNone(actions_assertion, "Should have c2pa.actions assertion") + actions_assertions = [ + assertion for assertion in assertions + if assertion["label"].startswith("c2pa.actions") + ] + self.assertTrue(actions_assertions, "Should have c2pa.actions assertion") # Verify exactly one c2pa.opened action exists for EDIT intent - actions = actions_assertion["data"]["actions"] opened_actions = [ - action for action in actions + action for assertion in actions_assertions + for action in assertion["data"]["actions"] if action["action"] == "c2pa.opened" ] self.assertEqual(len(opened_actions), 1, "Should have exactly one c2pa.opened action") # Verify the c2pa.opened action has the correct structure with parameters and ingredients - opened_action = opened_actions[0] + opened_action, = opened_actions self.assertIn("parameters", opened_action, "c2pa.opened action should have parameters") self.assertIn("ingredients", opened_action["parameters"], "parameters should have ingredients array") ingredients_params = opened_action["parameters"]["ingredients"] @@ -3807,6 +3806,7 @@ def test_builder_sign_with_ingredients_edit_intent(self): for ingredient_ref in ingredients_params: self.assertIn("url", ingredient_ref, "Ingredient reference should have url") self.assertIn("hash", ingredient_ref, "Ingredient reference should have hash") + self.assertEqual(ingredient_ref["url"].rsplit("/", 1)[-1], parent_ingredient["label"]) builder.close() @@ -5182,7 +5182,7 @@ def test_builder_minimal_manifest_add_actions_and_sign_no_auto_add(self): load_settings('{"builder":{"actions":{"auto_placed_action":{"enabled":true},"auto_opened_action":{"enabled":true},"auto_created_action":{"enabled":true,"source_type":"http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation"}}}}') def test_builder_minimal_manifest_add_actions_and_sign_with_auto_add(self): - # For testing, remove auto-added actions + # Enable auto-added actions. load_settings('{"builder":{"actions":{"auto_placed_action":{"enabled":true},"auto_opened_action":{"enabled":true},"auto_created_action":{"enabled":true,"source_type":"http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation"}}}}') initial_manifest_definition = { @@ -5219,36 +5219,36 @@ def test_builder_minimal_manifest_add_actions_and_sign_with_auto_add(self): self.assertIn("assertions", active_manifest) assertions = active_manifest["assertions"] - # Find the c2pa.actions.v2 assertion to look for what we added - actions_assertion = None - for assertion in assertions: - if assertion.get("label") == "c2pa.actions.v2": - actions_assertion = assertion - break - - self.assertIsNotNone(actions_assertion) - self.assertIn("data", actions_assertion) - assertion_data = actions_assertion["data"] - # Verify the manifest now contains actions - self.assertIn("actions", assertion_data) - actions = assertion_data["actions"] - # Verify "c2pa.created" action exists anywhere in the actions array - created_action_found = False - for action in actions: - if action.get("action") == "c2pa.created": - created_action_found = True - break - - self.assertTrue(created_action_found) - - # Verify "c2pa.color_adjustments" action also exists in the same actions array - color_adjustments_found = False - for action in actions: - if action.get("action") == "c2pa.color_adjustments": - color_adjustments_found = True - break - - self.assertTrue(color_adjustments_found) + # c2pa-rs #2720 (309b99abe) separates synthesized created actions from gathered caller actions. + actions_assertions = [ + assertion for assertion in assertions + if assertion["label"].startswith("c2pa.actions") + ] + self.assertTrue(actions_assertions) + actions = [] + for assertion in actions_assertions: + self.assertIn("data", assertion) + self.assertIn("actions", assertion["data"]) + actions.extend(enumerate(assertion["data"]["actions"])) + + inception_actions = [ + (index, action) for index, action in actions + if action["action"] in ("c2pa.created", "c2pa.opened") + ] + self.assertEqual(len(inception_actions), 1, "Should have exactly one inception action") + (created_index, created_action), = inception_actions + self.assertEqual(created_index, 0, "Inception must be the first action of its assertion") + self.assertEqual(created_action["action"], "c2pa.created") + self.assertEqual(created_action["digitalSourceType"], + "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation") + + color_adjustments = [ + action for _, action in actions + if action["action"] == "c2pa.color_adjustments" + ] + self.assertEqual(len(color_adjustments), 1, "Should have exactly one color adjustment") + color_adjustment, = color_adjustments + self.assertEqual(color_adjustment["parameters"]["name"], "brightnesscontrast") builder.close() From 6adcc40861c9c91d7cd7802686e508e1ca30491b Mon Sep 17 00:00:00 2001 From: Michael Stattmann Date: Sat, 17 Oct 2026 10:44:05 +0200 Subject: [PATCH 10/10] test: retain actions v2 checks across assertion instances --- tests/test_unit_tests.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index 349292ea..a09f6a7f 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -3782,7 +3782,7 @@ def test_builder_sign_with_ingredients_edit_intent(self): actions_assertions = [ assertion for assertion in assertions - if assertion["label"].startswith("c2pa.actions") + if assertion["label"].split("__", 1)[0] == "c2pa.actions.v2" ] self.assertTrue(actions_assertions, "Should have c2pa.actions assertion") @@ -5222,7 +5222,7 @@ def test_builder_minimal_manifest_add_actions_and_sign_with_auto_add(self): # c2pa-rs #2720 (309b99abe) separates synthesized created actions from gathered caller actions. actions_assertions = [ assertion for assertion in assertions - if assertion["label"].startswith("c2pa.actions") + if assertion["label"].split("__", 1)[0] == "c2pa.actions.v2" ] self.assertTrue(actions_assertions) actions = []