colin/playerctl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor tests

Browse Source
This commit is contained in:
Tony Crisci
2020-01-25 02:34:44 -05:00
parent 9491e95624
commit 145e576110
10 changed files with 175 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
.flake8 Normal file
View File

@@ -0,0 +1,14 @@
[flake8]
ignore=
E501
E126
E402
F722
# F821 is still relevant, but causes too many false positives in tests and
# examples
per-file-ignores=
test/*:F821
test/util.py:F401
examples/*:F821
*/__init__.py:F401

3
Makefile
View File

@@ -17,4 +17,7 @@ format:
yapf -rip test examples
clang-format -i ${FORMAT_C_SOURCE}
lint:
flake8 test
all: format docker-test

1
test/conftest.py
View File

@@ -1,4 +1,3 @@
from dbus_next.aio import MessageBus
import pytest
import asyncio

51
test/mpris.py
View File

@@ -1,29 +1,28 @@
from dbus_next.service import ServiceInterface, dbus_property, method, signal
from dbus_next.service import ServiceInterface, dbus_property, method, signal, Variant
from dbus_next import PropertyAccess, RequestNameReply
from dbus_next.aio import MessageBus
import asyncio
async def setup_buses(*names, bus_address=None):
async def setup_mpris(*names, bus_address=None):
async def setup(name):
bus = await MessageBus(bus_address=bus_address).connect()
reply = await bus.request_name(f'org.mpris.MediaPlayer2.{name}')
assert reply == RequestNameReply.PRIMARY_OWNER
bus.export('/org/mpris/MediaPlayer2', MprisPlayer())
return bus
player = MprisPlayer(bus)
bus.export('/org/mpris/MediaPlayer2', player)
return player
return await asyncio.gather(*(setup(name) for name in names))
def get_interfaces(bus):
return bus._path_exports.get('/org/mpris/MediaPlayer2', [])
class MprisPlayer(ServiceInterface):
def __init__(self):
def __init__(self, bus):
super().__init__('org.mpris.MediaPlayer2.Player')
self.counter = 0
self.reset()
self.bus = bus
def reset(self):
# method calls
@@ -57,6 +56,40 @@ class MprisPlayer(ServiceInterface):
# signals
self.seeked_value = 0
async def ping(self):
await self.bus.introspect('org.freedesktop.DBus',
'/org/freedesktop/DBus')
async def set_artist_title(self, artist, title):
self.counter += 1
self.metadata = {
'xesam:title': Variant('s', title),
'xesam:artist': Variant('as', [artist]),
'mpris:trackid': Variant('o', '/' + str(self.counter)),
}
self.emit_properties_changed({
'Metadata': self.metadata,
'PlaybackStatus': self.playback_status,
'CanPlay': True,
})
await self.ping()
async def clear_metadata(self):
self.counter += 1
self.metadata = {
'mpris:trackid': Variant('o', '/' + str(self.counter)),
}
self.emit_properties_changed({
'Metadata': self.metadata,
'PlaybackStatus': self.playback_status,
'CanPlay': True,
})
await self.ping()
def disconnect(self):
self.bus.disconnect()
@method()
def Next(self):
self.next_called = True

38
test/playerctl.py
View File

@@ -9,12 +9,42 @@ class CommandResult:
self.returncode = returncode
class PlayerctlProcess:
def __init__(self, proc):
self.queue = asyncio.Queue()
self.proc = proc
async def reader(stream):
while True:
line = await stream.readline()
if not line:
break
line = line.decode().strip()
if 'playerctl-DEBUG:' in line:
print(line)
else:
await self.queue.put(line)
async def printer(stream):
while True:
line = await stream.readline()
print(line)
if not line:
break
asyncio.get_event_loop().create_task(reader(proc.stdout))
# asyncio.get_event_loop().create_task(printer(proc.stderr))
def running(self):
return self.proc.returncode is None
class PlayerctlCli:
def __init__(self, bus_address=None, debug=False):
self.bus_address = bus_address
self.debug = debug
async def start(self, cmd):
async def _start(self, cmd):
env = os.environ.copy()
shell_cmd = f'playerctl {cmd}'
@@ -29,8 +59,12 @@ class PlayerctlCli:
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE)
async def start(self, cmd):
proc = await self._start(cmd)
return PlayerctlProcess(proc)
async def run(self, cmd):
proc = await self.start(cmd)
proc = await self._start(cmd)
stdout, stderr = await proc.communicate()
await proc.wait()
return CommandResult(stdout, stderr, proc.returncode)

14
test/test_basics.py
View File

@@ -1,4 +1,4 @@
from .mpris import setup_buses
from .mpris import setup_mpris
from .playerctl import PlayerctlCli
import asyncio
@@ -40,10 +40,10 @@ async def test_basics():
@pytest.mark.asyncio
async def test_list_names(bus_address):
[bus1, bus2, bus3] = await setup_buses('basics1',
'basics2',
'basics3',
bus_address=bus_address)
mpris_players = await setup_mpris('basics1',
'basics2',
'basics3',
bus_address=bus_address)
playerctl = PlayerctlCli(bus_address)
result = await playerctl.run('--list-all')
@@ -53,5 +53,5 @@ async def test_list_names(bus_address):
assert 'basics2' in players
assert 'basics3' in players
for bus in [bus1, bus2, bus3]:
bus.disconnect()
for mpris in mpris_players:
mpris.disconnect()

7
test/test_commands.py
View File

@@ -1,4 +1,4 @@
from .mpris import setup_buses, get_interfaces
from .mpris import setup_mpris
from .playerctl import PlayerctlCli
import asyncio
@@ -9,13 +9,12 @@ import pytest
@pytest.mark.asyncio
async def test_commands(bus_address):
[bus] = await setup_buses('commands', bus_address=bus_address)
[interface] = get_interfaces(bus)
[mpris] = await setup_mpris('commands', bus_address=bus_address)
commands = ('play', 'pause', 'play-pause', 'stop', 'next', 'previous')
def get_called(cmd):
return getattr(interface, f'{cmd.replace("-", "_")}_called')
return getattr(mpris, f'{cmd.replace("-", "_")}_called')
playerctl = PlayerctlCli(bus_address)

184
test/test_follow.py
View File

@@ -1,95 +1,23 @@
from .mpris import setup_buses, get_interfaces
from .mpris import setup_mpris
from .playerctl import PlayerctlCli
from dbus_next import Variant
import pytest
import asyncio
class MprisWrapper:
def __init__(self, bus):
self.bus = bus
self.counter = 0
self.iface = get_interfaces(bus)[0]
self.iface.playback_status = 'Playing'
async def ping(self):
await self.bus.introspect('org.freedesktop.DBus',
'/org/freedesktop/DBus')
async def set_metadata(self, artist, title):
self.counter += 1
self.iface.metadata = {
'xesam:title': Variant('s', title),
'xesam:artist': Variant('as', [artist]),
'mpris:trackid': Variant('o', '/' + str(self.counter)),
}
self.iface.emit_properties_changed({
'Metadata': self.iface.metadata,
'PlaybackStatus': self.iface.playback_status,
'CanPlay': True,
})
await self.ping()
async def clear_metadata(self):
self.counter += 1
self.iface.metadata = {
'mpris:trackid': Variant('o', '/' + str(self.counter)),
}
self.iface.emit_properties_changed({
'Metadata': self.iface.metadata,
'PlaybackStatus': self.iface.playback_status,
'CanPlay': True,
})
await self.ping()
class ProcWrapper:
def __init__(self, proc):
self.queue = asyncio.Queue()
self.proc = proc
async def reader(stream):
while True:
line = await stream.readline()
if not line:
break
line = line.decode().strip()
if 'playerctl-DEBUG:' in line:
print(line)
else:
await self.queue.put(line)
async def printer(stream):
while True:
line = await stream.readline()
print(line)
if not line:
break
asyncio.get_event_loop().create_task(reader(proc.stdout))
# asyncio.get_event_loop().create_task(printer(proc.stderr))
def running(self):
return self.proc.returncode == None
@pytest.mark.asyncio
async def test_follow(bus_address):
player1 = 'test1'
[bus1] = await setup_buses(player1, bus_address=bus_address)
mpris1 = MprisWrapper(bus1)
[mpris1] = await setup_mpris(player1, bus_address=bus_address)
playerctl = PlayerctlCli(bus_address)
pctl_cmd = 'metadata --format "{{playerInstance}}: {{artist}} - {{title}}" --follow'
proc = ProcWrapper(await playerctl.start(pctl_cmd))
proc = await playerctl.start(pctl_cmd)
await mpris1.set_metadata('artist', 'title')
await mpris1.set_artist_title('artist', 'title')
line = await proc.queue.get()
assert line == 'test1: artist - title'
await mpris1.set_metadata('artist2', 'title2')
await mpris1.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test1: artist2 - title2'
@@ -97,11 +25,11 @@ async def test_follow(bus_address):
line = await proc.queue.get()
assert line == 'test1: -'
await mpris1.set_metadata('artist3', 'title3')
await mpris1.set_artist_title('artist3', 'title3')
line = await proc.queue.get()
assert line == 'test1: artist3 - title3'
bus1.disconnect()
mpris1.disconnect()
line = await proc.queue.get()
assert line == ''
@@ -113,65 +41,64 @@ async def test_follow_selection(bus_address):
player2 = 'test2'
player3 = 'test3'
player4 = 'test4'
[bus1, bus2, bus3, bus4] = await setup_buses(player1,
player2,
player3,
player4,
bus_address=bus_address)
[mpris1, mpris2, mpris3,
mpris4] = [MprisWrapper(bus) for bus in [bus1, bus2, bus3, bus4]]
mpris4] = await setup_mpris(player1,
player2,
player3,
player4,
bus_address=bus_address)
await mpris1.set_metadata('artist', 'title')
await mpris1.set_artist_title('artist', 'title')
playerctl = PlayerctlCli(bus_address)
pctl_cmd = '--player test3,test2,test1 metadata --format "{{playerInstance}}: {{artist}} - {{title}}" --follow'
proc = ProcWrapper(await playerctl.start(pctl_cmd))
proc = await playerctl.start(pctl_cmd)
line = await proc.queue.get()
assert line == 'test1: artist - title'
# player4 is ignored
await mpris4.set_metadata('artist', 'title')
await mpris4.set_artist_title('artist', 'title')
assert proc.queue.empty()
# setting metadata the same twice doesn't print
await mpris1.set_metadata('artist', 'title')
await mpris1.set_artist_title('artist', 'title')
assert proc.queue.empty()
await mpris2.set_metadata('artist', 'title')
await mpris2.set_artist_title('artist', 'title')
line = await proc.queue.get()
assert line == 'test2: artist - title'
# player2 takes precedence
await mpris1.set_metadata('artist2', 'title2')
await mpris1.set_artist_title('artist2', 'title2')
assert proc.queue.empty()
await mpris3.set_metadata('artist', 'title')
await mpris3.set_artist_title('artist', 'title')
line = await proc.queue.get()
assert line == 'test3: artist - title'
# player 3 takes precedence
await mpris2.set_metadata('artist2', 'title2')
await mpris2.set_artist_title('artist2', 'title2')
assert proc.queue.empty()
# when bus3 disconnects, it should show the next one
bus3.disconnect()
mpris3.disconnect()
await mpris2.ping()
line = await proc.queue.get()
assert line == 'test2: artist2 - title2'
# same for bus2
bus2.disconnect()
mpris2.disconnect()
await mpris1.ping()
line = await proc.queue.get()
assert line == 'test1: artist2 - title2'
bus1.disconnect()
mpris1.disconnect()
line = await proc.queue.get()
assert line == ''
bus4.disconnect()
mpris4.disconnect()
@pytest.mark.asyncio
@@ -180,43 +107,41 @@ async def test_follow_selection_any(bus_address):
player2 = 'test2'
player3 = 'test3'
player4 = 'test4'
[bus1, bus2, bus3, bus4] = await setup_buses(player1,
player2,
player3,
player4,
bus_address=bus_address)
[mpris1, mpris2, mpris3,
mpris4] = [MprisWrapper(bus) for bus in [bus1, bus2, bus3, bus4]]
mpris4] = await setup_mpris(player1,
player2,
player3,
player4,
bus_address=bus_address)
playerctl = PlayerctlCli(bus_address)
pctl_cmd = '--player test3,%any,test1 metadata --format "{{playerInstance}}: {{artist}} - {{title}}" --follow'
proc = ProcWrapper(await playerctl.start(pctl_cmd))
proc = await playerctl.start(pctl_cmd)
# test3 takes first precedence
await mpris3.set_metadata('artist', 'title')
await mpris3.set_artist_title('artist', 'title')
line = await proc.queue.get()
assert line == 'test3: artist - title', proc.queue
await mpris2.set_metadata('artist', 'title')
await mpris2.set_artist_title('artist', 'title')
assert proc.queue.empty()
await mpris1.set_metadata('artist', 'title')
await mpris1.set_artist_title('artist', 'title')
assert proc.queue.empty()
bus3.disconnect()
mpris3.disconnect()
line = await proc.queue.get()
assert line == 'test2: artist - title'
bus2.disconnect()
mpris2.disconnect()
line = await proc.queue.get()
assert line == 'test1: artist - title'
bus1.disconnect()
mpris1.disconnect()
line = await proc.queue.get()
assert line == ''
bus4.disconnect()
mpris4.disconnect()
@pytest.mark.asyncio
@@ -225,64 +150,63 @@ async def test_follow_all_players(bus_address):
player2 = 'test2'
player3 = 'test3'
player4 = 'test4'
[bus1, bus2, bus3, bus4] = await setup_buses(player1,
player2,
player3,
player4,
bus_address=bus_address)
[mpris1, mpris2, mpris3,
mpris4] = [MprisWrapper(bus) for bus in [bus1, bus2, bus3, bus4]]
mpris4] = await setup_mpris(player1,
player2,
player3,
player4,
bus_address=bus_address)
await asyncio.gather(*[
mpris.set_metadata('artist', 'title')
mpris.set_artist_title('artist', 'title')
for mpris in [mpris1, mpris2, mpris3, mpris4]
])
playerctl = PlayerctlCli(bus_address)
pctl_cmd = '--all-players --player test3,test2,test1 metadata --format "{{playerInstance}}: {{artist}} - {{title}}" --follow'
proc = ProcWrapper(await playerctl.start(pctl_cmd))
proc = await playerctl.start(pctl_cmd)
# player4 is ignored
await mpris4.set_metadata('artist', 'title')
await mpris4.set_artist_title('artist', 'title')
assert proc.queue.empty()
# no precedence, just whoever changes metadata last
await mpris1.set_metadata('artist2', 'title2')
await mpris1.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test1: artist2 - title2'
await mpris2.set_metadata('artist2', 'title2')
await mpris2.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test2: artist2 - title2'
await mpris3.set_metadata('artist2', 'title2')
await mpris3.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test3: artist2 - title2'
await mpris2.set_metadata('artist2', 'title2')
await mpris2.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test2: artist2 - title2'
await mpris1.set_metadata('artist2', 'title2')
await mpris1.set_artist_title('artist2', 'title2')
line = await proc.queue.get()
assert line == 'test1: artist2 - title2'
bus1.disconnect()
mpris1.disconnect()
await mpris4.ping()
line = await proc.queue.get()
assert line == 'test2: artist2 - title2'
bus2.disconnect()
mpris2.disconnect()
await mpris4.ping()
line = await proc.queue.get()
assert line == 'test3: artist2 - title2'
bus3.disconnect()
mpris3.disconnect()
await mpris4.ping()
line = await proc.queue.get()
assert line == ''
bus4.disconnect()
mpris4.disconnect()

17
test/test_format.py
View File

@@ -1,21 +1,15 @@
from dbus_next import RequestNameReply, Variant
from dbus_next.aio import MessageBus
from .mpris import MprisPlayer
from dbus_next import Variant
from .mpris import setup_mpris
from .playerctl import PlayerctlCli
import pytest
import asyncio
# TODO: test missing function does not segv
@pytest.mark.asyncio
async def test_format(bus_address):
bus = await MessageBus(bus_address=bus_address).connect()
reply = await bus.request_name(f'org.mpris.MediaPlayer2.format_test')
assert reply == RequestNameReply.PRIMARY_OWNER
mpris = MprisPlayer()
bus.export('/org/mpris/MediaPlayer2', mpris)
[mpris] = await setup_mpris('format-test', bus_address=bus_address)
TITLE = 'A Title'
ARTIST = 'An Artist'
mpris.metadata = {
@@ -24,11 +18,12 @@ async def test_format(bus_address):
'xesam:escapeme': Variant('s', '<hi>'),
'mpris:length': Variant('x', 100000)
}
await mpris.ping()
playerctl = PlayerctlCli(bus_address)
cmd = await playerctl.run('metadata --format "{{artist}} - {{title}}"')
assert cmd.stdout == f'{ARTIST} - {TITLE}'
assert cmd.stdout == f'{ARTIST} - {TITLE}', cmd.stderr
cmd = await playerctl.run(
'metadata --format "{{markup_escape(xesam:escapeme)}}"')
@@ -67,3 +62,5 @@ async def test_format(bus_address):
cmd = await playerctl.run('metadata --format \'{{default("ok", "not")}}\'')
assert cmd.stdout == 'ok', cmd.stderr
mpris.disconnect()

14
test/test_selection.py
View File

@@ -1,7 +1,6 @@
from .mpris import setup_buses
from .mpris import setup_mpris
from .playerctl import PlayerctlCli
import pytest
import asyncio
def selector(bus_address):
@@ -38,7 +37,12 @@ async def test_selection(bus_address):
s6i = 'selection6.instance2'
any_player = '%any'
buses = await setup_buses(s1, s1i, s2, s3, s6i, bus_address=bus_address)
mpris_players = await setup_mpris(s1,
s1i,
s2,
s3,
s6i,
bus_address=bus_address)
# TODO: test ignored players
selections = {
@@ -68,5 +72,5 @@ async def test_selection(bus_address):
result = await select_many(*selection)
assert result == expected
for bus in buses:
bus.disconnect()
for mpris in mpris_players:
mpris.disconnect()