1# Copyright 2021-2023 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#      https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15# -----------------------------------------------------------------------------
16# Imports
17# -----------------------------------------------------------------------------
18import asyncio
19import dataclasses
20import pytest
21import pytest_asyncio
22import struct
23import logging
24
25from bumble import device
26from bumble.profiles import mcp
27from tests.test_utils import TwoDevices
28
29
30# -----------------------------------------------------------------------------
31# Logging
32# -----------------------------------------------------------------------------
33logger = logging.getLogger(__name__)
34
35
36# -----------------------------------------------------------------------------
37# Helpers
38# -----------------------------------------------------------------------------
39TIMEOUT = 0.1
40
41
42@dataclasses.dataclass
43class GmcsContext:
44    devices: TwoDevices
45    client: mcp.GenericMediaControlServiceProxy
46    server: mcp.GenericMediaControlService
47
48
49# -----------------------------------------------------------------------------
50@pytest_asyncio.fixture
51async def gmcs_context():
52    devices = TwoDevices()
53    server = mcp.GenericMediaControlService()
54    devices[0].add_service(server)
55
56    await devices.setup_connection()
57    devices.connections[0].encryption = 1
58    devices.connections[1].encryption = 1
59    peer = device.Peer(devices.connections[1])
60    client = await peer.discover_service_and_create_proxy(
61        mcp.GenericMediaControlServiceProxy
62    )
63    await client.subscribe_characteristics()
64
65    return GmcsContext(devices=devices, server=server, client=client)
66
67
68# -----------------------------------------------------------------------------
69@pytest.mark.asyncio
70async def test_update_media_state(gmcs_context):
71    state = asyncio.Queue()
72    gmcs_context.client.on('media_state', state.put_nowait)
73
74    await gmcs_context.devices[0].notify_subscribers(
75        gmcs_context.server.media_state_characteristic,
76        value=bytes([mcp.MediaState.PLAYING]),
77    )
78
79    assert (await asyncio.wait_for(state.get(), TIMEOUT)) == mcp.MediaState.PLAYING
80
81
82# -----------------------------------------------------------------------------
83@pytest.mark.asyncio
84async def test_update_track_title(gmcs_context):
85    state = asyncio.Queue()
86    gmcs_context.client.on('track_title', state.put_nowait)
87
88    await gmcs_context.devices[0].notify_subscribers(
89        gmcs_context.server.track_title_characteristic,
90        value="My Song".encode(),
91    )
92
93    assert (await asyncio.wait_for(state.get(), TIMEOUT)) == "My Song"
94
95
96# -----------------------------------------------------------------------------
97@pytest.mark.asyncio
98async def test_update_track_duration(gmcs_context):
99    state = asyncio.Queue()
100    gmcs_context.client.on('track_duration', state.put_nowait)
101
102    await gmcs_context.devices[0].notify_subscribers(
103        gmcs_context.server.track_duration_characteristic,
104        value=struct.pack("<i", 1000),
105    )
106
107    assert (await asyncio.wait_for(state.get(), TIMEOUT)) == 1000
108
109
110# -----------------------------------------------------------------------------
111@pytest.mark.asyncio
112async def test_update_track_position(gmcs_context):
113    state = asyncio.Queue()
114    gmcs_context.client.on('track_position', state.put_nowait)
115
116    await gmcs_context.devices[0].notify_subscribers(
117        gmcs_context.server.track_position_characteristic,
118        value=struct.pack("<i", 1000),
119    )
120
121    assert (await asyncio.wait_for(state.get(), TIMEOUT)) == 1000
122
123
124# -----------------------------------------------------------------------------
125@pytest.mark.asyncio
126async def test_write_media_control_point(gmcs_context):
127    assert (
128        await asyncio.wait_for(
129            gmcs_context.client.write_control_point(mcp.MediaControlPointOpcode.PAUSE),
130            TIMEOUT,
131        )
132    ) == mcp.MediaControlPointResultCode.SUCCESS
133