test: verify FIT binary preservation

tests/fit/builders.py

@@ -18,15 +18,43 @@ def make_fit(data_records: bytes, *, header_size: int = 14) -> bytes:
     return bytes(body + struct.pack("<H", fit_crc(body)))
 
 
-def definition(local: int, global_num: int, fields: list[tuple[int, int, int]], *, endian: str = "<") -> bytes:
+def definition(
+    local: int,
+    global_num: int,
+    fields: list[tuple[int, int, int]],
+    *,
+    endian: str = "<",
+    developer_fields: list[tuple[int, int, int]] | None = None,
+) -> bytes:
+    """Build a FIT definition message.
+
+    ``developer_fields`` is an optional list of (field_num, size, developer_data_index)
+    triples. When provided (and non-empty), the record header's "has developer fields"
+    bit (0x20) is set and a developer field definition block is appended after the
+    native field definitions, per the FIT format.
+    """
     architecture = 1 if endian == ">" else 0
+    header_byte = 0x40 | local
+    if developer_fields:
+        header_byte |= 0x20
     payload = bytearray([0, architecture])
     payload.extend(struct.pack(f"{endian}H", global_num))
     payload.append(len(fields))
     for num, size, base_type in fields:
         payload.extend(bytes([num, size, base_type]))
-    return bytes([0x40 | local]) + bytes(payload)
+    if developer_fields:
+        payload.append(len(developer_fields))
+        for num, size, dev_index in developer_fields:
+            payload.extend(bytes([num, size, dev_index]))
+    return bytes([header_byte]) + bytes(payload)
 
 
 def data(local: int, payload: bytes) -> bytes:
     return bytes([local]) + payload
+
+
+def compressed_timestamp_data(local: int, time_offset: int, payload: bytes) -> bytes:
+    """Build a compressed-timestamp data record header (bit 0x80 set, local message
+    type in bits 5-6, time offset in bits 0-4) followed by the record payload."""
+    header = 0x80 | ((local & 0x03) << 5) | (time_offset & 0x1F)
+    return bytes([header]) + payload

tests/fit/test_rewriter_preservation.py

@@ -0,0 +1,187 @@
+import struct
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from app.fit.rewriter import FitFormatError, convert_fit_device, is_fit_file, read_device_field_values
+from tests.fit.builders import compressed_timestamp_data, data, definition, make_fit
+
+FILE_ID_MESG_NUM = 0
+DEVICE_INFO_MESG_NUM = 23
+RECORD_MESG_NUM = 20
+HEADER_SIZE = 14
+
+
+@dataclass(frozen=True)
+class ComplexFixture:
+    fit_bytes: bytes
+    metadata_offsets: set[int]
+    file_id_manufacturer_offset: int
+    file_id_product_offset: int
+    creator_manufacturer_offset: int
+    creator_product_offset: int
+    preserved_ranges: tuple[tuple[int, int], ...]
+
+
+def _build_complex_fixture() -> ComplexFixture:
+    """Builds one synthetic FIT data section covering all of the advanced record
+    shapes the parser must handle end-to-end through convert_fit_device():
+
+    1. a normal file_id definition/data pair (local 0);
+    2. a device_info definition with a creator record (local 1, device_index=0);
+    3. a record definition with one native field and one developer field, plus one
+       data record using it (local 2);
+    4. a compressed-timestamp data header referring to that same local 2 definition;
+    5. a later replacement definition for local message number 2 (redefined as a
+       non-creator device_info record, device_index=2), plus a data record using it.
+
+    All byte offsets are derived here from the construction arithmetic itself
+    (running length of the accumulated byte stream) -- this function never calls
+    into app.fit.rewriter's parser.
+    """
+    records = bytearray()
+
+    def pos() -> int:
+        return HEADER_SIZE + len(records)
+
+    # 1. normal file_id definition/data pair.
+    records.extend(definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)]))
+
+    file_id_data_start = pos()
+    records.extend(data(0, struct.pack("<HH", 255, 999)))
+    file_id_manufacturer_offset = file_id_data_start + 1  # +1 for the record header byte
+    file_id_product_offset = file_id_manufacturer_offset + 2  # manufacturer is a u16
+
+    # 2. device_info definition with a creator record (device_index == 0).
+    records.extend(definition(1, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84)]))
+
+    creator_data_start = pos()
+    records.extend(data(1, struct.pack("<BHH", 0, 255, 999)))
+    creator_manufacturer_offset = creator_data_start + 1 + 1  # header byte + device_index(u8)
+    creator_product_offset = creator_manufacturer_offset + 2  # manufacturer is a u16
+
+    # 3. record definition with one native field (power) and one developer field,
+    #    plus one data record using it.
+    records.extend(definition(2, RECORD_MESG_NUM, [(7, 2, 0x84)], developer_fields=[(0, 4, 0)]))
+
+    record_data_start = pos()
+    records.extend(data(2, struct.pack("<H", 1234) + b"\xDE\xAD\xBE\xEF"))
+    record_data_end = pos()
+
+    # 4. compressed-timestamp data header referring to the still-active local
+    #    definition 2 (the record definition above). Same field layout, so the
+    #    payload is a power u16 followed by the 4-byte developer field.
+    compressed_start = pos()
+    records.extend(compressed_timestamp_data(2, 17, struct.pack("<H", 5678) + b"\xCA\xFE\xBA\xBE"))
+    compressed_end = pos()
+
+    # 5. later replacement definition for the same local message number (2): local 2
+    #    is redefined as a *non-creator* device_info record (device_index=2), so the
+    #    patcher must skip it entirely and its bytes must stay identical.
+    records.extend(definition(2, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84)]))
+
+    sensor_data_start = pos()
+    records.extend(data(2, struct.pack("<BHH", 2, 77, 555)))
+    sensor_data_end = pos()
+
+    metadata_offsets: set[int] = set()
+    metadata_offsets.update(range(file_id_manufacturer_offset, file_id_manufacturer_offset + 2))
+    metadata_offsets.update(range(file_id_product_offset, file_id_product_offset + 2))
+    metadata_offsets.update(range(creator_manufacturer_offset, creator_manufacturer_offset + 2))
+    metadata_offsets.update(range(creator_product_offset, creator_product_offset + 2))
+
+    return ComplexFixture(
+        fit_bytes=make_fit(bytes(records)),
+        metadata_offsets=metadata_offsets,
+        file_id_manufacturer_offset=file_id_manufacturer_offset,
+        file_id_product_offset=file_id_product_offset,
+        creator_manufacturer_offset=creator_manufacturer_offset,
+        creator_product_offset=creator_product_offset,
+        preserved_ranges=(
+            (record_data_start, record_data_end),
+            (compressed_start, compressed_end),
+            (sensor_data_start, sensor_data_end),
+        ),
+    )
+
+
+# Built once at import time so the fixture bytes and the expected-offsets bookkeeping
+# can never drift apart from each other.
+_FIXTURE = _build_complex_fixture()
+
+
+@pytest.fixture
+def complex_fit_bytes() -> bytes:
+    return _FIXTURE.fit_bytes
+
+
+def find_expected_device_metadata_offsets(before: bytes) -> set[int]:
+    """Byte offsets convert_fit_device() is expected to touch, taken from the
+    fixture's own construction-time bookkeeping (see _build_complex_fixture above).
+    Deliberately does not parse `before` or call any part of app.fit.rewriter, so
+    this test cannot become circular (the parser being tested against itself)."""
+    del before
+    return set(_FIXTURE.metadata_offsets)
+
+
+def test_only_target_fields_and_crcs_change(tmp_path: Path, complex_fit_bytes: bytes) -> None:
+    source = tmp_path / "source.fit"
+    output = tmp_path / "output.fit"
+    source.write_bytes(complex_fit_bytes)
+
+    convert_fit_device(source, output)
+
+    before = source.read_bytes()
+    after = output.read_bytes()
+    assert len(before) == len(after)
+
+    changed = {index for index, (a, b) in enumerate(zip(before, after)) if a != b}
+    expected_metadata_offsets = set(find_expected_device_metadata_offsets(before))
+    crc_offsets = {12, 13, len(before) - 2, len(before) - 1}
+    assert changed <= expected_metadata_offsets | crc_offsets
+
+
+def test_complex_fixture_patches_targets_and_preserves_advanced_records(
+    tmp_path: Path, complex_fit_bytes: bytes
+) -> None:
+    source = tmp_path / "source.fit"
+    output = tmp_path / "output.fit"
+    source.write_bytes(complex_fit_bytes)
+
+    result = convert_fit_device(source, output)
+
+    assert is_fit_file(output) is True
+    assert result.patched_field_count == 4
+
+    after = output.read_bytes()
+    assert struct.unpack_from("<H", after, _FIXTURE.file_id_manufacturer_offset)[0] == 1
+    assert struct.unpack_from("<H", after, _FIXTURE.file_id_product_offset)[0] == 3570
+    assert struct.unpack_from("<H", after, _FIXTURE.creator_manufacturer_offset)[0] == 1
+    assert struct.unpack_from("<H", after, _FIXTURE.creator_product_offset)[0] == 3570
+
+    # The record message (developer field), the compressed-timestamp record reusing
+    # its definition, and the redefined local-2 sensor device_info record are all
+    # untouched by patching -- verify their raw payload bytes are byte-identical.
+    for start, end in _FIXTURE.preserved_ranges:
+        assert complex_fit_bytes[start:end] == after[start:end]
+
+    values = read_device_field_values(output)
+    assert any(
+        v.global_message_num == FILE_ID_MESG_NUM and v.field_num == 2 and v.value == 3570 for v in values
+    )
+
+
+def test_truncated_definition_is_non_recoverable(tmp_path: Path) -> None:
+    path = tmp_path / "truncated.fit"
+    path.write_bytes(make_fit(bytes([0x40, 0x00, 0x00])))
+    assert is_fit_file(path) is False
+
+
+def test_convert_fit_device_rejects_truncated_definition(tmp_path: Path) -> None:
+    source = tmp_path / "truncated.fit"
+    source.write_bytes(make_fit(bytes([0x40, 0x00, 0x00])))
+    output = tmp_path / "output.fit"
+
+    with pytest.raises(FitFormatError):
+        convert_fit_device(source, output)