~naglis/aio_mpv_ipc

ref: 0.0.1a7 aio_mpv_ipc/tests/test_client.py -rw-r--r-- 4.3 KiB
61756659Naglis Jonaitis Bump version 1 year, 2 days 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
import asyncio
import contextlib
import pathlib
import shutil
import tempfile

import pytest

import aio_mpv_ipc

TEST_FILE_SILENCE = pathlib.Path(__file__).parent / "test_data" / "silence.ogg"

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",
]


@contextlib.asynccontextmanager
async def temp_file(*args, **kwargs):
    """Awaitable version of :class:`tempfile.NamedTemporaryFile`."""
    with tempfile.NamedTemporaryFile(*args, **kwargs) as tmp:
        yield tmp


@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 = await exit_stack.enter_async_context(
            temp_file(prefix=f"veil.test.", suffix=".socket", delete=True)
        )
        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=1.0)
    assert result["event"] == "file-loaded"


@pytest.mark.mpv
@pytest.mark.asyncio
async def test_mpv_started_with_unknown_flag_exception_is_raised():
    async with temp_file(
        prefix=f"veil.test.", suffix=".socket", delete=True
    ) 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, max_connect_attempts=3)
        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():
    async with temp_file(
        prefix=f"veil.test.", suffix=".socket", delete=True
    ) as socket_file:
        client = aio_mpv_ipc.MpvClient(socket_file.name, max_connect_attempts=3)
        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():
    async with temp_file(
        prefix=f"veil.test.", suffix=".socket", delete=True
    ) as socket_file:
        client = aio_mpv_ipc.MpvClient(socket_file.name, max_connect_attempts=3)
        with pytest.raises(
            aio_mpv_ipc.AioMpvIPCException,
            match=r"mpv IPC socket writer was not initialized",
        ):
            await client.ipc("client_name")