diff --git a/tests/unit/data/corrupt_exif_large_thumbnail.jpg b/tests/unit/data/corrupt_exif_large_thumbnail.jpg new file mode 100644 index 000000000..389c4eacc Binary files /dev/null and b/tests/unit/data/corrupt_exif_large_thumbnail.jpg differ diff --git a/tests/unit/data/corrupt_exif_trusted_wrong_type.jpg b/tests/unit/data/corrupt_exif_trusted_wrong_type.jpg new file mode 100644 index 000000000..336169364 Binary files /dev/null and b/tests/unit/data/corrupt_exif_trusted_wrong_type.jpg differ diff --git a/tests/unit/data/corrupt_exif_wrong_type.jpg b/tests/unit/data/corrupt_exif_wrong_type.jpg new file mode 100644 index 000000000..a6656932d Binary files /dev/null and b/tests/unit/data/corrupt_exif_wrong_type.jpg differ diff --git a/tests/unit/test_api_v4.py b/tests/unit/test_api_v4.py index c0e259824..07287c795 100644 --- a/tests/unit/test_api_v4.py +++ b/tests/unit/test_api_v4.py @@ -7,7 +7,6 @@ import pytest import requests - from mapillary_tools import api_v4 diff --git a/tests/unit/test_camm_parser.py b/tests/unit/test_camm_parser.py index e4bd12d62..23f274a33 100644 --- a/tests/unit/test_camm_parser.py +++ b/tests/unit/test_camm_parser.py @@ -10,7 +10,11 @@ from mapillary_tools import geo, telemetry, types, uploader from mapillary_tools.camm import camm_builder, camm_parser -from mapillary_tools.mp4 import construct_mp4_parser as cparser, simple_mp4_builder +from mapillary_tools.mp4 import ( + construct_mp4_parser as cparser, + mp4_sample_parser as sample_parser, + simple_mp4_builder, +) def test_filter_points_by_edit_list(): @@ -356,8 +360,6 @@ def test_build_and_parse3(): def test_camm_trak_carries_mvhd_timestamps(): """Verify that creation_time and modification_time from the source video's mvhd are carried into the CAMM track's tkhd and mdhd boxes.""" - from mapillary_tools.mp4 import mp4_sample_parser as sample_parser - movie_timescale = 1_000_000 src_creation_time = 3_692_845_200 # 2021-01-01 in MP4 epoch src_modification_time = 3_692_845_300 diff --git a/tests/unit/test_exifedit.py b/tests/unit/test_exifedit.py index 571b41efc..926ebbbeb 100644 --- a/tests/unit/test_exifedit.py +++ b/tests/unit/test_exifedit.py @@ -27,6 +27,15 @@ CORRUPT_EXIF_FILE_2 = data_dir.joinpath("corrupt_exif_2.jpg") FIXED_EXIF_FILE = data_dir.joinpath("fixed_exif.jpg") FIXED_EXIF_FILE_2 = data_dir.joinpath("fixed_exif_2.jpg") +# JPEGs whose EXIF can be read but cannot be written back out unchanged. +# UNDUMPABLE_EXIF_FILE carries a non-essential tag that cannot be saved; +# UNDUMPABLE_ESSENTIAL_EXIF_FILE carries an essential one. +UNDUMPABLE_EXIF_FILE = data_dir.joinpath("corrupt_exif_wrong_type.jpg") +UNDUMPABLE_ESSENTIAL_EXIF_FILE = data_dir.joinpath( + "corrupt_exif_trusted_wrong_type.jpg" +) +# A JPEG whose embedded thumbnail is too large to be written back out. +LARGE_THUMBNAIL_EXIF_FILE = data_dir.joinpath("corrupt_exif_large_thumbnail.jpg") def add_image_description_general(_test_obj, filename): @@ -216,6 +225,54 @@ def test_add_negative_lat_lon(self): assert (test_longitude, test_latitude) == exif_data.extract_lon_lat() + def test_add_make_and_model(self): + empty_exifedit = ExifEdit(EMPTY_EXIF_FILE_TEST) + empty_exifedit.add_make("Canon") + empty_exifedit.add_model("EOS 5D") + empty_exifedit.write(EMPTY_EXIF_FILE_TEST) + + exif_data = ExifRead(EMPTY_EXIF_FILE_TEST) + self.assertEqual("Canon", exif_data.extract_make()) + self.assertEqual("EOS 5D", exif_data.extract_model()) + + def test_add_make_empty_raises(self): + empty_exifedit = ExifEdit(EMPTY_EXIF_FILE_TEST) + with self.assertRaises(ValueError): + empty_exifedit.add_make("") + + def test_add_model_empty_raises(self): + empty_exifedit = ExifEdit(EMPTY_EXIF_FILE_TEST) + with self.assertRaises(ValueError): + empty_exifedit.add_model("") + + def test_add_orientation_invalid_raises(self): + empty_exifedit = ExifEdit(EMPTY_EXIF_FILE_TEST) + with self.assertRaises(ValueError): + empty_exifedit.add_orientation(99) + + def test_write_bytes_without_filename_raises(self): + with open(EMPTY_EXIF_FILE_TEST, "rb") as fp: + edit = ExifEdit(fp.read()) + edit.add_orientation(1) + # The source is raw bytes, so write() has no filename to fall back on. + with self.assertRaises(ValueError): + edit.write() + + def test_unwritable_non_essential_tag_is_dropped(self): + """An image with an unsavable non-essential tag is still saved without it.""" + edit = ExifEdit(UNDUMPABLE_EXIF_FILE) + image_bytes = edit.dump_image_bytes() + self.assertGreater(len(image_bytes), 0) + # The unsavable tag is absent from the output. + saved = piexif.load(image_bytes) + self.assertNotIn(piexif.ImageIFD.Software, saved["0th"]) + + def test_unwritable_essential_tag_raises(self): + """Saving fails if an essential tag cannot be written, rather than dropping it.""" + edit = ExifEdit(UNDUMPABLE_ESSENTIAL_EXIF_FILE) + with self.assertRaises(ValueError): + edit.dump_image_bytes() + # REPEAT CERTAIN TESTS AND ADD ADDITIONAL TESTS FOR THE CORRUPT EXIF def test_load_and_dump_corrupt_exif(self): corrupt_exifedit = ExifEdit(CORRUPT_EXIF_FILE) @@ -268,57 +325,17 @@ def test_add_repeatedly_time_original_corrupt_exif_2(self): add_repeatedly_time_original_general(self, CORRUPT_EXIF_FILE_2) def test_large_thumbnail_handling(self): - """Test that images with thumbnails larger than 64kB are handled gracefully.""" - # Create a test image with a large thumbnail (>64kB) - test_image_path = data_dir.joinpath("tmp", "large_thumbnail.jpg") - - # Create a simple test image - img = Image.new("RGB", (100, 100), color="red") - img.save(test_image_path, "JPEG") - - # Create a large thumbnail (>64kB) by creating a high-quality large thumbnail - # Use a larger size and add noise to make it incompressible - large_thumbnail = Image.new("RGB", (2048, 2048)) - # Fill with random-like data to prevent compression - pixels = large_thumbnail.load() - for i in range(2048): - for j in range(2048): - pixels[i, j] = ( - (i * 7 + j * 13) % 256, - (i * 11 + j * 17) % 256, - (i * 19 + j * 23) % 256, - ) - - thumbnail_bytes = io.BytesIO() - large_thumbnail.save(thumbnail_bytes, "JPEG", quality=100) - thumbnail_data = thumbnail_bytes.getvalue() - - # Verify thumbnail is larger than 64kB - self.assertGreater( - len(thumbnail_data), - 64 * 1024, - f"Test thumbnail should be larger than 64kB but got {len(thumbnail_data)} bytes", - ) + """An oversized embedded thumbnail is dropped on save; other EXIF survives. - # Load the image and add GPS data first - exif_edit = ExifEdit(test_image_path) + The image carries a thumbnail too large to be written back out. Saving + should still succeed, and GPS data added through the API is preserved. + """ + exif_edit = ExifEdit(LARGE_THUMBNAIL_EXIF_FILE) test_latitude = 50.5475894785 test_longitude = 15.595866685 exif_edit.add_lat_lon(test_latitude, test_longitude) - # Manually insert the large thumbnail into the internal EXIF structure - # This simulates what would happen if an image came in with a large thumbnail - exif_edit._ef["thumbnail"] = thumbnail_data - exif_edit._ef["1st"] = { - piexif.ImageIFD.Compression: 6, - piexif.ImageIFD.XResolution: (72, 1), - piexif.ImageIFD.YResolution: (72, 1), - piexif.ImageIFD.ResolutionUnit: 2, - piexif.ImageIFD.JPEGInterchangeFormat: 0, - piexif.ImageIFD.JPEGInterchangeFormatLength: len(thumbnail_data), - } - - # Given thumbnail is too large, max 64kB, thumbnail and 1st metadata should be removed. + # The oversized thumbnail should be dropped so the image can be saved. image_bytes = exif_edit.dump_image_bytes() # Verify the output is valid @@ -330,25 +347,23 @@ def test_large_thumbnail_handling(self): self.assertEqual(result_image.format, "JPEG") self.assertEqual(result_image.size, (100, 100)) - # Verify we can read the GPS data from the result + # The GPS data added before saving is still present. output_exif = piexif.load(image_bytes) self.assertIn("GPS", output_exif) self.assertIn(piexif.GPSIFD.GPSLatitude, output_exif["GPS"]) self.assertIn(piexif.GPSIFD.GPSLongitude, output_exif["GPS"]) - # CRITICAL: Verify the large thumbnail was actually removed - # The fix should have deleted both "thumbnail" and "1st" to handle the error - # piexif.load() may include "thumbnail": None after removal + # The oversized thumbnail is no longer present in the saved image. thumbnail_value = output_exif.get("thumbnail") self.assertTrue( thumbnail_value is None or thumbnail_value == b"", - f"Large thumbnail should have been removed but got: {thumbnail_value[:100] if thumbnail_value else None}", + f"thumbnail should have been removed but got: {thumbnail_value[:100] if thumbnail_value else None}", ) first_value = output_exif.get("1st") self.assertTrue( first_value is None or first_value == {} or len(first_value) == 0, - f"1st metadata should have been removed but got: {first_value}", + f"thumbnail metadata should have been removed but got: {first_value}", ) diff --git a/tests/unit/test_exifread.py b/tests/unit/test_exifread.py index 8e7d51c49..b0866aae9 100644 --- a/tests/unit/test_exifread.py +++ b/tests/unit/test_exifread.py @@ -4,17 +4,22 @@ # LICENSE file in the root directory of this source tree. import datetime +import io import os +import struct import typing as T +import xml.etree.ElementTree as ET from pathlib import Path import py.path import pytest -from mapillary_tools import geo from mapillary_tools.exif_read import ( - _parse_coord, ExifRead, + ExifReadFromEXIF, + ExifReadFromXMP, + extract_xmp_efficiently, parse_datetimestr_with_subsec_and_offset, + XMP_NAMESPACES, ) from mapillary_tools.exif_write import ExifEdit @@ -43,6 +48,19 @@ def gps_to_decimal(value, ref): return sign * (float(degrees) + float(minutes) / 60 + float(seconds) / 3600) +def as_unix_time(dt: datetime.datetime) -> float: + try: + # if dt is naive, assume it's in local timezone + return dt.timestamp() + except ValueError: + # Some datetimes can't be converted to timestamp + # e.g. 0001-01-01 00:00:00 will throw ValueError: year 0 is out of range + try: + return dt.replace(year=1970).timestamp() + except ValueError: + return 0.0 + + def test_read_orientation_general(): exif_data_ExifRead = ExifRead(TEST_EXIF_FILE) orientation_ExifRead = exif_data_ExifRead.extract_orientation() @@ -203,6 +221,9 @@ def test_parse(): assert str(dt) == "2021-10-10 17:29:54.124000-02:00", dt +# Coordinate parsing is exercised through the public ExifReadFromXMP.extract_lon_lat: +# the raw value is supplied as the latitude (with a fixed, always-valid longitude so +# the call returns), and the returned latitude reflects how the raw value was parsed. @pytest.mark.parametrize( "raw_coord,raw_ref,expected", [ @@ -225,7 +246,20 @@ def test_parse(): def test_parse_coordinates( raw_coord: T.Optional[str], raw_ref: str, expected: T.Optional[float] ): - assert _parse_coord(raw_coord, raw_ref) == pytest.approx(expected) + tags = {"exif:GPSLongitude": "15.5", "exif:GPSLongitudeRef": "E"} + if raw_coord is not None: + tags["exif:GPSLatitude"] = raw_coord + tags["exif:GPSLatitudeRef"] = raw_ref + + lonlat = _make_xmp_reader(tags).extract_lon_lat() + + if expected is None: + assert lonlat is None + else: + assert lonlat is not None + lon, lat = lonlat + assert lon == pytest.approx(15.5) + assert lat == pytest.approx(expected) # test ExifWrite write a timestamp and ExifRead read it back @@ -253,7 +287,7 @@ def test_read_and_write(setup_data: py.path.local): read = ExifRead(image_path) actual = read.extract_capture_time() assert actual - assert geo.as_unix_time(dt) == geo.as_unix_time(actual), (dt, actual) + assert as_unix_time(dt) == as_unix_time(actual), (dt, actual) for dt in dts: edit = ExifEdit(image_path) @@ -262,7 +296,7 @@ def test_read_and_write(setup_data: py.path.local): read = ExifRead(image_path) actual = read.extract_gps_datetime() assert actual - assert geo.as_unix_time(dt) == geo.as_unix_time(actual) + assert as_unix_time(dt) == as_unix_time(actual) # Tests for extract_camera_uuid @@ -280,7 +314,6 @@ class TestExtractCameraUuidFromEXIF: def test_body_serial_only(self): """Test with only body serial number present""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -290,7 +323,6 @@ def test_body_serial_only(self): def test_lens_serial_only(self): """Test with only lens serial number present""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -300,7 +332,6 @@ def test_lens_serial_only(self): def test_both_body_and_lens_serial(self): """Test with both body and lens serial numbers present""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -311,7 +342,6 @@ def test_both_body_and_lens_serial(self): def test_no_serial_numbers(self): """Test with no serial numbers present""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = {} @@ -319,7 +349,6 @@ def test_no_serial_numbers(self): def test_generic_serial_fallback(self): """Test fallback to generic EXIF SerialNumber""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -329,7 +358,6 @@ def test_generic_serial_fallback(self): def test_makernote_serial_fallback(self): """Test fallback to MakerNote SerialNumber""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -339,7 +367,6 @@ def test_makernote_serial_fallback(self): def test_body_serial_priority_over_generic(self): """Test that BodySerialNumber takes priority over generic SerialNumber""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -350,7 +377,6 @@ def test_body_serial_priority_over_generic(self): def test_whitespace_stripped(self): """Test that whitespace is stripped from serial numbers""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -361,7 +387,6 @@ def test_whitespace_stripped(self): def test_special_characters_removed(self): """Test that special characters are removed from serial numbers""" - from mapillary_tools.exif_read import ExifReadFromEXIF reader = ExifReadFromEXIF.__new__(ExifReadFromEXIF) reader.tags = { @@ -376,9 +401,6 @@ class TestExtractCameraUuidFromXMP: def _create_xmp_reader(self, tags_dict: dict): """Helper to create an ExifReadFromXMP with mocked tags""" - from mapillary_tools.exif_read import ExifReadFromXMP, XMP_NAMESPACES - import xml.etree.ElementTree as ET - # Build a minimal XMP document rdf_ns = XMP_NAMESPACES["rdf"] xmp_xml = f''' @@ -451,179 +473,264 @@ def test_real_image_camera_uuid(self): assert result is None or isinstance(result, str) -class TestVideoExtractCameraUuid: - """Test extract_camera_uuid for video EXIF reader""" +def _build_xmp_doc(tags: T.Dict[str, str]) -> str: + """Build an XMP packet whose rdf:Description carries ``tags`` as attributes.""" + rdf_ns = XMP_NAMESPACES["rdf"] + xml = ( + '' + '' + f'' + " ExifReadFromXMP: + return ExifReadFromXMP(ET.ElementTree(ET.fromstring(_build_xmp_doc(tags)))) - # Add child elements for each tag - for key, value in tags_dict.items(): - prefix, tag_name = key.split(":") - if prefix in EXIFTOOL_NAMESPACES: - full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name - child = ET.SubElement(root, full_tag) - child.text = value - - etree = ET.ElementTree(root) - return ExifToolReadVideo(etree) - - def test_gopro_serial(self): - """Test extraction of GoPro serial number""" - reader = self._create_video_exif_reader( - {"GoPro:SerialNumber": "C3456789012345"} - ) - assert reader.extract_camera_uuid() == "C3456789012345" - def test_insta360_serial(self): - """Test extraction of Insta360 serial number""" - reader = self._create_video_exif_reader( - {"Insta360:SerialNumber": "INST360SERIAL"} +def _build_jpeg_with_xmp(xmp_xml: str) -> bytes: + """Build a minimal JPEG containing ``xmp_xml`` in an APP1 XMP segment.""" + identifier = b"http://ns.adobe.com/xap/1.0/\x00" + payload = identifier + xmp_xml.encode("utf-8") + # APP1 length field counts itself (2 bytes) plus the payload + app1 = b"\xff\xe1" + struct.pack(">H", len(payload) + 2) + payload + return b"\xff\xd8" + app1 + b"\xff\xd9" # SOI ... EOI + + +class TestExifReadFromXMPMetadata: + """Tests for reading metadata from XMP.""" + + def test_extract_altitude(self): + assert ( + _make_xmp_reader({"exif:GPSAltitude": "123.5"}).extract_altitude() == 123.5 ) - assert reader.extract_camera_uuid() == "INST360SERIAL" - def test_exif_body_serial(self): - """Test extraction of standard EXIF body serial number""" - reader = self._create_video_exif_reader({"ExifIFD:BodySerialNumber": "BODY123"}) - assert reader.extract_camera_uuid() == "BODY123" + def test_extract_altitude_missing(self): + assert _make_xmp_reader({}).extract_altitude() is None - def test_exif_body_and_lens_serial(self): - """Test extraction of both body and lens serial numbers""" - reader = self._create_video_exif_reader( + def test_extract_lon_lat_numeric(self): + reader = _make_xmp_reader( { - "ExifIFD:BodySerialNumber": "BODY123", - "ExifIFD:LensSerialNumber": "LENS456", + "exif:GPSLatitude": "50.5", + "exif:GPSLatitudeRef": "N", + "exif:GPSLongitude": "15.5", + "exif:GPSLongitudeRef": "E", } ) - assert reader.extract_camera_uuid() == "BODY123_LENS456" + assert reader.extract_lon_lat() == (15.5, 50.5) - def test_no_serial(self): - """Test with no serial numbers present""" - reader = self._create_video_exif_reader({}) - assert reader.extract_camera_uuid() is None - - def test_gopro_priority(self): - """Test that GoPro serial takes priority over generic serial""" - reader = self._create_video_exif_reader( + def test_extract_lon_lat_adobe_format(self): + reader = _make_xmp_reader( { - "GoPro:SerialNumber": "GOPRO123", - "IFD0:SerialNumber": "GENERIC789", + "exif:GPSLatitude": "33,18.32N", + "exif:GPSLatitudeRef": "N", + "exif:GPSLongitude": "44,24.54E", + "exif:GPSLongitudeRef": "E", } ) - assert reader.extract_camera_uuid() == "GOPRO123" + lonlat = reader.extract_lon_lat() + assert lonlat is not None + lon, lat = lonlat + assert lat == pytest.approx(33.30533, abs=1e-4) + assert lon == pytest.approx(44.40900, abs=1e-4) + + def test_extract_lon_lat_missing(self): + assert _make_xmp_reader({}).extract_lon_lat() is None + + def test_extract_make_and_model_stripped(self): + reader = _make_xmp_reader({"tiff:Make": "Canon ", "tiff:Model": " EOS "}) + assert reader.extract_make() == "Canon" + assert reader.extract_model() == "EOS" + + def test_extract_make_lens_fallback(self): + assert ( + _make_xmp_reader({"exifEX:LensMake": "LensCo"}).extract_make() == "LensCo" + ) + def test_extract_make_missing(self): + assert _make_xmp_reader({}).extract_make() is None + assert _make_xmp_reader({}).extract_model() is None -class TestExifToolReadExtractCameraUuid: - """Test extract_camera_uuid for ExifToolRead (image EXIF via ExifTool XML)""" + def test_extract_width_height(self): + reader = _make_xmp_reader( + {"exif:PixelXDimension": "1920", "exif:PixelYDimension": "1080"} + ) + assert reader.extract_width() == 1920 + assert reader.extract_height() == 1080 - def _create_exiftool_reader(self, tags_dict: dict): - """Helper to create an ExifToolRead with mocked tags""" - from mapillary_tools.exiftool_read import ExifToolRead, EXIFTOOL_NAMESPACES - import xml.etree.ElementTree as ET + def test_extract_width_height_gpano_fallback(self): + assert ( + _make_xmp_reader({"GPano:FullPanoWidthPixels": "4096"}).extract_width() + == 4096 + ) + assert ( + _make_xmp_reader( + {"GPano:CroppedAreaImageHeightPixels": "2048"} + ).extract_height() + == 2048 + ) - # Build XML structure that ExifToolRead expects - root = ET.Element("rdf:Description") + def test_extract_orientation(self): + assert _make_xmp_reader({"tiff:Orientation": "3"}).extract_orientation() == 3 - for tag, value in tags_dict.items(): - prefix, tag_name = tag.split(":", 1) - if prefix in EXIFTOOL_NAMESPACES: - full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name - child = ET.SubElement(root, full_tag) - child.text = value + def test_extract_orientation_invalid_defaults_to_1(self): + assert _make_xmp_reader({"tiff:Orientation": "99"}).extract_orientation() == 1 - etree = ET.ElementTree(root) - return ExifToolRead(etree) + def test_extract_orientation_missing_defaults_to_1(self): + assert _make_xmp_reader({}).extract_orientation() == 1 - def test_body_serial_only(self): - """Test extraction with only body serial number""" - reader = self._create_exiftool_reader({"ExifIFD:BodySerialNumber": "BODY12345"}) - assert reader.extract_camera_uuid() == "BODY12345" + def test_extract_direction(self): + assert ( + _make_xmp_reader({"exif:GPSImgDirection": "180.5"}).extract_direction() + == 180.5 + ) - def test_lens_serial_only(self): - """Test extraction with only lens serial number""" - reader = self._create_exiftool_reader({"ExifIFD:LensSerialNumber": "LENS67890"}) - assert reader.extract_camera_uuid() == "LENS67890" + def test_extract_direction_track_fallback(self): + assert _make_xmp_reader({"exif:GPSTrack": "90.0"}).extract_direction() == 90.0 - def test_both_body_and_lens_serial(self): - """Test extraction with both body and lens serial numbers""" - reader = self._create_exiftool_reader( - { - "ExifIFD:BodySerialNumber": "BODY123", - "ExifIFD:LensSerialNumber": "LENS456", - } + def test_extract_direction_missing(self): + assert _make_xmp_reader({}).extract_direction() is None + + def test_extract_exif_datetime(self): + reader = _make_xmp_reader({"exif:DateTimeOriginal": "2021:07:15 15:37:30"}) + assert reader.extract_exif_datetime() == datetime.datetime( + 2021, 7, 15, 15, 37, 30 ) - assert reader.extract_camera_uuid() == "BODY123_LENS456" - def test_no_serial_numbers(self): - """Test with no serial numbers present""" - reader = self._create_exiftool_reader({}) - assert reader.extract_camera_uuid() is None + def test_extract_exif_datetime_digitized_fallback(self): + reader = _make_xmp_reader({"exif:DateTimeDigitized": "2020:01:02 03:04:05"}) + assert reader.extract_exif_datetime() == datetime.datetime(2020, 1, 2, 3, 4, 5) - def test_generic_serial_fallback(self): - """Test that ExifIFD:SerialNumber is used as fallback for body serial""" - reader = self._create_exiftool_reader({"ExifIFD:SerialNumber": "GENERIC123"}) - assert reader.extract_camera_uuid() == "GENERIC123" + def test_extract_exif_datetime_missing(self): + assert _make_xmp_reader({}).extract_exif_datetime() is None - def test_ifd0_serial_fallback(self): - """Test that IFD0:SerialNumber is used as fallback""" - reader = self._create_exiftool_reader({"IFD0:SerialNumber": "IFD0SN123"}) - assert reader.extract_camera_uuid() == "IFD0SN123" + def test_extract_gps_datetime_iso(self): + reader = _make_xmp_reader({"exif:GPSTimeStamp": "2021-07-15T05:37:30Z"}) + assert reader.extract_gps_datetime() == datetime.datetime( + 2021, 7, 15, 5, 37, 30, tzinfo=datetime.timezone.utc + ) - def test_body_serial_priority_over_generic(self): - """Test that BodySerialNumber takes priority over generic SerialNumber""" - reader = self._create_exiftool_reader( + def test_extract_gps_datetime_separate_date_and_time(self): + reader = _make_xmp_reader( + { + "exif:GPSDateStamp": "2021:07:15", + "exif:GPSTimeStamp": "05:37:30", + } + ) + assert reader.extract_gps_datetime() == datetime.datetime( + 2021, 7, 15, 5, 37, 30, tzinfo=datetime.timezone.utc + ) + + def test_extract_gps_datetime_missing(self): + assert _make_xmp_reader({}).extract_gps_datetime() is None + + def test_extract_capture_time_prefers_gps(self): + reader = _make_xmp_reader( { - "ExifIFD:BodySerialNumber": "BODY999", - "ExifIFD:SerialNumber": "GENERIC888", + "exif:GPSTimeStamp": "2021-07-15T05:37:30Z", + "exif:DateTimeOriginal": "2000:01:01 00:00:00", } ) - assert reader.extract_camera_uuid() == "BODY999" + assert reader.extract_capture_time() == datetime.datetime( + 2021, 7, 15, 5, 37, 30, tzinfo=datetime.timezone.utc + ) - def test_xmp_exifex_body_serial(self): - """Test XMP-exifEX:BodySerialNumber extraction""" - reader = self._create_exiftool_reader( - {"XMP-exifEX:BodySerialNumber": "XMPBODY123"} + def test_extract_capture_time_falls_back_to_exif(self): + reader = _make_xmp_reader({"exif:DateTimeOriginal": "2021:07:15 15:37:30"}) + assert reader.extract_capture_time() == datetime.datetime( + 2021, 7, 15, 15, 37, 30 ) - assert reader.extract_camera_uuid() == "XMPBODY123" - def test_xmp_aux_serial(self): - """Test XMP-aux:SerialNumber extraction""" - reader = self._create_exiftool_reader({"XMP-aux:SerialNumber": "AUXSN456"}) - assert reader.extract_camera_uuid() == "AUXSN456" + def test_extract_capture_time_missing(self): + assert _make_xmp_reader({}).extract_capture_time() is None + + +class TestExtractXmpEfficiently: + """Tests for locating XMP metadata embedded in a JPEG.""" + + def test_returns_xmp_when_present(self): + xmp = _build_xmp_doc({"tiff:Make": "Canon"}) + result = extract_xmp_efficiently(io.BytesIO(_build_jpeg_with_xmp(xmp))) + assert result is not None + assert "" in result + + def test_returns_none_without_soi(self): + assert extract_xmp_efficiently(io.BytesIO(b"not a jpeg")) is None - def test_xmp_aux_lens_serial(self): - """Test XMP-aux:LensSerialNumber extraction""" - reader = self._create_exiftool_reader( - {"XMP-aux:LensSerialNumber": "AUXLENS789"} + def test_returns_none_when_no_xmp_segment(self): + # SOI immediately followed by EOI: valid JPEG start, no APP1/XMP + assert extract_xmp_efficiently(io.BytesIO(b"\xff\xd8\xff\xd9")) is None + + def test_skips_non_xmp_app1_segment(self): + # An APP1 segment that is not XMP (e.g. an EXIF identifier) is skipped, + # and the following XMP APP1 segment is still found. + exif_id = b"Exif\x00\x00rest-of-exif" + exif_app1 = b"\xff\xe1" + struct.pack(">H", len(exif_id) + 2) + exif_id + xmp_app1 = _build_jpeg_with_xmp(_build_xmp_doc({"tiff:Make": "Canon"}))[2:] + data = b"\xff\xd8" + exif_app1 + xmp_app1 + result = extract_xmp_efficiently(io.BytesIO(data)) + assert result is not None + assert "Canon" in result + + +class TestExifReadXmpFallback: + """Reading metadata from a JPEG whose values live in XMP, not EXIF.""" + + def _make_reader(self, tags: T.Dict[str, str]) -> ExifRead: + jpeg = _build_jpeg_with_xmp(_build_xmp_doc(tags)) + return ExifRead(io.BytesIO(jpeg)) + + def test_make_model_fallback(self): + reader = self._make_reader({"tiff:Make": "XMPMake", "tiff:Model": "XMPModel"}) + assert reader.extract_make() == "XMPMake" + assert reader.extract_model() == "XMPModel" + + def test_altitude_fallback(self): + assert ( + self._make_reader({"exif:GPSAltitude": "123.5"}).extract_altitude() == 123.5 ) - assert reader.extract_camera_uuid() == "AUXLENS789" - def test_xmp_combined(self): - """Test XMP body and lens serial combined""" - reader = self._create_exiftool_reader( + def test_lon_lat_fallback(self): + reader = self._make_reader( { - "XMP-exifEX:BodySerialNumber": "XMPBODY", - "XMP-exifEX:LensSerialNumber": "XMPLENS", + "exif:GPSLatitude": "50.5", + "exif:GPSLatitudeRef": "N", + "exif:GPSLongitude": "15.5", + "exif:GPSLongitudeRef": "E", } ) - assert reader.extract_camera_uuid() == "XMPBODY_XMPLENS" + assert reader.extract_lon_lat() == (15.5, 50.5) - def test_whitespace_stripped(self): - """Test that whitespace is stripped from serial numbers""" - reader = self._create_exiftool_reader( - { - "ExifIFD:BodySerialNumber": " BODY123 ", - "ExifIFD:LensSerialNumber": " LENS456 ", - } + def test_width_height_fallback(self): + reader = self._make_reader( + {"exif:PixelXDimension": "1920", "exif:PixelYDimension": "1080"} ) - assert reader.extract_camera_uuid() == "BODY123_LENS456" + assert reader.extract_width() == 1920 + assert reader.extract_height() == 1080 + + def test_capture_time_fallback(self): + reader = self._make_reader({"exif:DateTimeOriginal": "2020:01:02 03:04:05"}) + assert reader.extract_capture_time() == datetime.datetime(2020, 1, 2, 3, 4, 5) + + def test_camera_uuid_fallback(self): + reader = self._make_reader( + {"exif:SerialNumber": "BODYX", "exif:LensSerialNumber": "LENSY"} + ) + assert reader.extract_camera_uuid() == "BODYX_LENSY" + + def test_no_xmp_and_no_exif_returns_none(self): + # A JPEG with neither EXIF nor XMP: every extractor returns None. + reader = ExifRead(io.BytesIO(b"\xff\xd8\xff\xd9")) + assert reader.extract_make() is None + assert reader.extract_lon_lat() is None + assert reader.extract_capture_time() is None + assert reader.extract_camera_uuid() is None diff --git a/tests/unit/test_exiftool_read.py b/tests/unit/test_exiftool_read.py new file mode 100644 index 000000000..cb2df2604 --- /dev/null +++ b/tests/unit/test_exiftool_read.py @@ -0,0 +1,114 @@ +# Copyright (c) Meta Platforms, Inc. and affiliates. +# +# This source code is licensed under the BSD license found in the +# LICENSE file in the root directory of this source tree. + +from __future__ import annotations + +import typing as T +import xml.etree.ElementTree as ET + +from mapillary_tools.exiftool_read import EXIFTOOL_NAMESPACES, ExifToolRead + + +class TestExifToolReadExtractCameraUuid: + """Test extract_camera_uuid for ExifToolRead (image EXIF via ExifTool XML)""" + + def _create_exiftool_reader(self, tags_dict: T.Dict[str, str]) -> ExifToolRead: + """Helper to create an ExifToolRead with mocked tags""" + # Build XML structure that ExifToolRead expects + root = ET.Element("rdf:Description") + + for tag, value in tags_dict.items(): + prefix, tag_name = tag.split(":", 1) + if prefix in EXIFTOOL_NAMESPACES: + full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name + child = ET.SubElement(root, full_tag) + child.text = value + + etree = ET.ElementTree(root) + return ExifToolRead(etree) + + def test_body_serial_only(self): + """Test extraction with only body serial number""" + reader = self._create_exiftool_reader({"ExifIFD:BodySerialNumber": "BODY12345"}) + assert reader.extract_camera_uuid() == "BODY12345" + + def test_lens_serial_only(self): + """Test extraction with only lens serial number""" + reader = self._create_exiftool_reader({"ExifIFD:LensSerialNumber": "LENS67890"}) + assert reader.extract_camera_uuid() == "LENS67890" + + def test_both_body_and_lens_serial(self): + """Test extraction with both body and lens serial numbers""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": "BODY123", + "ExifIFD:LensSerialNumber": "LENS456", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + def test_no_serial_numbers(self): + """Test with no serial numbers present""" + reader = self._create_exiftool_reader({}) + assert reader.extract_camera_uuid() is None + + def test_generic_serial_fallback(self): + """Test that ExifIFD:SerialNumber is used as fallback for body serial""" + reader = self._create_exiftool_reader({"ExifIFD:SerialNumber": "GENERIC123"}) + assert reader.extract_camera_uuid() == "GENERIC123" + + def test_ifd0_serial_fallback(self): + """Test that IFD0:SerialNumber is used as fallback""" + reader = self._create_exiftool_reader({"IFD0:SerialNumber": "IFD0SN123"}) + assert reader.extract_camera_uuid() == "IFD0SN123" + + def test_body_serial_priority_over_generic(self): + """Test that BodySerialNumber takes priority over generic SerialNumber""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": "BODY999", + "ExifIFD:SerialNumber": "GENERIC888", + } + ) + assert reader.extract_camera_uuid() == "BODY999" + + def test_xmp_exifex_body_serial(self): + """Test XMP-exifEX:BodySerialNumber extraction""" + reader = self._create_exiftool_reader( + {"XMP-exifEX:BodySerialNumber": "XMPBODY123"} + ) + assert reader.extract_camera_uuid() == "XMPBODY123" + + def test_xmp_aux_serial(self): + """Test XMP-aux:SerialNumber extraction""" + reader = self._create_exiftool_reader({"XMP-aux:SerialNumber": "AUXSN456"}) + assert reader.extract_camera_uuid() == "AUXSN456" + + def test_xmp_aux_lens_serial(self): + """Test XMP-aux:LensSerialNumber extraction""" + reader = self._create_exiftool_reader( + {"XMP-aux:LensSerialNumber": "AUXLENS789"} + ) + assert reader.extract_camera_uuid() == "AUXLENS789" + + def test_xmp_combined(self): + """Test XMP body and lens serial combined""" + reader = self._create_exiftool_reader( + { + "XMP-exifEX:BodySerialNumber": "XMPBODY", + "XMP-exifEX:LensSerialNumber": "XMPLENS", + } + ) + assert reader.extract_camera_uuid() == "XMPBODY_XMPLENS" + + def test_whitespace_stripped(self): + """Test that whitespace is stripped from serial numbers""" + reader = self._create_exiftool_reader( + { + "ExifIFD:BodySerialNumber": " BODY123 ", + "ExifIFD:LensSerialNumber": " LENS456 ", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" diff --git a/tests/unit/test_exiftool_read_video.py b/tests/unit/test_exiftool_read_video.py index ced1c5024..b5890b421 100644 --- a/tests/unit/test_exiftool_read_video.py +++ b/tests/unit/test_exiftool_read_video.py @@ -8,7 +8,6 @@ import xml.etree.ElementTree as ET import pytest - from mapillary_tools.exiftool_read_video import ( _aggregate_gps_track, _aggregate_gps_track_by_sample_time, @@ -17,6 +16,7 @@ _extract_alternative_fields, _index_text_by_tag, _same_gps_point, + EXIFTOOL_NAMESPACES, ExifToolReadVideo, expand_tag, ) @@ -1338,3 +1338,74 @@ def test_rdf_description_with_no_children(self): assert reader.extract_make() is None assert reader.extract_model() is None assert reader.extract_camera_uuid() is None + + +# --------------------------------------------------------------------------- +# ExifToolReadVideo.extract_camera_uuid (built from attribute-style tag dicts) +# --------------------------------------------------------------------------- + + +class TestVideoExtractCameraUuid: + """Test extract_camera_uuid for video EXIF reader""" + + def _create_video_exif_reader(self, tags_dict: dict) -> ExifToolReadVideo: + """Helper to create an ExifToolReadVideo with mocked tags""" + # Build XML with child elements (not attributes) - this is how ExifTool XML works + root = ET.Element( + "rdf:RDF", {"xmlns:rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#"} + ) + + # Add child elements for each tag + for key, value in tags_dict.items(): + prefix, tag_name = key.split(":") + if prefix in EXIFTOOL_NAMESPACES: + full_tag = "{" + EXIFTOOL_NAMESPACES[prefix] + "}" + tag_name + child = ET.SubElement(root, full_tag) + child.text = value + + etree = ET.ElementTree(root) + return ExifToolReadVideo(etree) + + def test_gopro_serial(self): + """Test extraction of GoPro serial number""" + reader = self._create_video_exif_reader( + {"GoPro:SerialNumber": "C3456789012345"} + ) + assert reader.extract_camera_uuid() == "C3456789012345" + + def test_insta360_serial(self): + """Test extraction of Insta360 serial number""" + reader = self._create_video_exif_reader( + {"Insta360:SerialNumber": "INST360SERIAL"} + ) + assert reader.extract_camera_uuid() == "INST360SERIAL" + + def test_exif_body_serial(self): + """Test extraction of standard EXIF body serial number""" + reader = self._create_video_exif_reader({"ExifIFD:BodySerialNumber": "BODY123"}) + assert reader.extract_camera_uuid() == "BODY123" + + def test_exif_body_and_lens_serial(self): + """Test extraction of both body and lens serial numbers""" + reader = self._create_video_exif_reader( + { + "ExifIFD:BodySerialNumber": "BODY123", + "ExifIFD:LensSerialNumber": "LENS456", + } + ) + assert reader.extract_camera_uuid() == "BODY123_LENS456" + + def test_no_serial(self): + """Test with no serial numbers present""" + reader = self._create_video_exif_reader({}) + assert reader.extract_camera_uuid() is None + + def test_gopro_priority(self): + """Test that GoPro serial takes priority over generic serial""" + reader = self._create_video_exif_reader( + { + "GoPro:SerialNumber": "GOPRO123", + "IFD0:SerialNumber": "GENERIC789", + } + ) + assert reader.extract_camera_uuid() == "GOPRO123" diff --git a/tests/unit/test_geo.py b/tests/unit/test_geo.py index 27790ad82..bd3a785e7 100644 --- a/tests/unit/test_geo.py +++ b/tests/unit/test_geo.py @@ -5,12 +5,14 @@ import dataclasses import datetime +import math import random import typing as T import unittest from mapillary_tools import geo from mapillary_tools.geo import Point +from mapillary_tools.telemetry import CAMMGPSPoint, GPSFix, GPSPoint # lat, lon, bearing, alt @@ -778,7 +780,6 @@ def test_avg_speed_with_base_points(self): def test_avg_speed_with_gps_points_using_epoch_time(self): """Test avg_speed with GPSPoint using epoch_time field.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix # Video time is 0-10 seconds, but GPS epoch time spans 100 seconds # This simulates timelapse where video time != GPS time @@ -814,7 +815,6 @@ def test_avg_speed_with_gps_points_using_epoch_time(self): def test_avg_speed_with_gps_points_fallback_to_time(self): """Test avg_speed with GPSPoint falls back to time when epoch_time is None.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix points = [ GPSPoint( @@ -847,7 +847,6 @@ def test_avg_speed_with_gps_points_fallback_to_time(self): def test_avg_speed_with_gps_points_zero_epoch_time_fallback(self): """Test avg_speed falls back to time when epoch_time is 0.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix points = [ GPSPoint( @@ -879,7 +878,6 @@ def test_avg_speed_with_gps_points_zero_epoch_time_fallback(self): def test_avg_speed_with_camm_gps_points(self): """Test avg_speed with CAMMGPSPoint using time_gps_epoch field.""" - from mapillary_tools.telemetry import CAMMGPSPoint # Video time is 0-10 seconds, but GPS epoch time spans 50 seconds points = [ @@ -922,7 +920,6 @@ def test_avg_speed_with_camm_gps_points(self): def test_avg_speed_with_camm_gps_points_zero_epoch_fallback(self): """Test avg_speed with CAMMGPSPoint falls back when time_gps_epoch is 0.""" - from mapillary_tools.telemetry import CAMMGPSPoint points = [ CAMMGPSPoint( @@ -973,7 +970,6 @@ def test_avg_speed_single_point(self): def test_avg_speed_zero_time_diff_returns_nan(self): """Test avg_speed returns NaN when time difference is zero.""" - import math # Two points at the same timestamp points = [ @@ -989,7 +985,6 @@ class TestInterpolatePreservesPointType(unittest.TestCase): def test_interpolate_gps_points_returns_gps_point(self): """Test that interpolating GPSPoints returns a GPSPoint.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix points = [ GPSPoint( @@ -1035,7 +1030,6 @@ def test_interpolate_gps_points_returns_gps_point(self): def test_interpolate_gps_points_with_none_epoch_time(self): """Test interpolating GPSPoints when epoch_time is None.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix points = [ GPSPoint( @@ -1071,7 +1065,6 @@ def test_interpolate_gps_points_with_none_epoch_time(self): def test_interpolate_camm_gps_points_returns_camm_gps_point(self): """Test that interpolating CAMMGPSPoints returns a CAMMGPSPoint.""" - from mapillary_tools.telemetry import CAMMGPSPoint points = [ CAMMGPSPoint( @@ -1145,7 +1138,6 @@ def test_interpolate_base_points_returns_base_point(self): def test_interpolator_preserves_gps_point_type(self): """Test that Interpolator preserves GPSPoint type.""" - from mapillary_tools.telemetry import GPSPoint, GPSFix track = [ GPSPoint( @@ -1180,7 +1172,6 @@ def test_interpolator_preserves_gps_point_type(self): def test_interpolator_preserves_camm_gps_point_type(self): """Test that Interpolator preserves CAMMGPSPoint type.""" - from mapillary_tools.telemetry import CAMMGPSPoint track = [ CAMMGPSPoint( diff --git a/tests/unit/test_gpmf_gps_filter.py b/tests/unit/test_gpmf_gps_filter.py index c86b5c760..c15ef40e5 100644 --- a/tests/unit/test_gpmf_gps_filter.py +++ b/tests/unit/test_gpmf_gps_filter.py @@ -8,7 +8,6 @@ import statistics import pytest - from mapillary_tools.geo import Point from mapillary_tools.gpmf import gps_filter from mapillary_tools.gpmf.gpmf_gps_filter import remove_noisy_points, remove_outliers diff --git a/tests/unit/test_gpmf_parser.py b/tests/unit/test_gpmf_parser.py index bae30ce11..f8ac7e944 100644 --- a/tests/unit/test_gpmf_parser.py +++ b/tests/unit/test_gpmf_parser.py @@ -5,10 +5,10 @@ import datetime import os +import struct from pathlib import Path import pytest - from mapillary_tools import telemetry from mapillary_tools.gpmf import gpmf_parser @@ -331,8 +331,6 @@ def _build_gps9_sample_bytes( self, lat, lon, alt, speed2d, speed3d, days, secs_ms, dop, fix ): """Encode raw GPS9 values as bytes using the 'lllllllSS' format.""" - import struct - return struct.pack( ">iiiiiiiHH", lat, @@ -566,8 +564,6 @@ def test_no_strm_key(self): def test_gps9_preferred_over_gps5(self): """GPS9 is tried first within each STRM; GPS5 is fallback.""" - import struct - sample_bytes = struct.pack( ">iiiiiiiHH", 510776007, diff --git a/tests/unit/test_gpx_serializer.py b/tests/unit/test_gpx_serializer.py index 7744d548e..3a52cad53 100644 --- a/tests/unit/test_gpx_serializer.py +++ b/tests/unit/test_gpx_serializer.py @@ -11,12 +11,7 @@ from mapillary_tools.geo import Point from mapillary_tools.serializer.gpx import GPXSerializer from mapillary_tools.telemetry import CAMMGPSPoint, GPSFix, GPSPoint -from mapillary_tools.types import ( - ErrorMetadata, - FileType, - ImageMetadata, - VideoMetadata, -) +from mapillary_tools.types import ErrorMetadata, FileType, ImageMetadata, VideoMetadata def _make_image( diff --git a/tests/unit/test_history.py b/tests/unit/test_history.py index a64f5aafb..dd0783768 100644 --- a/tests/unit/test_history.py +++ b/tests/unit/test_history.py @@ -9,7 +9,6 @@ from unittest.mock import patch import pytest - from mapillary_tools import history, types diff --git a/tests/unit/test_http.py b/tests/unit/test_http.py index 9e5b5c42e..f23676732 100644 --- a/tests/unit/test_http.py +++ b/tests/unit/test_http.py @@ -6,7 +6,6 @@ from unittest.mock import MagicMock import requests - from mapillary_tools import http diff --git a/tests/unit/test_ipc.py b/tests/unit/test_ipc.py index 190421de5..37458e6ee 100644 --- a/tests/unit/test_ipc.py +++ b/tests/unit/test_ipc.py @@ -8,7 +8,6 @@ from unittest.mock import patch import pytest - from mapillary_tools import ipc diff --git a/tests/unit/test_sample_video.py b/tests/unit/test_sample_video.py index e92aef4a8..0743eeb42 100644 --- a/tests/unit/test_sample_video.py +++ b/tests/unit/test_sample_video.py @@ -15,7 +15,6 @@ import py.path import pytest - from mapillary_tools import ( exceptions, exif_read,