# This file is part of beets.
# Copyright 2016, Adrian Sampson.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
"""Tests for BPD's implementation of the MPD protocol.
"""
import asyncio
import importlib.util
import multiprocessing as mp
import os
import socket
import sys
import tempfile
import threading
import unittest
from contextlib import contextmanager
# Mock GstPlayer so that the forked process doesn't attempt to import gi:
from unittest import mock
import confuse
import yaml
from beets.test.helper import TestHelper
from beets.util import bluelet, py3_path
from beetsplug import bpd
gstplayer = importlib.util.module_from_spec(
importlib.util.find_spec("beetsplug.bpd.gstplayer")
)
def _gstplayer_play(*_): # noqa: 42
bpd.gstplayer._GstPlayer.playing = True
return mock.DEFAULT
gstplayer._GstPlayer = mock.MagicMock(
spec_set=[
"time",
"volume",
"playing",
"run",
"play_file",
"pause",
"stop",
"seek",
"play",
"get_decoders",
],
**{
"playing": False,
"volume": 0,
"time.return_value": (0, 0),
"play_file.side_effect": _gstplayer_play,
"play.side_effect": _gstplayer_play,
"get_decoders.return_value": {"default": ({"audio/mpeg"}, {"mp3"})},
},
)
gstplayer.GstPlayer = lambda _: gstplayer._GstPlayer
sys.modules["beetsplug.bpd.gstplayer"] = gstplayer
bpd.gstplayer = gstplayer
class CommandParseTest(unittest.TestCase):
def test_no_args(self):
s = r"command"
c = bpd.Command(s)
self.assertEqual(c.name, "command")
self.assertEqual(c.args, [])
def test_one_unquoted_arg(self):
s = r"command hello"
c = bpd.Command(s)
self.assertEqual(c.name, "command")
self.assertEqual(c.args, ["hello"])
def test_two_unquoted_args(self):
s = r"command hello there"
c = bpd.Command(s)
self.assertEqual(c.name, "command")
self.assertEqual(c.args, ["hello", "there"])
def test_one_quoted_arg(self):
s = r'command "hello there"'
c = bpd.Command(s)
self.assertEqual(c.name, "command")
self.assertEqual(c.args, ["hello there"])
def test_heterogenous_args(self):
s = r'command "hello there" sir'
c = bpd.Command(s)
self.assertEqual(c.name, "command")
self.assertEqual(c.args, ["hello there", "sir"])
def test_quote_in_arg(self):
s = r'command "hello \" there"'
c = bpd.Command(s)
self.assertEqual(c.args, ['hello " there'])
def test_backslash_in_arg(self):
s = r'command "hello \\ there"'