1# Copyright 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
15# -----------------------------------------------------------------------------
16# Imports
17# -----------------------------------------------------------------------------
18import pytest
19import pytest_asyncio
20
21from bumble import device
22
23from bumble.att import ATT_Error
24
25from bumble.profiles.aics import (
26    Mute,
27    AICSService,
28    AudioInputState,
29    AICSServiceProxy,
30    GainMode,
31    AudioInputStatus,
32    AudioInputControlPointOpCode,
33    ErrorCode,
34)
35from bumble.profiles.vcp import VolumeControlService, VolumeControlServiceProxy
36
37from .test_utils import TwoDevices
38
39
40# -----------------------------------------------------------------------------
41# Tests
42# -----------------------------------------------------------------------------
43aics_service = AICSService()
44vcp_service = VolumeControlService(
45    volume_setting=32, muted=1, volume_flags=1, included_services=[aics_service]
46)
47
48
49@pytest_asyncio.fixture
50async def aics_client():
51    devices = TwoDevices()
52    devices[0].add_service(vcp_service)
53
54    await devices.setup_connection()
55
56    assert devices.connections[0]
57    assert devices.connections[1]
58
59    devices.connections[0].encryption = 1
60    devices.connections[1].encryption = 1
61
62    peer = device.Peer(devices.connections[1])
63
64    vcp_client = await peer.discover_service_and_create_proxy(VolumeControlServiceProxy)
65
66    assert vcp_client
67    included_services = await peer.discover_included_services(vcp_client.service_proxy)
68    assert included_services
69    aics_service_discovered = included_services[0]
70    await peer.discover_characteristics(service=aics_service_discovered)
71    aics_client = AICSServiceProxy(aics_service_discovered)
72
73    yield aics_client
74
75
76# -----------------------------------------------------------------------------
77@pytest.mark.asyncio
78async def test_init_service(aics_client: AICSServiceProxy):
79    assert await aics_client.audio_input_state.read_value() == AudioInputState(
80        gain_settings=0,
81        mute=Mute.NOT_MUTED,
82        gain_mode=GainMode.MANUAL,
83        change_counter=0,
84    )
85    assert await aics_client.gain_settings_properties.read_value() == (1, 0, 255)
86    assert await aics_client.audio_input_status.read_value() == (
87        AudioInputStatus.ACTIVE
88    )
89
90
91@pytest.mark.asyncio
92async def test_wrong_opcode_raise_error(aics_client: AICSServiceProxy):
93    with pytest.raises(ATT_Error) as e:
94        await aics_client.audio_input_control_point.write_value(
95            bytes(
96                [
97                    0xFF,
98                ]
99            ),
100            with_response=True,
101        )
102
103    assert e.value.error_code == ErrorCode.OPCODE_NOT_SUPPORTED
104
105
106@pytest.mark.asyncio
107async def test_set_gain_setting_when_gain_mode_automatic_only(
108    aics_client: AICSServiceProxy,
109):
110    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC_ONLY
111
112    change_counter = 0
113    gain_settings = 120
114    await aics_client.audio_input_control_point.write_value(
115        bytes(
116            [
117                AudioInputControlPointOpCode.SET_GAIN_SETTING,
118                change_counter,
119                gain_settings,
120            ]
121        )
122    )
123
124    # Unchanged
125    assert await aics_client.audio_input_state.read_value() == AudioInputState(
126        gain_settings=0,
127        mute=Mute.NOT_MUTED,
128        gain_mode=GainMode.AUTOMATIC_ONLY,
129        change_counter=0,
130    )
131
132
133@pytest.mark.asyncio
134async def test_set_gain_setting_when_gain_mode_automatic(aics_client: AICSServiceProxy):
135    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC
136    change_counter = 0
137    gain_settings = 120
138    await aics_client.audio_input_control_point.write_value(
139        bytes(
140            [
141                AudioInputControlPointOpCode.SET_GAIN_SETTING,
142                change_counter,
143                gain_settings,
144            ]
145        )
146    )
147
148    # Unchanged
149    assert await aics_client.audio_input_state.read_value() == AudioInputState(
150        gain_settings=0,
151        mute=Mute.NOT_MUTED,
152        gain_mode=GainMode.AUTOMATIC,
153        change_counter=0,
154    )
155
156
157@pytest.mark.asyncio
158async def test_set_gain_setting_when_gain_mode_MANUAL(aics_client: AICSServiceProxy):
159    aics_service.audio_input_state.gain_mode = GainMode.MANUAL
160    change_counter = 0
161    gain_settings = 120
162    await aics_client.audio_input_control_point.write_value(
163        bytes(
164            [
165                AudioInputControlPointOpCode.SET_GAIN_SETTING,
166                change_counter,
167                gain_settings,
168            ]
169        )
170    )
171
172    assert await aics_client.audio_input_state.read_value() == AudioInputState(
173        gain_settings=gain_settings,
174        mute=Mute.NOT_MUTED,
175        gain_mode=GainMode.MANUAL,
176        change_counter=change_counter,
177    )
178
179
180@pytest.mark.asyncio
181async def test_set_gain_setting_when_gain_mode_MANUAL_ONLY(
182    aics_client: AICSServiceProxy,
183):
184    aics_service.audio_input_state.gain_mode = GainMode.MANUAL_ONLY
185    change_counter = 0
186    gain_settings = 120
187    await aics_client.audio_input_control_point.write_value(
188        bytes(
189            [
190                AudioInputControlPointOpCode.SET_GAIN_SETTING,
191                change_counter,
192                gain_settings,
193            ]
194        )
195    )
196
197    assert await aics_client.audio_input_state.read_value() == AudioInputState(
198        gain_settings=gain_settings,
199        mute=Mute.NOT_MUTED,
200        gain_mode=GainMode.MANUAL_ONLY,
201        change_counter=change_counter,
202    )
203
204
205@pytest.mark.asyncio
206async def test_unmute_when_muted(aics_client: AICSServiceProxy):
207    aics_service.audio_input_state.mute = Mute.MUTED
208    change_counter = 0
209    await aics_client.audio_input_control_point.write_value(
210        bytes(
211            [
212                AudioInputControlPointOpCode.UNMUTE,
213                change_counter,
214            ]
215        )
216    )
217
218    change_counter += 1
219
220    state: AudioInputState = await aics_client.audio_input_state.read_value()
221    assert state.mute == Mute.NOT_MUTED
222    assert state.change_counter == change_counter
223
224
225@pytest.mark.asyncio
226async def test_unmute_when_mute_disabled(aics_client: AICSServiceProxy):
227    aics_service.audio_input_state.mute = Mute.DISABLED
228    aics_service.audio_input_state.change_counter = 0
229    change_counter = 0
230
231    with pytest.raises(ATT_Error) as e:
232        await aics_client.audio_input_control_point.write_value(
233            bytes(
234                [
235                    AudioInputControlPointOpCode.UNMUTE,
236                    change_counter,
237                ]
238            ),
239            with_response=True,
240        )
241
242    assert e.value.error_code == ErrorCode.MUTE_DISABLED
243
244    state: AudioInputState = await aics_client.audio_input_state.read_value()
245    assert state.mute == Mute.DISABLED
246    assert state.change_counter == change_counter
247
248
249@pytest.mark.asyncio
250async def test_mute_when_not_muted(aics_client: AICSServiceProxy):
251    aics_service.audio_input_state.mute = Mute.NOT_MUTED
252    aics_service.audio_input_state.change_counter = 0
253    change_counter = 0
254
255    await aics_client.audio_input_control_point.write_value(
256        bytes(
257            [
258                AudioInputControlPointOpCode.MUTE,
259                change_counter,
260            ]
261        )
262    )
263
264    change_counter += 1
265    state: AudioInputState = await aics_client.audio_input_state.read_value()
266    assert state.mute == Mute.MUTED
267    assert state.change_counter == change_counter
268
269
270@pytest.mark.asyncio
271async def test_mute_when_mute_disabled(aics_client: AICSServiceProxy):
272    aics_service.audio_input_state.mute = Mute.DISABLED
273    aics_service.audio_input_state.change_counter = 0
274    change_counter = 0
275
276    with pytest.raises(ATT_Error) as e:
277        await aics_client.audio_input_control_point.write_value(
278            bytes(
279                [
280                    AudioInputControlPointOpCode.MUTE,
281                    change_counter,
282                ]
283            ),
284            with_response=True,
285        )
286
287    assert e.value.error_code == ErrorCode.MUTE_DISABLED
288
289    state: AudioInputState = await aics_client.audio_input_state.read_value()
290    assert state.mute == Mute.DISABLED
291    assert state.change_counter == change_counter
292
293
294@pytest.mark.asyncio
295async def test_set_manual_gain_mode_when_automatic(aics_client: AICSServiceProxy):
296    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC
297    aics_service.audio_input_state.change_counter = 0
298    change_counter = 0
299
300    await aics_client.audio_input_control_point.write_value(
301        bytes(
302            [
303                AudioInputControlPointOpCode.SET_MANUAL_GAIN_MODE,
304                change_counter,
305            ]
306        )
307    )
308
309    change_counter += 1
310    state: AudioInputState = await aics_client.audio_input_state.read_value()
311    assert state.gain_mode == GainMode.MANUAL
312    assert state.change_counter == change_counter
313
314
315@pytest.mark.asyncio
316async def test_set_manual_gain_mode_when_already_manual(aics_client: AICSServiceProxy):
317    aics_service.audio_input_state.gain_mode = GainMode.MANUAL
318    aics_service.audio_input_state.change_counter = 0
319    change_counter = 0
320
321    await aics_client.audio_input_control_point.write_value(
322        bytes(
323            [
324                AudioInputControlPointOpCode.SET_MANUAL_GAIN_MODE,
325                change_counter,
326            ]
327        )
328    )
329
330    # No change expected
331    state: AudioInputState = await aics_client.audio_input_state.read_value()
332    assert state.gain_mode == GainMode.MANUAL
333    assert state.change_counter == change_counter
334
335
336@pytest.mark.asyncio
337async def test_set_manual_gain_mode_when_manual_only(aics_client: AICSServiceProxy):
338    aics_service.audio_input_state.gain_mode = GainMode.MANUAL_ONLY
339    aics_service.audio_input_state.change_counter = 0
340    change_counter = 0
341
342    with pytest.raises(ATT_Error) as e:
343        await aics_client.audio_input_control_point.write_value(
344            bytes(
345                [
346                    AudioInputControlPointOpCode.SET_MANUAL_GAIN_MODE,
347                    change_counter,
348                ]
349            ),
350            with_response=True,
351        )
352
353    assert e.value.error_code == ErrorCode.GAIN_MODE_CHANGE_NOT_ALLOWED
354
355    state: AudioInputState = await aics_client.audio_input_state.read_value()
356    assert state.gain_mode == GainMode.MANUAL_ONLY
357    assert state.change_counter == change_counter
358
359
360@pytest.mark.asyncio
361async def test_set_manual_gain_mode_when_automatic_only(aics_client: AICSServiceProxy):
362    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC_ONLY
363    aics_service.audio_input_state.change_counter = 0
364    change_counter = 0
365
366    with pytest.raises(ATT_Error) as e:
367        await aics_client.audio_input_control_point.write_value(
368            bytes(
369                [
370                    AudioInputControlPointOpCode.SET_MANUAL_GAIN_MODE,
371                    change_counter,
372                ]
373            ),
374            with_response=True,
375        )
376
377    assert e.value.error_code == ErrorCode.GAIN_MODE_CHANGE_NOT_ALLOWED
378
379    # No change expected
380    state: AudioInputState = await aics_client.audio_input_state.read_value()
381    assert state.gain_mode == GainMode.AUTOMATIC_ONLY
382    assert state.change_counter == change_counter
383
384
385@pytest.mark.asyncio
386async def test_set_automatic_gain_mode_when_manual(aics_client: AICSServiceProxy):
387    aics_service.audio_input_state.gain_mode = GainMode.MANUAL
388    aics_service.audio_input_state.change_counter = 0
389    change_counter = 0
390
391    await aics_client.audio_input_control_point.write_value(
392        bytes(
393            [
394                AudioInputControlPointOpCode.SET_AUTOMATIC_GAIN_MODE,
395                change_counter,
396            ]
397        )
398    )
399
400    change_counter += 1
401    state: AudioInputState = await aics_client.audio_input_state.read_value()
402    assert state.gain_mode == GainMode.AUTOMATIC
403    assert state.change_counter == change_counter
404
405
406@pytest.mark.asyncio
407async def test_set_automatic_gain_mode_when_already_automatic(
408    aics_client: AICSServiceProxy,
409):
410    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC
411    aics_service.audio_input_state.change_counter = 0
412    change_counter = 0
413
414    await aics_client.audio_input_control_point.write_value(
415        bytes(
416            [
417                AudioInputControlPointOpCode.SET_AUTOMATIC_GAIN_MODE,
418                change_counter,
419            ]
420        )
421    )
422
423    # No change expected
424    state: AudioInputState = await aics_client.audio_input_state.read_value()
425    assert state.gain_mode == GainMode.AUTOMATIC
426    assert state.change_counter == change_counter
427
428
429@pytest.mark.asyncio
430async def test_set_automatic_gain_mode_when_manual_only(aics_client: AICSServiceProxy):
431    aics_service.audio_input_state.gain_mode = GainMode.MANUAL_ONLY
432    aics_service.audio_input_state.change_counter = 0
433    change_counter = 0
434
435    with pytest.raises(ATT_Error) as e:
436        await aics_client.audio_input_control_point.write_value(
437            bytes(
438                [
439                    AudioInputControlPointOpCode.SET_AUTOMATIC_GAIN_MODE,
440                    change_counter,
441                ]
442            ),
443            with_response=True,
444        )
445
446    assert e.value.error_code == ErrorCode.GAIN_MODE_CHANGE_NOT_ALLOWED
447
448    # No change expected
449    state: AudioInputState = await aics_client.audio_input_state.read_value()
450    assert state.gain_mode == GainMode.MANUAL_ONLY
451    assert state.change_counter == change_counter
452
453
454@pytest.mark.asyncio
455async def test_set_automatic_gain_mode_when_automatic_only(
456    aics_client: AICSServiceProxy,
457):
458    aics_service.audio_input_state.gain_mode = GainMode.AUTOMATIC_ONLY
459    aics_service.audio_input_state.change_counter = 0
460    change_counter = 0
461
462    with pytest.raises(ATT_Error) as e:
463        await aics_client.audio_input_control_point.write_value(
464            bytes(
465                [
466                    AudioInputControlPointOpCode.SET_AUTOMATIC_GAIN_MODE,
467                    change_counter,
468                ]
469            ),
470            with_response=True,
471        )
472
473    assert e.value.error_code == ErrorCode.GAIN_MODE_CHANGE_NOT_ALLOWED
474
475    # No change expected
476    state: AudioInputState = await aics_client.audio_input_state.read_value()
477    assert state.gain_mode == GainMode.AUTOMATIC_ONLY
478    assert state.change_counter == change_counter
479
480
481@pytest.mark.asyncio
482async def test_audio_input_description_initial_value(aics_client: AICSServiceProxy):
483    description = await aics_client.audio_input_description.read_value()
484    assert description.decode('utf-8') == "Bluetooth"
485
486
487@pytest.mark.asyncio
488async def test_audio_input_description_write_and_read(aics_client: AICSServiceProxy):
489    new_description = "Line Input".encode('utf-8')
490
491    await aics_client.audio_input_description.write_value(new_description)
492
493    description = await aics_client.audio_input_description.read_value()
494    assert description == new_description
495