miband-bot/mi-fitness-python/tests/unit/test_models.py

"""测试数据模型 (models.py)。

覆盖：模型构造、字段解析、响应包装器的属性方法。
"""

from __future__ import annotations

import json

from mi_fitness.models import (
    AggregatedDataItem,
    AggregatedDataResponse,
    AuthToken,
    BloodPressureData,
    CaloriesData,
    CheckNewMsgResponse,
    DeleteRelativeResponse,
    FamilyMember,
    FamilyMemberResponse,
    GoalData,
    GoalMetric,
    HeartRateData,
    IntensityData,
    InviteMessage,
    InviteResponse,
    InviteUniqueIdResponse,
    LatestDataItem,
    LatestDataResponse,
    LatestDataSnapshot,
    LatestHeartRate,
    MessageListResponse,
    OperateInviteResponse,
    RelativeListResponse,
    SharedDataTypesResponse,
    SleepData,
    SleepSegment,
    Spo2Data,
    Spo2SummaryData,
    StepData,
    ValidStandData,
    VerifiedUserInfo,
    VerifyUserResponse,
    WeightData,
)


# region 基础模型测试
class TestAuthToken:
    """AuthToken 模型测试。"""

    def test_defaults(self) -> None:
        """空构造应全部默认空串。"""
        t = AuthToken()
        assert t.user_id == ""
        assert t.ssecurity == ""
        assert t.service_token == ""

    def test_serialization_roundtrip(self) -> None:
        """JSON 序列化往返。"""
        t = AuthToken(user_id="123", ssecurity="abc")
        data = t.model_dump_json()
        t2 = AuthToken.model_validate_json(data)
        assert t2.user_id == t.user_id
        assert t2.ssecurity == t.ssecurity


class TestFamilyMember:
    """FamilyMember 模型测试。"""

    def test_construction(self) -> None:
        m = FamilyMember(relative_uid=123, relative_note="妈妈")
        assert m.relative_uid == 123
        assert m.relative_note == "妈妈"
        assert m.latest_data_time == 0

    def test_from_dict(self) -> None:
        data = {
            "relative_uid": 9999,
            "relative_note": "爸爸",
            "relative_icon": "https://img.example.com/a.jpg",
            "latest_data_time": 1717488000,
            "latest_abnormal_record_time": 0,
            "source_tag": 1,
        }
        m = FamilyMember(**data)
        assert m.relative_uid == 9999
        assert m.source_tag == 1


class TestLatestHeartRate:
    def test_defaults(self) -> None:
        hr = LatestHeartRate()
        assert hr.bpm == 0
        assert hr.time == 0


class TestHeartRateData:
    def test_full_construction(self) -> None:
        hr = HeartRateData(
            time=1717488000,
            avg_hr=72,
            max_hr=120,
            min_hr=55,
            latest_hr=LatestHeartRate(bpm=75, time=1717488000),
        )
        assert hr.avg_hr == 72
        assert hr.latest_hr is not None
        assert hr.latest_hr.bpm == 75


class TestSleepData:
    def test_with_segments(self) -> None:
        seg = SleepSegment(bedtime=100, wake_up_time=200, duration=100)
        sd = SleepData(
            time=1717488000,
            total_duration=480,
            sleep_score=85,
            segment_details=[seg],
        )
        assert sd.total_duration == 480
        assert len(sd.segment_details) == 1
        assert sd.segment_details[0].duration == 100


class TestStepData:
    def test_construction(self) -> None:
        s = StepData(time=1717488000, steps=8500, distance=6200, calories=320)
        assert s.steps == 8500


class TestWeightData:
    def test_construction(self) -> None:
        w = WeightData(time=1717488000, weight=65.5, bmi=22.1)
        assert w.weight == 65.5
        assert w.bmi == 22.1


class TestVerifiedUserInfo:
    def test_alias_user_id(self) -> None:
        """userId 别名应映射到 user_id。"""
        info = VerifiedUserInfo(**{"userId": 12345, "nickname": "测试", "icon": "url"})
        assert info.user_id == 12345
        assert info.nickname == "测试"

    def test_populate_by_name(self) -> None:
        """也可以用 user_id 直接构造。"""
        info = VerifiedUserInfo(user_id=999, nickname="直接")  # type: ignore[call-arg]
        assert info.user_id == 999


# endregion


# region LatestDataItem 测试
class TestLatestDataItem:
    def test_parse_json_value(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="heart_rate",
            value='{"avg_hr": 72}',
        )
        parsed = item.parse_value()
        assert isinstance(parsed, dict)
        assert parsed["avg_hr"] == 72

    def test_parse_numeric_value(self) -> None:
        item = LatestDataItem(time=1717488000, key="goal", value=10000)
        parsed = item.parse_value()
        assert parsed == 10000

    def test_parse_invalid_json(self) -> None:
        item = LatestDataItem(time=1717488000, key="bad", value="not{json")
        parsed = item.parse_value()
        assert parsed == {}

    def test_parse_dict_value(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="weight",
            value={"weight": 65.5},  # type: ignore[arg-type]
        )
        parsed = item.parse_value()
        assert parsed == {"weight": 65.5}

    def test_parse_json_list_returns_empty_dict(self) -> None:
        item = LatestDataItem(time=1717488000, key="bad", value='["unexpected"]')
        assert item.parse_value() == {}

    def test_as_goal(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="goal",
            value={
                "date_time": 1717488000,
                "goal_items": [
                    {"field": 1, "target_value": 6000, "achieved_value": 3716},
                ],
            },  # type: ignore[arg-type]
        )
        goal = item.as_goal()
        assert isinstance(goal, GoalData)
        assert goal.time == 1717488000
        assert len(goal.goal_items) == 1
        assert goal.goal_items[0].target_value == 6000
        assert goal.goal_items[0].metric == GoalMetric.STEPS
        assert goal.goal_items[0].metric_key == "steps"
        assert goal.goal_items[0].metric_label == "步数"

    def test_goal_accessors_and_unknown_items(self) -> None:
        goal = GoalData.model_validate(
            {
                "time": 1717488000,
                "goal_items": [
                    {"field": 2, "target_value": 400, "achieved_value": 347},
                    {"field": 1, "target_value": 6000, "achieved_value": 6203},
                    {"field": 4, "target_value": 30, "achieved_value": 27},
                    {"field": 99, "target_value": 1, "achieved_value": 0},
                ],
            }
        )

        assert goal.calories_goal is not None
        assert goal.calories_goal.target_value == 400
        assert goal.steps_goal is not None
        assert goal.steps_goal.achieved_value == 6203
        assert goal.intensity_goal is not None
        assert goal.intensity_goal.target_value == 30
        assert goal.available_metrics == [
            GoalMetric.CALORIES,
            GoalMetric.STEPS,
            GoalMetric.INTENSITY,
        ]
        assert len(goal.unknown_goal_items) == 1
        assert goal.unknown_goal_items[0].metric is None
        assert goal.unknown_goal_items[0].metric_key == "unknown:99"
        assert goal.get_item(GoalMetric.STEPS) is goal.steps_goal

    def test_as_heart_rate(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="heart_rate",
            value='{"time":1717491600,"bpm":84}',
        )
        heart_rate = item.as_heart_rate()
        assert isinstance(heart_rate, LatestHeartRate)
        assert heart_rate.time == 1717491600
        assert heart_rate.bpm == 84

    def test_as_heart_rate_returns_none_for_invalid_payload(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="heart_rate",
            value='{"time":1717491600,"bpm":{"bad":1}}',
        )
        assert item.as_heart_rate() is None

    def test_as_sleep_uses_date_time_alias(self) -> None:
        item = LatestDataItem(
            time=1717488000,
            key="sleep",
            value='{"date_time":1717488000,"total_duration":436,"sleep_score":86}',
        )
        sleep = item.as_sleep()
        assert isinstance(sleep, SleepData)
        assert sleep.time == 1717488000
        assert sleep.total_duration == 436
        assert sleep.sleep_score == 86

    def test_as_steps_and_latest_metrics(self) -> None:
        steps_item = LatestDataItem(
            time=1717488000,
            key="steps",
            value='{"date_time":1717488000,"steps":3716,"distance":2193,"calories":143,"goal":6000}',
        )
        calories_item = LatestDataItem(
            time=1717488000,
            key="calories",
            value='{"date_time":1717488000,"calories":230,"goal":300}',
        )
        stand_item = LatestDataItem(
            time=1717488000,
            key="valid_stand",
            value='{"date_time":1717488000,"count":7}',
        )
        intensity_item = LatestDataItem(
            time=1717488000,
            key="intensity",
            value='{"date_time":1717488000,"duration":15}',
        )
        spo2_item = LatestDataItem(
            time=1717488000,
            key="spo2",
            value='{"time":1717495200,"spo2":96}',
        )

        steps = steps_item.as_steps()
        calories = calories_item.as_calories()
        stand = stand_item.as_valid_stand()
        intensity = intensity_item.as_intensity()
        spo2 = spo2_item.as_spo2()

        assert isinstance(steps, StepData)
        assert steps.goal == 6000
        assert isinstance(calories, CaloriesData)
        assert calories.goal == 300
        assert isinstance(stand, ValidStandData)
        assert stand.count == 7
        assert isinstance(intensity, IntensityData)
        assert intensity.duration == 15
        assert isinstance(spo2, Spo2Data)
        assert spo2.spo2 == 96

    def test_as_blood_pressure_returns_none_when_value_missing(self) -> None:
        item = LatestDataItem(time=1717488000, key="blood_pressure")
        assert item.as_blood_pressure() is None

    def test_as_blood_pressure_supports_fitness_aliases(self) -> None:
        item = LatestDataItem(
            time=1773753098,
            key="blood_pressure",
            value='{"systolic_pressure":33,"diastolic_pressure":30,"pulse":60,"time":1773753098}',
        )

        blood_pressure = item.as_blood_pressure()

        assert isinstance(blood_pressure, BloodPressureData)
        assert blood_pressure.systolic == 33
        assert blood_pressure.diastolic == 30
        assert blood_pressure.pulse == 60


# endregion


# region AggregatedDataItem 测试
class TestAggregatedDataItem:
    def test_stringify_dict_value(self) -> None:
        """dict value 应被自动转为 JSON 字符串。"""
        item = AggregatedDataItem(
            time=1717488000,
            key="heart_rate",
            value={"avg_hr": 72},  # type: ignore[arg-type]
        )
        assert isinstance(item.value, str)
        parsed = json.loads(item.value)
        assert parsed["avg_hr"] == 72

    def test_as_heart_rate(self) -> None:
        value = json.dumps(
            {
                "avg_hr": 72,
                "max_hr": 120,
                "min_hr": 55,
                "avg_rhr": 62,
                "latest_hr": {"bpm": 75, "time": 1717488000},
            }
        )
        item = AggregatedDataItem(time=1717430400, key="heart_rate", value=value)
        hr = item.as_heart_rate()
        assert isinstance(hr, HeartRateData)
        assert hr.avg_hr == 72
        assert hr.time == 1717430400
        assert hr.latest_hr is not None
        assert hr.latest_hr.bpm == 75

    def test_as_sleep(self) -> None:
        value = json.dumps(
            {
                "total_duration": 480,
                "sleep_score": 85,
                "segment_details": [
                    {"bedtime": 100, "wake_up_time": 200, "duration": 100},
                ],
            }
        )
        item = AggregatedDataItem(time=1717430400, key="sleep", value=value)
        sd = item.as_sleep()
        assert isinstance(sd, SleepData)
        assert sd.total_duration == 480
        assert len(sd.segment_details) == 1

    def test_as_sleep_ignores_invalid_segments(self) -> None:
        value = json.dumps(
            {
                "total_duration": 480,
                "segment_details": [
                    {"bedtime": 100, "wake_up_time": 200, "duration": 100},
                    "bad-segment",
                    1,
                ],
            }
        )
        item = AggregatedDataItem(time=1717430400, key="sleep", value=value)
        sd = item.as_sleep()
        assert len(sd.segment_details) == 1
        assert sd.segment_details[0].duration == 100

    def test_as_steps(self) -> None:
        value = json.dumps({"steps": 8500, "distance": 6200, "calories": 320})
        item = AggregatedDataItem(time=1717430400, key="steps", value=value)
        st = item.as_steps()
        assert isinstance(st, StepData)
        assert st.steps == 8500
        assert st.time == 1717430400

    def test_as_simple_latest_metrics(self) -> None:
        calories_item = AggregatedDataItem(
            time=1717430400, key="calories", value='{"calories":338}'
        )
        stand_item = AggregatedDataItem(time=1717430400, key="valid_stand", value='{"count":10}')
        intensity_item = AggregatedDataItem(
            time=1717430400, key="intensity", value='{"duration":19}'
        )

        calories = calories_item.as_calories()
        stand = stand_item.as_valid_stand()
        intensity = intensity_item.as_intensity()

        assert isinstance(calories, CaloriesData)
        assert calories.time == 1717430400
        assert calories.calories == 338
        assert isinstance(stand, ValidStandData)
        assert stand.count == 10
        assert isinstance(intensity, IntensityData)
        assert intensity.duration == 19

    def test_as_weight_and_blood_pressure(self) -> None:
        weight_item = AggregatedDataItem(
            time=1773753142,
            key="weight",
            value='{"time":1773753142,"weight":55.0,"bmi":16.97531}',
        )
        blood_pressure_item = AggregatedDataItem(
            time=1773753098,
            key="blood_pressure",
            value='{"systolic_pressure":33,"diastolic_pressure":30,"pulse":60,"time":1773753098}',
        )

        weight = weight_item.as_weight()
        blood_pressure = blood_pressure_item.as_blood_pressure()

        assert isinstance(weight, WeightData)
        assert weight.weight == 55.0
        assert weight.bmi == 16.97531
        assert isinstance(blood_pressure, BloodPressureData)
        assert blood_pressure.systolic == 33
        assert blood_pressure.diastolic == 30
        assert blood_pressure.pulse == 60

    def test_as_spo2_summary(self) -> None:
        value = json.dumps(
            {
                "avg_spo2": 96,
                "lack_spo2_count": 0,
                "latest_spo2": {
                    "spo2": 96,
                    "time": 1761161730,
                    "dbKey": "single_spo2",
                },
                "max_spo2": 96,
                "min_spo2": 96,
            }
        )
        item = AggregatedDataItem(time=1761091200, key="spo2", value=value)
        spo2 = item.as_spo2()

        assert isinstance(spo2, Spo2SummaryData)
        assert spo2.time == 1761091200
        assert spo2.avg_spo2 == 96
        assert spo2.latest_spo2 is not None
        assert spo2.latest_spo2.spo2 == 96


# endregion


# region 响应包装器测试
class TestRelativeListResponse:
    def test_parse_relatives(self) -> None:
        resp = RelativeListResponse(
            code=0,
            result={
                "relative_list": [
                    {"relative_uid": 111, "relative_note": "A"},
                    {"relative_uid": 222, "relative_note": "B"},
                ]
            },
        )
        members = resp.relatives
        assert len(members) == 2
        assert members[0].relative_uid == 111

    def test_empty_list(self) -> None:
        resp = RelativeListResponse(code=0, result={"relative_list": []})
        assert resp.relatives == []

    def test_skips_invalid_relatives(self) -> None:
        resp = RelativeListResponse(
            code=0,
            result={
                "relative_list": [
                    123,
                    {"relative_note": "缺少 uid"},
                    {"relative_uid": 222, "relative_note": "B"},
                ]
            },
        )
        members = resp.relatives
        assert len(members) == 1
        assert members[0].relative_uid == 222


class TestLatestDataResponse:
    def test_parse_data_items(self) -> None:
        resp = LatestDataResponse(
            code=0,
            result={
                "data_list": [
                    {"time": 100, "key": "hr", "value": "{}"},
                ]
            },
        )
        items = resp.data_items
        assert len(items) == 1
        assert items[0].key == "hr"

    def test_non_mapping_result_is_tolerated(self) -> None:
        resp = LatestDataResponse(code=0, result=None)  # type: ignore[arg-type]
        assert resp.data_items == []

    def test_snapshot_returns_typed_metrics(self) -> None:
        resp = LatestDataResponse(
            code=0,
            result={
                "data_list": [
                    {
                        "time": 1717488000,
                        "key": "heart_rate",
                        "value": '{"time":1717491600,"bpm":84}',
                    },
                    {
                        "time": 1717488000,
                        "key": "steps",
                        "value": '{"date_time":1717488000,"steps":3716,"distance":2193,"calories":143,"goal":6000}',
                    },
                    {"time": 1717488000, "key": "spo2", "value": '{"time":1717495200,"spo2":96}'},
                    {"time": 1717488000, "key": "mood", "value": '{"score":80}'},
                ],
                "latest_data_time": 1717495200,
            },
        )
        snapshot = resp.snapshot
        assert isinstance(snapshot, LatestDataSnapshot)
        assert snapshot.updated_time == 1717495200
        assert snapshot.heart_rate is not None
        assert snapshot.heart_rate.bpm == 84
        assert snapshot.steps is not None
        assert snapshot.steps.goal == 6000
        assert snapshot.spo2 is not None
        assert snapshot.spo2.spo2 == 96
        assert snapshot.extras == {"mood": {"score": 80}}
        assert snapshot.available_keys == ["heart_rate", "steps", "spo2", "mood"]

    def test_snapshot_preserves_invalid_known_payload_in_extras(self) -> None:
        resp = LatestDataResponse(
            code=0,
            result={
                "data_list": [
                    {
                        "time": 1717488000,
                        "key": "heart_rate",
                        "value": '{"time":1717491600,"bpm":{"bad":1}}',
                    },
                ],
                "latest_data_time": 1717495200,
            },
        )
        snapshot = resp.snapshot
        assert snapshot.heart_rate is None
        assert snapshot.extras == {"heart_rate": {"time": 1717491600, "bpm": {"bad": 1}}}
        assert snapshot.available_keys == ["heart_rate"]


class TestAggregatedDataResponse:
    def test_parse_data_items(self) -> None:
        resp = AggregatedDataResponse(
            code=0,
            result={
                "data_list": [
                    {
                        "sid": "test",
                        "tag": "daily_report",
                        "key": "steps",
                        "time": 100,
                        "value": "{}",
                        "update_time": 200,
                    }
                ],
                "has_more": True,
                "next_key": "abc",
            },
        )
        assert len(resp.data_items) == 1
        assert resp.has_more is True
        assert resp.next_key == "abc"

    def test_string_flags_are_normalized(self) -> None:
        resp = AggregatedDataResponse(
            code=0,
            result={
                "data_list": [],
                "has_more": "false",
                "next_key": 123,
            },
        )
        assert resp.has_more is False
        assert resp.next_key == "123"


class TestVerifyUserResponse:
    def test_with_user(self) -> None:
        resp = VerifyUserResponse(
            code=0,
            result={"userId": 12345, "nickname": "用户", "icon": "url"},
        )
        info = resp.user_info
        assert info is not None
        assert info.user_id == 12345

    def test_no_user(self) -> None:
        resp = VerifyUserResponse(code=0, result={})
        assert resp.user_info is None


class TestInviteResponse:
    def test_success(self) -> None:
        resp = InviteResponse(code=0, result={"send_ret": 1})
        assert resp.success is True

    def test_failure(self) -> None:
        resp = InviteResponse(code=0, result={"send_ret": 0})
        assert resp.success is False

    def test_string_success_flag(self) -> None:
        resp = InviteResponse(code=0, result={"send_ret": "1"})
        assert resp.success is True


class TestDeleteRelativeResponse:
    def test_success(self) -> None:
        resp = DeleteRelativeResponse(code=0, result={"delete_ret": True})
        assert resp.success is True

    def test_failure(self) -> None:
        resp = DeleteRelativeResponse(code=0, result={"delete_ret": False})
        assert resp.success is False

    def test_string_false_is_false(self) -> None:
        resp = DeleteRelativeResponse(code=0, result={"delete_ret": "false"})
        assert resp.success is False


class TestOperateInviteResponse:
    def test_success(self) -> None:
        resp = OperateInviteResponse(code=0, result={"operate_ret": True})
        assert resp.success is True

    def test_failure(self) -> None:
        resp = OperateInviteResponse(code=0, result={"operate_ret": False})
        assert resp.success is False


class TestSharedDataTypesResponse:
    def test_parse_keys(self) -> None:
        resp = SharedDataTypesResponse(
            code=0,
            result={"keys": ["goal", "heart_rate", "sleep"]},
        )
        assert resp.keys == ["goal", "heart_rate", "sleep"]

    def test_empty(self) -> None:
        resp = SharedDataTypesResponse(code=0, result={})
        assert resp.keys == []

    def test_ignores_non_string_keys(self) -> None:
        resp = SharedDataTypesResponse(
            code=0,
            result={"keys": ["goal", 1, None, "sleep"]},
        )
        assert resp.keys == ["goal", "sleep"]


class TestInviteUniqueIdResponse:
    def test_parse_id(self) -> None:
        resp = InviteUniqueIdResponse(
            code=0,
            result={"invite_link_id": 467184968352742400},
        )
        assert resp.invite_link_id == 467184968352742400


class TestFamilyMemberResponse:
    def test_parse_list(self) -> None:
        resp = FamilyMemberResponse(
            code=0,
            result={"family_user_list": [{"userId": 1}, {"userId": 2}]},
        )
        assert len(resp.family_user_list) == 2

    def test_empty(self) -> None:
        resp = FamilyMemberResponse(code=0, result={})
        assert resp.family_user_list == []


# endregion


# region 消息模型测试
class TestInviteMessage:
    def test_pending_invite(self) -> None:
        msg = InviteMessage(
            msg_id=152824796151809,
            module=1,
            type=1,
            sender=1452722403,
            extra_data='{"invite_id":4777767,"nick_name":"测试","icon":"https://example.com/a.jpg"}',
            data_status=0,
        )
        assert msg.invite_id == 4777767
        assert msg.nick_name == "测试"
        assert msg.icon == "https://example.com/a.jpg"
        assert msg.is_pending is True

    def test_processed_notification(self) -> None:
        msg = InviteMessage(type=5, data_status=1, extra_data='{"nick_name":"用户"}')
        assert msg.invite_id is None
        assert msg.nick_name == "用户"
        assert msg.is_pending is False

    def test_invalid_extra_data(self) -> None:
        msg = InviteMessage(extra_data="not json")
        assert msg.invite_id is None
        assert msg.nick_name == ""
        assert msg.icon == ""

    def test_empty_extra_data(self) -> None:
        msg = InviteMessage(extra_data="")
        assert msg.invite_id is None
        assert msg.nick_name == ""

    def test_json_list_extra_data_is_ignored(self) -> None:
        msg = InviteMessage(extra_data='["bad-shape"]')
        assert msg.invite_id is None
        assert msg.nick_name == ""
        assert msg.icon == ""


class TestMessageListResponse:
    def test_parse_messages(self) -> None:
        resp = MessageListResponse(
            code=0,
            result={
                "messages": [
                    {"msg_id": 1, "type": 1, "sender": 100, "data_status": 0},
                    {"msg_id": 2, "type": 5, "sender": 200, "data_status": 1},
                ],
                "msg_total": 2,
            },
        )
        assert len(resp.messages) == 2
        assert resp.msg_total == 2
        assert resp.messages[0].msg_id == 1

    def test_empty(self) -> None:
        resp = MessageListResponse(code=0, result={})
        assert resp.messages == []
        assert resp.msg_total == 0

    def test_single_message_dict_is_tolerated(self) -> None:
        resp = MessageListResponse(
            code=0,
            result={"messages": {"msg_id": 1, "type": 1, "sender": 100}, "msg_total": "1"},
        )
        assert len(resp.messages) == 1
        assert resp.msg_total == 1


class TestCheckNewMsgResponse:
    def test_has_new(self) -> None:
        resp = CheckNewMsgResponse(code=0, result=[{"module": 1, "is_new": True}])
        assert resp.has_new(1) is True
        assert resp.has_new(2) is False

    def test_no_new(self) -> None:
        resp = CheckNewMsgResponse(code=0, result=[{"module": 1, "is_new": False}])
        assert resp.has_new(1) is False

    def test_empty(self) -> None:
        resp = CheckNewMsgResponse(code=0, result=[])
        assert resp.has_new(1) is False

    def test_single_dict_result_is_tolerated(self) -> None:
        resp = CheckNewMsgResponse(code=0, result={"module": 1, "is_new": "true"})  # type: ignore[arg-type]
        assert resp.has_new(1) is True


# endregion