/
/
/
1"""Tests for the metadata the AriaCast receiver exchanges with its senders."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import Any, cast
7from unittest.mock import AsyncMock
8
9import pytest
10from music_assistant_models.streamdetails import StreamMetadata
11
12from music_assistant.providers.ariacast_receiver import AriaCastReceiver
13
14
15def _meta_dict(elapsed_time: int | None) -> dict[str, Any]:
16 """Serialise the outgoing wire payload for a stream at the given position."""
17 receiver = SimpleNamespace(
18 _stream_meta=StreamMetadata(title="Test Track", elapsed_time=elapsed_time),
19 _is_playing=True,
20 )
21 return AriaCastReceiver._meta_dict(cast("AriaCastReceiver", receiver))
22
23
24async def _apply_meta(data: dict[str, Any]) -> StreamMetadata:
25 """Merge an incoming sender update into a stream that already has a position."""
26 receiver = SimpleNamespace(
27 _stream_meta=StreamMetadata(title="Test Track", elapsed_time=120, duration=240),
28 _last_artwork_url=None,
29 _broadcast_meta=AsyncMock(),
30 )
31 await AriaCastReceiver._apply_meta(cast("AriaCastReceiver", receiver), data)
32 return cast("StreamMetadata", receiver._stream_meta)
33
34
35def test_zero_position_is_reported() -> None:
36 """The start of a track is sent as position 0, not as 'unknown'."""
37 assert _meta_dict(0)["position_ms"] == 0
38
39
40def test_position_is_reported_in_milliseconds() -> None:
41 """A known position is converted from seconds to milliseconds."""
42 assert _meta_dict(42)["position_ms"] == 42000
43
44
45def test_missing_position_stays_none() -> None:
46 """A stream without a position reports none."""
47 assert _meta_dict(None)["position_ms"] is None
48
49
50async def test_zero_position_from_sender_is_applied() -> None:
51 """A sender rewinding to the start of a track replaces the previous position."""
52 assert (await _apply_meta({"positionMs": 0})).elapsed_time == 0
53
54
55@pytest.mark.parametrize("key", ["positionMs", "position_ms"])
56async def test_position_from_sender_is_applied(key: str) -> None:
57 """Senders may use either casing for the position."""
58 assert (await _apply_meta({key: 42000})).elapsed_time == 42
59
60
61async def test_update_without_position_keeps_the_previous_one() -> None:
62 """An update that carries no position leaves the known position alone."""
63 assert (await _apply_meta({"title": "Other Track"})).elapsed_time == 120
64