1import numbers
2import pytest
3from threading import Timer
4
5import pyglet
6pyglet.options['debug_media'] = False
7
8from pyglet.media.codecs import AudioFormat
9from pyglet.media.synthesis import Silence
10
11try:
12    from pyglet.media.drivers.pulse import interface
13except ImportError:
14    interface = None
15
16
17pytestmark = pytest.mark.skipif(interface is None, reason='requires PulseAudio')
18
19
20@pytest.fixture
21def mainloop():
22    return interface.PulseAudioMainLoop()
23
24
25def test_mainloop_run(mainloop):
26    mainloop.start()
27    mainloop.delete()
28
29
30def test_mainloop_lock(mainloop):
31    mainloop.start()
32    mainloop.lock()
33    mainloop.unlock()
34    mainloop.delete()
35
36
37def test_mainloop_signal(mainloop):
38    mainloop.start()
39    with mainloop:
40        mainloop.signal()
41    mainloop.delete()
42
43
44def test_mainloop_wait_signal(mainloop):
45    mainloop.start()
46
47    def signal():
48        with mainloop:
49            mainloop.signal()
50    t = Timer(.1, signal)
51    t.start()
52
53    with mainloop:
54        mainloop.wait()
55    mainloop.delete()
56
57
58@pytest.fixture
59def context(mainloop):
60    mainloop.start()
61    with mainloop:
62        context = mainloop.create_context()
63    yield context
64
65    with context:
66        context.delete()
67    mainloop.delete()
68
69
70def test_context_not_connected(context):
71    assert context.is_ready == False
72    assert context.is_failed == False
73    assert context.is_terminated == False
74    assert context.server == None
75    assert isinstance(context.protocol_version, numbers.Integral)
76    assert context.server_protocol_version == None
77    assert context.is_local == None
78
79    with context:
80        context.delete()
81
82    assert context.is_ready == False
83    assert context.is_failed == False
84    assert context.is_terminated == False  # Never connected, so state is not set yet
85    assert context.server == None
86    assert context.protocol_version == None
87    assert context.server_protocol_version == None
88    assert context.is_local == None
89
90
91def test_context_connect(context):
92    assert context.is_ready == False
93    assert context.is_failed == False
94    assert context.is_terminated == False
95    assert context.server == None
96    assert isinstance(context.protocol_version, numbers.Integral)
97    assert context.server_protocol_version == None
98    assert context.is_local == None
99
100    context.connect()
101
102    assert context.is_ready == True
103    assert context.is_failed == False
104    assert context.is_terminated == False
105    assert isinstance(context.server, str)
106    assert isinstance(context.protocol_version, numbers.Integral)
107    assert isinstance(context.server_protocol_version, numbers.Integral)
108    assert context.is_local == True
109
110    with context:
111        context.delete()
112
113    assert context.is_ready == False
114    assert context.is_failed == False
115    assert context.is_terminated == True
116    assert context.server == None
117    assert context.protocol_version == None
118    assert context.server_protocol_version == None
119    assert context.is_local == None
120
121
122@pytest.fixture
123def stream(context):
124    context.connect()
125    audio_format = AudioFormat(1, 16, 44100)
126    with context:
127        stream = context.create_stream(audio_format)
128    return stream
129
130
131@pytest.fixture
132def audio_source():
133    return Silence(10.0, 44100, 16)
134
135
136@pytest.fixture
137def filled_stream(stream, audio_source):
138    with stream:
139        stream.connect_playback()
140
141    assert stream.is_ready
142    assert stream.writable_size > 0
143
144    nbytes = min(1024, stream.writable_size)
145    audio_data = audio_source.get_audio_data(nbytes)
146    with stream:
147        stream.write(audio_data)
148
149    assert stream.is_ready
150    return stream
151
152
153def test_stream_create(stream):
154    assert stream.is_unconnected == True
155    assert stream.is_creating == False
156    assert stream.is_ready == False
157    assert stream.is_failed == False
158    assert stream.is_terminated == False
159
160    with stream:
161        stream.delete()
162
163    assert stream.is_unconnected == True
164    assert stream.is_creating == False
165    assert stream.is_ready == False
166    assert stream.is_failed == False
167    assert stream.is_terminated == False
168
169
170def test_stream_connect(stream):
171    assert stream.is_unconnected == True
172    assert stream.is_creating == False
173    assert stream.is_ready == False
174    assert stream.is_failed == False
175    assert stream.is_terminated == False
176    assert isinstance(stream.index, numbers.Integral)
177
178    with stream:
179        stream.connect_playback()
180
181    assert stream.is_unconnected == False
182    assert stream.is_creating == False
183    assert stream.is_ready == True
184    assert stream.is_failed == False
185    assert stream.is_terminated == False
186    assert isinstance(stream.index, numbers.Integral)
187
188    with stream:
189        stream.delete()
190
191    assert stream.is_unconnected == False
192    assert stream.is_creating == False
193    assert stream.is_ready == False
194    assert stream.is_failed == False
195    assert stream.is_terminated == True
196
197
198
199def test_stream_write(stream, audio_source):
200    with stream:
201        stream.connect_playback()
202
203    assert stream.is_ready
204    assert stream.writable_size > 0
205
206    nbytes = min(1024, stream.writable_size)
207    audio_data = audio_source.get_audio_data(nbytes)
208    with stream:
209        written = stream.write(audio_data)
210    assert written == nbytes
211
212    assert stream.is_ready
213
214    with stream:
215        stream.delete()
216
217    assert stream.is_terminated
218
219
220
221def test_stream_timing_info(filled_stream):
222    with filled_stream:
223        op = filled_stream.update_timing_info()
224        op.wait()
225    assert op.is_done
226    info = filled_stream.get_timing_info()
227    assert info is not None
228
229    with filled_stream:
230        op.delete()
231
232
233def test_stream_trigger(filled_stream):
234    with filled_stream:
235        op = filled_stream.trigger()
236        op.wait()
237    assert op.is_done
238
239    with filled_stream:
240        op.delete()
241
242
243def test_stream_prebuf(filled_stream):
244    with filled_stream:
245        op = filled_stream.prebuf()
246        op.wait()
247    assert op.is_done
248
249    with filled_stream:
250        op.delete()
251
252
253def test_stream_cork(filled_stream):
254    assert filled_stream.is_corked
255
256    with filled_stream:
257        op = filled_stream.resume()
258        op.wait()
259    assert op.is_done
260    assert not filled_stream.is_corked
261
262    with filled_stream:
263        op.delete()
264        op = filled_stream.pause()
265        op.wait()
266    assert op.is_done
267    assert filled_stream.is_corked
268
269    with filled_stream:
270        op.delete()
271
272
273def test_stream_update_sample_rate(filled_stream):
274    with filled_stream:
275        op = filled_stream.update_sample_rate(44100)
276        op.wait()
277    assert op.is_done
278
279    with filled_stream:
280        op.delete()
281
282
283def test_stream_write_needed(stream, audio_source, event_loop):
284    with stream:
285        stream.connect_playback()
286
287    assert stream.is_ready
288    assert stream.writable_size > 0
289
290    @stream.event
291    def on_write_needed(nbytes, underflow):
292        on_write_needed.nbytes = nbytes
293        on_write_needed.underflow = underflow
294        if underflow is True:
295            event_loop.interrupt_event_loop()
296        return pyglet.event.EVENT_HANDLED
297
298    on_write_needed.nbytes = None
299    on_write_needed.underflow = None
300
301    audio_data = audio_source.get_audio_data(stream.writable_size)
302    with stream:
303        stream.write(audio_data)
304    assert stream.is_ready
305    assert not stream.underflow
306    with stream:
307        stream.resume().wait().delete()
308
309    event_loop.run_event_loop(duration=0.1)
310
311    assert on_write_needed.nbytes > 0
312    assert on_write_needed.underflow == False
313    assert not stream.underflow
314
315    on_write_needed.nbytes = None
316    on_write_needed.underflow = None
317
318    event_loop.run_event_loop(duration=5.0)
319    assert on_write_needed.nbytes > 0
320    assert on_write_needed.underflow == True
321    assert stream.underflow
322
323