mywhoosh2garmin/tests/fit/test_rewriter_patching.py

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,
    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_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 _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_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)