1# Copyright 2021-2024 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
15import asyncio
16import pytest
17import struct
18from unittest import mock
19
20from bumble import device as bumble_device
21from bumble.profiles import asha
22
23from .test_utils import TwoDevices
24
25# -----------------------------------------------------------------------------
26HI_SYNC_ID = b'\x00\x01\x02\x03\x04\x05\x06\x07'
27TIMEOUT = 0.1
28
29
30# -----------------------------------------------------------------------------
31@pytest.mark.asyncio
32async def test_get_only_properties():
33    devices = TwoDevices()
34    await devices.setup_connection()
35
36    asha_service = asha.AshaService(
37        hisyncid=HI_SYNC_ID,
38        device=devices[0],
39        protocol_version=0x01,
40        capability=0x02,
41        feature_map=0x03,
42        render_delay_milliseconds=0x04,
43        supported_codecs=0x05,
44    )
45    devices[0].add_service(asha_service)
46
47    async with bumble_device.Peer(devices.connections[1]) as peer:
48        asha_client = peer.create_service_proxy(asha.AshaServiceProxy)
49        assert asha_client
50
51        read_only_properties = (
52            await asha_client.read_only_properties_characteristic.read_value()
53        )
54        (
55            protocol_version,
56            capabilities,
57            hi_sync_id,
58            feature_map,
59            render_delay_milliseconds,
60            _,
61            supported_codecs,
62        ) = struct.unpack("<BB8sBHHH", read_only_properties)
63        assert protocol_version == 0x01
64        assert capabilities == 0x02
65        assert hi_sync_id == HI_SYNC_ID
66        assert feature_map == 0x03
67        assert render_delay_milliseconds == 0x04
68        assert supported_codecs == 0x05
69
70
71# -----------------------------------------------------------------------------
72@pytest.mark.asyncio
73async def test_get_psm():
74    devices = TwoDevices()
75    await devices.setup_connection()
76
77    asha_service = asha.AshaService(
78        hisyncid=HI_SYNC_ID,
79        device=devices[0],
80        capability=0,
81    )
82    devices[0].add_service(asha_service)
83
84    async with bumble_device.Peer(devices.connections[1]) as peer:
85        asha_client = peer.create_service_proxy(asha.AshaServiceProxy)
86        assert asha_client
87
88        psm = (await asha_client.psm_characteristic.read_value())[0]
89        assert psm == asha_service.psm
90
91
92# -----------------------------------------------------------------------------
93@pytest.mark.asyncio
94async def test_write_audio_control_point_start():
95    devices = TwoDevices()
96    await devices.setup_connection()
97
98    asha_service = asha.AshaService(
99        hisyncid=HI_SYNC_ID,
100        device=devices[0],
101        capability=0,
102    )
103    devices[0].add_service(asha_service)
104
105    async with bumble_device.Peer(devices.connections[1]) as peer:
106        asha_client = peer.create_service_proxy(asha.AshaServiceProxy)
107        assert asha_client
108        status_notifications = asyncio.Queue()
109        await asha_client.audio_status_point_characteristic.subscribe(
110            status_notifications.put_nowait
111        )
112
113        start_cb = mock.MagicMock()
114        asha_service.on('started', start_cb)
115        await asha_client.audio_control_point_characteristic.write_value(
116            bytes(
117                [asha.OpCode.START, asha.Codec.G_722_16KHZ, asha.AudioType.MEDIA, 0, 1]
118            )
119        )
120        status = (await asyncio.wait_for(status_notifications.get(), TIMEOUT))[0]
121        assert status == asha.AudioStatus.OK
122
123        start_cb.assert_called_once()
124        assert asha_service.active_codec == asha.Codec.G_722_16KHZ
125        assert asha_service.volume == 0
126        assert asha_service.other_state == 1
127        assert asha_service.audio_type == asha.AudioType.MEDIA
128
129
130# -----------------------------------------------------------------------------
131@pytest.mark.asyncio
132async def test_write_audio_control_point_stop():
133    devices = TwoDevices()
134    await devices.setup_connection()
135
136    asha_service = asha.AshaService(
137        hisyncid=HI_SYNC_ID,
138        device=devices[0],
139        capability=0,
140    )
141    devices[0].add_service(asha_service)
142
143    async with bumble_device.Peer(devices.connections[1]) as peer:
144        asha_client = peer.create_service_proxy(asha.AshaServiceProxy)
145        assert asha_client
146        status_notifications = asyncio.Queue()
147        await asha_client.audio_status_point_characteristic.subscribe(
148            status_notifications.put_nowait
149        )
150
151        stop_cb = mock.MagicMock()
152        asha_service.on('stopped', stop_cb)
153        await asha_client.audio_control_point_characteristic.write_value(
154            bytes([asha.OpCode.STOP])
155        )
156        status = (await asyncio.wait_for(status_notifications.get(), TIMEOUT))[0]
157        assert status == asha.AudioStatus.OK
158
159        stop_cb.assert_called_once()
160        assert asha_service.active_codec is None
161        assert asha_service.volume is None
162        assert asha_service.other_state is None
163        assert asha_service.audio_type is None
164