~naglis/aio_mpv_ipc

ref: 0.0.1a10 aio_mpv_ipc/tests/test_client.py -rw-r--r-- 5.5 KiB
1e6cba77Naglis Jonaitis Bump version 11 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
import asyncio
import contextlib
import functools
import pathlib
import shutil
import tempfile

import pytest

import aio_mpv_ipc

TEST_FILE_SILENCE = 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(event_loop, 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(event_loop, 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")