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
|
from .mpris import setup_mpris, setup_playerctld
from .playerctl import PlayerctlCli
import pytest
import asyncio
def selector(bus_address):
playerctl = PlayerctlCli(bus_address)
async def select(*players):
assert players
cmd = '-p ' + str.join(
',', players) + ' status --format "{{playerInstance}}"'
result = await playerctl.run(cmd)
assert result.returncode == 0, result.stderr
return result.stdout
async def select_many(*players):
assert players
cmd = '--all-players -p ' + str.join(
',', players) + ' status --format "{{playerInstance}}"'
result = await playerctl.run(cmd)
assert result.returncode == 0, result.stderr
return tuple(result.stdout.split('\n'))
return select, select_many
@pytest.mark.asyncio
async def test_selection(bus_address):
s1 = 'selection1'
s1i = 'selection1.i_123'
s2 = 'selection2'
s3 = 'selection3'
m4 = 'selection4'
m5 = 'selection5'
m6 = 'selection6'
s6i = 'selection6.i_2'
any_player = '%any'
mpris_players = await setup_mpris(s1,
s1i,
s2,
s3,
s6i,
bus_address=bus_address)
# TODO: test ignored players
selections = {
(s1, ): (s1, s1i),
(s3, s1): (s3, s1, s1i),
(s2, s1, s3): (s2, s1, s1i, s3),
(s1, s2): (s1, s1i, s2),
(m4, m5, s2, s3): (s2, s3),
(m5, s1, m4, s3): (s1, s1i, s3),
(s1, s1i): (s1, s1i),
(s1i, s1): (s1i, s1),
(m6, s1): (s6i, s1, s1i),
(m4, m6, s3): (s6i, s3),
(any_player, ): (s1, s1i, s2, s3, s6i),
(s1, any_player): (s1, s1i, s2, s3, s6i), # s1 first
(any_player, s1): (s2, s3, s6i, s1, s1i), # s1 last
(m6, any_player, s2): (s6i, s1, s1i, s3, s2), # s6 first, s2 last
(m6, s1, any_player, s2): (s6i, s1, s1i, s3, s2),
}
select, select_many = selector(bus_address)
for selection, expected in selections.items():
result = await select(*selection)
assert result == expected[0], (selection, expected, result)
result = await select_many(*selection)
assert result == expected
await asyncio.gather(*[mpris.disconnect() for mpris in mpris_players])
@pytest.mark.asyncio
async def test_daemon_selection(bus_address):
playerctld = await setup_playerctld(bus_address=bus_address)
playerctl = PlayerctlCli(bus_address)
def iface_name(player_name):
return f'org.mpris.MediaPlayer2.{player_name}'
def set_players(players):
playerctld.player_names = [iface_name(p) for p in players]
s1 = 'selection1'
s1i = 'selection1.i_123'
s2 = 'selection2'
s2i = 'selection2.i_123'
s3 = 'selection3'
m4 = 'selection4'
m5 = 'selection5'
m6 = 'selection6'
s6i = 'selection6.i_2'
any_player = '%any'
# selection, players, expected result
all_players = [s1, s1i, s2, s3, s6i]
tests = [
(None, all_players, all_players),
(all_players, all_players, all_players),
([s2], [s1, s2], [s2]),
([s1], [s2, s1i, s1], [s1i, s1]),
([s1], [s2, s1, s1i], [s1, s1i]),
([s1i, s1], [s1, s1i], [s1i, s1]),
([any_player], all_players, all_players),
([any_player, s1], [s1, s1i, s2i, s2], [s2i, s2, s1, s1i]),
([any_player, s1], [s1, s1i, s2, s2i], [s2, s2i, s1, s1i]),
([any_player, s1], [s1i, s1, s2i, s2], [s2i, s2, s1i, s1]),
([any_player, s1], [s1i, s1, s2, s2i], [s2, s2i, s1i, s1]),
([s2, any_player], [s1, s1i, s2i, s2], [s2i, s2, s1, s1i]),
([s2, any_player], [s1, s1i, s2, s2i], [s2, s2i, s1, s1i]),
([s2, any_player], [s1i, s1, s2i, s2], [s2i, s2, s1i, s1]),
([s2, any_player], [s1i, s1, s2, s2i], [s2, s2i, s1i, s1]),
([s2i, any_player], [s1i, s1, s2, s2i], [s2i, s1i, s1, s2]),
]
async def daemon_selection_test(test):
selection, players, expected = test
set_players(players)
result = await playerctl.list(players=selection)
assert result == expected, test
for test in tests:
# unfortunately it won't work in parallel because there's only one
# playerctld
await daemon_selection_test(test)
await playerctld.disconnect()
|
