ef9ca9eeca
Final-review fixes for Plan 2 (fit-rewriter). Every failure mode below now surfaces as FitFormatError so Plan 3 can classify invalid FIT input as a non-retryable activity error (spec 10.4). - Range-check numeric values against the field's declared size before struct.pack, so an oversized serial number or a 1-byte product field raises FitFormatError instead of leaking a raw struct.error. - Reject zero-size field definitions during parsing. A zero-size device_info field 0 read back as device_index == 0 via int.from_bytes(b"", ...), which could have let a paired sensor be rewritten as an Edge 1030 Plus (spec 10.2). - Add DeviceFieldValue.is_creator so callers can tell the creator device_info record from sensor records instead of silently keeping whichever record appeared last. - Implement the missing spec 10.4 post-patch step: read the patched buffer back and verify file_id 1/2/8 and creator device_info 2/4/27 hold the target values. A field that could not be written (e.g. a product_name field too small for the target string) now fails the whole conversion rather than producing a silent partial patch. Verification runs before the output is written, so a half-rewritten file never lands on disk. - Use the field's actual endianness in _read_field_value's fallback path. - Add curated re-exports in app/fit/__init__.py for Plan 3. - Document _iter_data_fields' caller invariant (validate the container first; end_offset is not clamped). - Extend the preservation fixture with a product_name string field so the zero-filling string write path is covered by the byte-preservation proof, and test convert_fit_device against a 12-byte header. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
346 lines
14 KiB
Python
346 lines
14 KiB
Python
import struct
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.fit.models import DeviceFieldValue, FitConversionResult, GarminDevice
|
|
from app.fit.rewriter import (
|
|
FitFormatError,
|
|
_iter_data_fields,
|
|
convert_fit_device,
|
|
is_fit_file,
|
|
read_device_field_values,
|
|
)
|
|
from tests.fit.builders import data, definition, make_fit
|
|
|
|
FILE_ID_MESG_NUM = 0
|
|
DEVICE_INFO_MESG_NUM = 23
|
|
|
|
|
|
def _build_fixture() -> bytes:
|
|
# file_id: manufacturer(1/u16), product(2/u16)
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)])
|
|
file_data = data(0, struct.pack("<HH", 255, 999))
|
|
|
|
# device_info: device_index(0/u8), manufacturer(2/u16), product(4/u16)
|
|
device_def = definition(1, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84)])
|
|
creator = data(1, struct.pack("<BHH", 0, 255, 999))
|
|
sensor = data(1, struct.pack("<BHH", 1, 32, 1234))
|
|
|
|
records = file_def + file_data + device_def + creator + sensor
|
|
return make_fit(records)
|
|
|
|
|
|
def _build_big_endian_fixture() -> bytes:
|
|
"""Big-endian (architecture=1) file_id definition/data pair. Closes the
|
|
big-endian coverage gap deferred from Task 2: no fixture anywhere in the FIT
|
|
test suite previously passed endian=">" to definition()/struct.pack, so the
|
|
architecture-byte branch in _read_definition (and the endian format string it
|
|
threads into every field read/write) was never actually exercised."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)], endian=">")
|
|
file_data = data(0, struct.pack(">HH", 255, 999))
|
|
return make_fit(file_def + file_data)
|
|
|
|
|
|
def _build_no_device_index_fixture() -> bytes:
|
|
"""device_info record entirely missing field 0 (device_index) must be left untouched."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)])
|
|
file_data = data(0, struct.pack("<HH", 255, 999))
|
|
|
|
# device_info WITHOUT device_index field at all: manufacturer(2/u16), product(4/u16)
|
|
device_def = definition(1, DEVICE_INFO_MESG_NUM, [(2, 2, 0x84), (4, 2, 0x84)])
|
|
no_index_device = data(1, struct.pack("<HH", 32, 1234))
|
|
|
|
records = file_def + file_data + device_def + no_index_device
|
|
return make_fit(records)
|
|
|
|
|
|
def _build_product_name_fixture() -> bytes:
|
|
"""file_id field 8 and device_info field 27 are both product_name (string) fields —
|
|
the only string field type this task patches. No sensor record here, so the single
|
|
device_info record present is unambiguously the creator."""
|
|
# file_id: manufacturer(1/u16), product(2/u16), product_name(8/string, size 20)
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84), (8, 20, 0x07)])
|
|
original_file_name = b"MyWhoosh\x00".ljust(20, b"\x00")
|
|
file_data = data(0, struct.pack("<HH", 255, 999) + original_file_name)
|
|
|
|
# device_info: device_index(0/u8), manufacturer(2/u16), product(4/u16), product_name(27/string, size 20)
|
|
device_def = definition(
|
|
1, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84), (27, 20, 0x07)]
|
|
)
|
|
original_device_name = b"MyWhoosh App\x00".ljust(20, b"\x00")
|
|
creator = data(1, struct.pack("<BHH", 0, 255, 999) + original_device_name)
|
|
|
|
records = file_def + file_data + device_def + creator
|
|
return make_fit(records)
|
|
|
|
|
|
def _build_zero_size_device_index_fixture() -> bytes:
|
|
"""Adversarial file declaring device_info field 0 (device_index) with size 0.
|
|
|
|
Without an explicit rejection, `int.from_bytes(b"", "little") == 0` would make
|
|
every record on this definition look like the creator (device_index == 0) and a
|
|
paired sensor would be rewritten as an Edge 1030 Plus."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)])
|
|
file_data = data(0, struct.pack("<HH", 255, 999))
|
|
|
|
device_def = definition(1, DEVICE_INFO_MESG_NUM, [(0, 0, 0x02), (2, 2, 0x84), (4, 2, 0x84)])
|
|
sensor = data(1, struct.pack("<HH", 32, 1234))
|
|
|
|
return make_fit(file_def + file_data + device_def + sensor)
|
|
|
|
|
|
def _build_undersized_serial_fixture() -> bytes:
|
|
"""file_id with a 2-byte serial_number field (3/u16) -- too small to hold a
|
|
32-bit serial number handed in through the public API."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84), (3, 2, 0x84)])
|
|
file_data = data(0, struct.pack("<HHH", 255, 999, 4242))
|
|
return make_fit(file_def + file_data)
|
|
|
|
|
|
def _build_undersized_product_fixture() -> bytes:
|
|
"""file_id declaring product (field 2) as a single byte, which cannot hold 3570."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 1, 0x02)])
|
|
file_data = data(0, struct.pack("<HB", 255, 99))
|
|
return make_fit(file_def + file_data)
|
|
|
|
|
|
def _build_undersized_product_name_fixture() -> bytes:
|
|
"""file_id product_name field of size 10 -- too small for "Edge 1030 Plus"
|
|
(14 bytes plus a null terminator), so the string write silently no-ops."""
|
|
file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84), (8, 10, 0x07)])
|
|
file_data = data(0, struct.pack("<HH", 255, 999) + b"MyWhoosh\x00\x00")
|
|
return make_fit(file_def + file_data)
|
|
|
|
|
|
def _read_device_info_records(path: Path) -> list[dict[int, int]]:
|
|
"""Parse raw bytes and return one dict of {field_num: value} per device_info record,
|
|
in file order, so creator and sensor records can be distinguished positionally."""
|
|
raw = bytearray(path.read_bytes())
|
|
records: list[dict[int, int]] = []
|
|
for local_def, field_offsets in _iter_data_fields(raw):
|
|
if local_def.global_message_num != DEVICE_INFO_MESG_NUM:
|
|
continue
|
|
values: dict[int, int] = {}
|
|
for field, offset in field_offsets:
|
|
raw_bytes = bytes(raw[offset : offset + field.size])
|
|
if field.size == 1:
|
|
values[field.num] = raw_bytes[0]
|
|
elif field.size == 2:
|
|
values[field.num] = struct.unpack(f"{local_def.endian}H", raw_bytes)[0]
|
|
records.append(values)
|
|
return records
|
|
|
|
|
|
def test_convert_fit_device_patches_file_id(tmp_path: Path) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
result = convert_fit_device(source, output)
|
|
|
|
assert isinstance(result, FitConversionResult)
|
|
assert result.source_path == source
|
|
assert result.output_path == output
|
|
assert result.file_crc is not None
|
|
assert result.patched_field_count > 0
|
|
|
|
values = read_device_field_values(output)
|
|
values_by_key = {(v.global_message_num, v.field_num): v.value for v in values}
|
|
|
|
assert values_by_key[(FILE_ID_MESG_NUM, 1)] == 1
|
|
assert values_by_key[(FILE_ID_MESG_NUM, 2)] == 3570
|
|
|
|
|
|
def test_convert_fit_device_patches_creator_and_leaves_sensor_untouched(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
convert_fit_device(source, output)
|
|
|
|
device_info_records = _read_device_info_records(output)
|
|
assert len(device_info_records) == 2
|
|
creator_values, sensor_values = device_info_records
|
|
|
|
# creator (device_index=0) patched to Edge 1030 Plus
|
|
assert creator_values[0] == 0
|
|
assert creator_values[2] == 1
|
|
assert creator_values[4] == 3570
|
|
|
|
# sensor (device_index=1) must remain completely untouched
|
|
assert sensor_values[0] == 1
|
|
assert sensor_values[2] == 32
|
|
assert sensor_values[4] == 1234
|
|
|
|
|
|
def test_convert_fit_device_leaves_device_info_without_device_index_untouched(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_no_device_index_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
convert_fit_device(source, output)
|
|
|
|
device_info_records = _read_device_info_records(output)
|
|
assert len(device_info_records) == 1
|
|
values = device_info_records[0]
|
|
|
|
# device_info without a device_index field must remain completely untouched
|
|
assert values[2] == 32
|
|
assert values[4] == 1234
|
|
|
|
|
|
def test_convert_fit_device_patches_product_name_string_fields(tmp_path: Path) -> None:
|
|
"""file_id field 8 and device_info field 27 (creator) are both string fields, written via
|
|
_write_field_value's string branch (buffer sizing + null-terminator handling). Neither is
|
|
exercised by the numeric-only base fixture used elsewhere in this module."""
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_product_name_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
convert_fit_device(source, output)
|
|
|
|
values = read_device_field_values(output)
|
|
values_by_key = {(v.global_message_num, v.field_num): v.value for v in values}
|
|
|
|
assert values_by_key[(FILE_ID_MESG_NUM, 8)] == "Edge 1030 Plus"
|
|
assert values_by_key[(DEVICE_INFO_MESG_NUM, 27)] == "Edge 1030 Plus"
|
|
|
|
|
|
def test_convert_fit_device_round_trips_big_endian_fields(tmp_path: Path) -> None:
|
|
"""Proves both the read side (parsing manufacturer/product under architecture=1)
|
|
and the write side (patching them back in big-endian byte order) are correct --
|
|
a byte-order bug on either side would flip 255/999 or the patched 1/3570 into an
|
|
unrelated value once read back with the (still big-endian) definition."""
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_big_endian_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
convert_fit_device(source, output)
|
|
|
|
values = read_device_field_values(output)
|
|
values_by_key = {(v.global_message_num, v.field_num): v.value for v in values}
|
|
|
|
assert values_by_key[(FILE_ID_MESG_NUM, 1)] == 1
|
|
assert values_by_key[(FILE_ID_MESG_NUM, 2)] == 3570
|
|
|
|
|
|
def test_convert_fit_device_defaults_to_edge_1030_plus() -> None:
|
|
device = GarminDevice()
|
|
assert device.manufacturer_id == 1
|
|
assert device.product_id == 3570
|
|
assert device.product_name == "Edge 1030 Plus"
|
|
|
|
|
|
def test_convert_fit_device_rejects_invalid_container(tmp_path: Path) -> None:
|
|
source = tmp_path / "bad.fit"
|
|
source.write_bytes(b"not a fit file")
|
|
output = tmp_path / "output.fit"
|
|
|
|
with pytest.raises(FitFormatError):
|
|
convert_fit_device(source, output)
|
|
|
|
|
|
def test_read_device_field_values_returns_device_field_value_instances(tmp_path: Path) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_fixture())
|
|
|
|
values = read_device_field_values(source)
|
|
|
|
assert all(isinstance(v, DeviceFieldValue) for v in values)
|
|
assert any(v.global_message_num == FILE_ID_MESG_NUM for v in values)
|
|
|
|
|
|
def test_zero_size_field_definition_is_rejected(tmp_path: Path) -> None:
|
|
"""A zero-size field is illegal FIT and must be rejected during parsing, so a
|
|
zero-size device_index can never make a sensor record read as the creator."""
|
|
source = tmp_path / "zero-size.fit"
|
|
source.write_bytes(_build_zero_size_device_index_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
assert is_fit_file(source) is False
|
|
with pytest.raises(FitFormatError):
|
|
convert_fit_device(source, output)
|
|
assert not output.exists()
|
|
|
|
|
|
def test_value_too_large_for_declared_field_size_raises_fit_format_error(tmp_path: Path) -> None:
|
|
"""A 32-bit serial number against a 2-byte serial field must surface as
|
|
FitFormatError, not a raw struct.error escaping the public API."""
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_undersized_serial_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
with pytest.raises(FitFormatError):
|
|
convert_fit_device(source, output, GarminDevice(serial_number=4294967295))
|
|
assert not output.exists()
|
|
|
|
|
|
def test_product_id_too_large_for_one_byte_field_raises_fit_format_error(tmp_path: Path) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_undersized_product_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
with pytest.raises(FitFormatError):
|
|
convert_fit_device(source, output)
|
|
assert not output.exists()
|
|
|
|
|
|
def test_unwritable_product_name_fails_conversion(tmp_path: Path) -> None:
|
|
"""A product_name field too small for the target string used to be silently left
|
|
unpatched while manufacturer/product reported success. The post-patch read-back
|
|
verification must turn that partial patch into a controlled failure."""
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_undersized_product_name_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
with pytest.raises(FitFormatError):
|
|
convert_fit_device(source, output)
|
|
assert not output.exists()
|
|
|
|
|
|
def test_read_device_field_values_marks_creator_records(tmp_path: Path) -> None:
|
|
"""The creator device_info and every file_id field are is_creator=True; a paired
|
|
sensor's device_info fields are is_creator=False, so a caller building a
|
|
{(mesg, field): value} dict can no longer be shadowed by sensor values."""
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_fixture())
|
|
output = tmp_path / "output.fit"
|
|
|
|
convert_fit_device(source, output)
|
|
|
|
values = read_device_field_values(output)
|
|
|
|
assert all(v.is_creator for v in values if v.global_message_num == FILE_ID_MESG_NUM)
|
|
|
|
creator = {
|
|
(v.global_message_num, v.field_num): v.value
|
|
for v in values
|
|
if v.global_message_num == DEVICE_INFO_MESG_NUM and v.is_creator
|
|
}
|
|
sensor = {
|
|
(v.global_message_num, v.field_num): v.value
|
|
for v in values
|
|
if v.global_message_num == DEVICE_INFO_MESG_NUM and not v.is_creator
|
|
}
|
|
|
|
assert creator[(DEVICE_INFO_MESG_NUM, 2)] == 1
|
|
assert creator[(DEVICE_INFO_MESG_NUM, 4)] == 3570
|
|
assert sensor[(DEVICE_INFO_MESG_NUM, 2)] == 32
|
|
assert sensor[(DEVICE_INFO_MESG_NUM, 4)] == 1234
|
|
|
|
|
|
def test_device_info_without_device_index_is_not_creator(tmp_path: Path) -> None:
|
|
source = tmp_path / "source.fit"
|
|
source.write_bytes(_build_no_device_index_fixture())
|
|
|
|
values = read_device_field_values(source)
|
|
|
|
device_info_values = [v for v in values if v.global_message_num == DEVICE_INFO_MESG_NUM]
|
|
assert device_info_values
|
|
assert all(v.is_creator is False for v in device_info_values)
|