~naglis/aio_mpv_ipc

ref: 7e8d90c699b225e33992091ef73cf24e3f8284c8 aio_mpv_ipc/tests/test_client.py -rw-r--r-- 5.9 KiB
7e8d90c6Naglis Jonaitis Update pytest-asyncio 6 months ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
import asyncio
import contextlib
import functools
import pathlib
import shutil
import tempfile

import pytest

import aio_mpv_ipc

TEST_FILE_SILENCE = str(pathlib.Path(__file__).parent / "test_data" / "silence.ogg")
TMP_FILE_PREFIX = "aio_mpv_ipc.test."

MPV_PATH = shutil.which("mpv")

MPV_ARGS = [
    "--idle",  # Do not exit immediately.
    "--no-audio-display",
    "--no-input-terminal",  # Do not intercept keyboard input.
    "--no-input-cursor",
    "--no-video",
    "--no-resume-playback",
    "--no-osc",
    "--ao=null",
    "--vo=null",
]


tmp_socket = functools.partial(
    tempfile.NamedTemporaryFile,
    prefix="aio_mpv_ipc.test.",
    suffix=".socket",
    delete=True,
)


@pytest.fixture
def queue():
    yield asyncio.Queue()


@pytest.fixture
@pytest.mark.asyncio
async def mpv_client():
    async with contextlib.AsyncExitStack() as exit_stack:
        tmp_file = exit_stack.enter_context(tmp_socket())
        socket_path = tmp_file.name

        mpv_ctx_mgr = aio_mpv_ipc.start_mpv(
            socket_path, *MPV_ARGS, terminate=True, mpv_path=MPV_PATH
        )
        await exit_stack.enter_async_context(mpv_ctx_mgr)

        client = await exit_stack.enter_async_context(
            aio_mpv_ipc.MpvClient(socket_path)
        )

        yield client


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_loadfile(mpv_client):
    await mpv_client.ipc("loadfile", TEST_FILE_SILENCE, "append")
    count = await mpv_client.ipc("get_property", "playlist-count")
    assert count == 1


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_unknown_command_ipc_exception_is_raised(mpv_client):
    with pytest.raises(aio_mpv_ipc.IPCException, match=r"^invalid parameter$"):
        await mpv_client.ipc("foobar")


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_unknown_command_nowait_ipc_exception_is_not_raised(mpv_client):
    try:
        await mpv_client.ipc("foobar", nowait=True)
    except aio_mpv_ipc.IPCException:
        pytest.fail("IPCException was raised")


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_ipc_exception_response_is_set(mpv_client):
    try:
        await mpv_client.ipc("foobar")
    except aio_mpv_ipc.IPCException as exc:
        assert exc.response == {"request_id": 1, "error": "invalid parameter"}
    else:
        pytest.fail("IPCException not raised")


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_event_subscription(mpv_client, queue):
    async def event_handler():
        event_data = await queue.get()
        queue.task_done()
        return event_data

    mpv_client.subscribe("file-loaded", queue)
    await mpv_client.ipc("loadfile", TEST_FILE_SILENCE)
    result = await asyncio.wait_for(event_handler(), timeout=10)
    assert result["event"] == "file-loaded"


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_event_unsubscription(mpv_client, queue):
    async def event_handler():
        event_data = await queue.get()
        queue.task_done()
        return event_data

    mpv_client.subscribe("file-loaded", queue)
    await mpv_client.ipc("loadfile", TEST_FILE_SILENCE)
    result = await asyncio.wait_for(event_handler(), timeout=10)
    assert result["event"] == "file-loaded"
    await mpv_client.ipc("playlist-clear")
    mpv_client.unsubscribe("file-loaded", queue)
    await mpv_client.ipc("loadfile", TEST_FILE_SILENCE)
    await asyncio.sleep(1)
    assert queue.empty()


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_mpv_started_with_unknown_flag_exception_is_raised():
    with tmp_socket() as socket_file:
        mpv = aio_mpv_ipc.start_mpv(
            socket_file.name,
            *[*MPV_ARGS, "--foo-bar"],
            terminate=True,
            mpv_path=MPV_PATH,
        )
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        with pytest.raises(
            aio_mpv_ipc.MpvStartException, match=r"mpv exited with status code: 1"
        ):
            async with mpv:
                await client.__aenter__()


@pytest.mark.asyncio
async def test_mpv_not_started_client_raises_error():
    with tmp_socket() as socket_file:
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        with pytest.raises(
            aio_mpv_ipc.IPCConnectionException, match=r"Failed to connect to mpv"
        ):
            await client.__aenter__()


@pytest.mark.asyncio
async def test_client_not_used_as_context_manager_ipc_raises_error():
    with tmp_socket() as socket_file:
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        with pytest.raises(
            aio_mpv_ipc.AioMpvIPCException,
            match=r"mpv JSON IPC socket writer was not initialized",
        ):
            await client.ipc("client_name")


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_exit_mpv_from_client():
    with tmp_socket() as socket_file:
        mpv = aio_mpv_ipc.start_mpv(socket_file.name, *MPV_ARGS, mpv_path=MPV_PATH)
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        async with mpv as mpv, client:
            await client.ipc("quit")
            await mpv.wait()

        assert mpv.returncode == 0


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_ipc_after_mpv_exit():
    with tmp_socket() as socket_file:
        mpv = aio_mpv_ipc.start_mpv(socket_file.name, *MPV_ARGS, mpv_path=MPV_PATH)
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        async with mpv as mpv, client:
            await client.ipc("quit")
            await mpv.wait()
            with pytest.raises(
                aio_mpv_ipc.exceptions.IPCConnectionException,
                match=r"IPC connection to mpv failed",
            ):
                await client.ipc("client_name")


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_client_context_manager_cleanup():
    with tmp_socket() as socket_file:
        mpv = aio_mpv_ipc.start_mpv(
            socket_file.name, *MPV_ARGS, terminate=True, mpv_path=MPV_PATH
        )
        client = aio_mpv_ipc.MpvClient(socket_file.name)
        async with mpv as mpv, client:
            pass

        assert client._poll_task is None