6a7b7bc36a
This is a huge configuration refactoring. The configuration system now uses Pydantic settings, with better validation, better support for environment variables and other goodies. The main config files now use TOML instead of INI, this allow for better handling of complex types, notably dict. Plugins, components and frontends can declare their config settings in a class which will be imported dynamically.
521 lines
21 KiB
Python
521 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
|
|
|
|
# SAT: a jabber client
|
|
# Copyright (C) 2009, 2010, 2011, 2012, 2013 Jérôme Poisson (goffi@goffi.org)
|
|
# Copyright (C) 2013 Adrien Cossa (souliane@mailoo.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
"""Tests for the plugin radiocol"""
|
|
|
|
from libervia.backend.core import exceptions
|
|
from libervia.backend.test import helpers, helpers_plugins
|
|
from libervia.backend.plugins import plugin_misc_radiocol as plugin
|
|
from libervia.backend.plugins import plugin_misc_room_game as plugin_room_game
|
|
from .constants import Const
|
|
|
|
from twisted.words.protocols.jabber.jid import JID
|
|
from twisted.words.xish import domish
|
|
from twisted.internet import reactor
|
|
from twisted.internet import defer
|
|
from twisted.python.failure import Failure
|
|
from twisted.trial.unittest import SkipTest
|
|
|
|
try:
|
|
from mutagen.oggvorbis import OggVorbis
|
|
from mutagen.mp3 import MP3
|
|
from mutagen.easyid3 import EasyID3
|
|
from mutagen.id3 import ID3NoHeaderError
|
|
except ImportError:
|
|
raise exceptions.MissingModule(
|
|
"Missing module Mutagen, please download/install from https://bitbucket.org/lazka/mutagen"
|
|
)
|
|
|
|
import uuid
|
|
import os
|
|
import copy
|
|
import shutil
|
|
|
|
|
|
ROOM_JID = JID(Const.MUC_STR[0])
|
|
PROFILE = Const.PROFILE[0]
|
|
REFEREE_FULL = JID(ROOM_JID.userhost() + "/" + Const.JID[0].user)
|
|
PLAYERS_INDICES = [0, 1, 3] # referee included
|
|
OTHER_PROFILES = [Const.PROFILE[1], Const.PROFILE[3]]
|
|
OTHER_PLAYERS = [Const.JID[1], Const.JID[3]]
|
|
|
|
|
|
class RadiocolTest(helpers.SatTestCase):
|
|
def setUp(self):
|
|
self.host = helpers.FakeSAT()
|
|
|
|
def reinit(self):
|
|
self.host.reinit()
|
|
self.host.plugins["ROOM-GAME"] = plugin_room_game.RoomGame(self.host)
|
|
self.plugin = plugin.Radiocol(self.host) # must be init after ROOM-GAME
|
|
self.plugin.testing = True
|
|
self.plugin_0045 = self.host.plugins["XEP-0045"] = helpers_plugins.FakeXEP_0045(
|
|
self.host
|
|
)
|
|
self.plugin_0249 = self.host.plugins["XEP-0249"] = helpers_plugins.FakeXEP_0249(
|
|
self.host
|
|
)
|
|
for profile in Const.PROFILE:
|
|
self.host.get_client(profile) # init self.host.profiles[profile]
|
|
self.songs = []
|
|
self.playlist = []
|
|
assert self.host.memory.config.common.media_dir is not None
|
|
self.sound_dir = str(self.host.memory.config.common.media_dir / "/test/sound/")
|
|
try:
|
|
for filename in os.listdir(self.sound_dir):
|
|
if filename.endswith(".ogg") or filename.endswith(".mp3"):
|
|
self.songs.append(filename)
|
|
except OSError:
|
|
raise SkipTest("The sound samples in sat_media/test/sound were not found")
|
|
|
|
def _build_players(self, players=[]):
|
|
"""@return: the "started" content built with the given players"""
|
|
content = "<started"
|
|
if not players:
|
|
content += "/>"
|
|
else:
|
|
content += ">"
|
|
for i in range(0, len(players)):
|
|
content += "<player index='%s'>%s</player>" % (i, players[i])
|
|
content += "</started>"
|
|
return content
|
|
|
|
def _expected_message(self, to_jid, type_, content):
|
|
"""
|
|
@param to_jid: recipient full jid
|
|
@param type_: message type ('normal' or 'groupchat')
|
|
@param content: content as unicode or list of domish elements
|
|
@return: the message XML built from the given recipient, message type and content
|
|
"""
|
|
if isinstance(content, list):
|
|
new_content = copy.deepcopy(content)
|
|
for element in new_content:
|
|
if not element.hasAttribute("xmlns"):
|
|
element["xmlns"] = ""
|
|
content = "".join([element.toXml() for element in new_content])
|
|
return "<message to='%s' type='%s'><%s xmlns='%s'>%s</%s></message>" % (
|
|
to_jid.full(),
|
|
type_,
|
|
plugin.RADIOC_TAG,
|
|
plugin.NC_RADIOCOL,
|
|
content,
|
|
plugin.RADIOC_TAG,
|
|
)
|
|
|
|
def _reject_song_cb(self, profile_index):
|
|
"""Check if the message "song_rejected" has been sent by the referee
|
|
and process the command with the profile of the uploader
|
|
@param profile_index: uploader's profile"""
|
|
sent = self.host.get_sent_message(0)
|
|
content = "<song_rejected xmlns='' reason='Too many songs in queue'/>"
|
|
self.assert_equal_xml(
|
|
sent.toXml(),
|
|
self._expected_message(
|
|
JID(
|
|
ROOM_JID.userhost()
|
|
+ "/"
|
|
+ self.plugin_0045.get_nick(0, profile_index),
|
|
"normal",
|
|
content,
|
|
)
|
|
),
|
|
)
|
|
self._room_game_cmd(
|
|
sent, ["radiocol_song_rejected", ROOM_JID.full(), "Too many songs in queue"]
|
|
)
|
|
|
|
def _no_upload_cb(self):
|
|
"""Check if the message "no_upload" has been sent by the referee
|
|
and process the command with the profiles of each room users"""
|
|
sent = self.host.get_sent_message(0)
|
|
content = "<no_upload xmlns=''/>"
|
|
self.assert_equal_xml(
|
|
sent.toXml(), self._expected_message(ROOM_JID, "groupchat", content)
|
|
)
|
|
self._room_game_cmd(sent, ["radiocol_no_upload", ROOM_JID.full()])
|
|
|
|
def _upload_ok_cb(self):
|
|
"""Check if the message "upload_ok" has been sent by the referee
|
|
and process the command with the profiles of each room users"""
|
|
sent = self.host.get_sent_message(0)
|
|
content = "<upload_ok xmlns=''/>"
|
|
self.assert_equal_xml(
|
|
sent.toXml(), self._expected_message(ROOM_JID, "groupchat", content)
|
|
)
|
|
self._room_game_cmd(sent, ["radiocol_upload_ok", ROOM_JID.full()])
|
|
|
|
def _preload_cb(self, attrs, profile_index):
|
|
"""Check if the message "preload" has been sent by the referee
|
|
and process the command with the profiles of each room users
|
|
@param attrs: information dict about the song
|
|
@param profile_index: profile index of the uploader
|
|
"""
|
|
sent = self.host.get_sent_message(0)
|
|
attrs["sender"] = self.plugin_0045.get_nick(0, profile_index)
|
|
radiocol_elt = next(domish.generateElementsNamed(sent.elements(), "radiocol"))
|
|
preload_elt = next(
|
|
domish.generateElementsNamed(radiocol_elt.elements(), "preload")
|
|
)
|
|
attrs["timestamp"] = preload_elt["timestamp"] # we could not guess it...
|
|
content = "<preload xmlns='' %s/>" % " ".join(
|
|
["%s='%s'" % (attr, attrs[attr]) for attr in attrs]
|
|
)
|
|
if sent.hasAttribute("from"):
|
|
del sent["from"]
|
|
self.assert_equal_xml(
|
|
sent.toXml(), self._expected_message(ROOM_JID, "groupchat", content)
|
|
)
|
|
self._room_game_cmd(
|
|
sent,
|
|
[
|
|
"radiocol_preload",
|
|
ROOM_JID.full(),
|
|
attrs["timestamp"],
|
|
attrs["filename"],
|
|
attrs["title"],
|
|
attrs["artist"],
|
|
attrs["album"],
|
|
attrs["sender"],
|
|
],
|
|
)
|
|
|
|
def _play_next_song_cb(self):
|
|
"""Check if the message "play" has been sent by the referee
|
|
and process the command with the profiles of each room users"""
|
|
sent = self.host.get_sent_message(0)
|
|
filename = self.playlist.pop(0)
|
|
content = "<play xmlns='' filename='%s' />" % filename
|
|
self.assert_equal_xml(
|
|
sent.toXml(), self._expected_message(ROOM_JID, "groupchat", content)
|
|
)
|
|
self._room_game_cmd(sent, ["radiocol_play", ROOM_JID.full(), filename])
|
|
|
|
game_data = self.plugin.games[ROOM_JID]
|
|
if len(game_data["queue"]) == plugin.QUEUE_LIMIT - 1:
|
|
self._upload_ok_cb()
|
|
|
|
def _add_song_cb(self, d, filepath, profile_index):
|
|
"""Check if the message "song_added" has been sent by the uploader
|
|
and process the command with the profile of the referee
|
|
@param d: deferred value or failure got from self.plugin.radiocol_song_added
|
|
@param filepath: full path to the sound file
|
|
@param profile_index: the profile index of the uploader
|
|
"""
|
|
if isinstance(d, Failure):
|
|
self.fail("OGG or MP3 song could not be added!")
|
|
|
|
game_data = self.plugin.games[ROOM_JID]
|
|
|
|
# this is copied from the plugin
|
|
if filepath.lower().endswith(".mp3"):
|
|
actual_song = MP3(filepath)
|
|
try:
|
|
song = EasyID3(filepath)
|
|
|
|
class Info(object):
|
|
def __init__(self, length):
|
|
self.length = length
|
|
|
|
song.info = Info(actual_song.info.length)
|
|
except ID3NoHeaderError:
|
|
song = actual_song
|
|
else:
|
|
song = OggVorbis(filepath)
|
|
|
|
attrs = {
|
|
"filename": os.path.basename(filepath),
|
|
"title": song.get("title", ["Unknown"])[0],
|
|
"artist": song.get("artist", ["Unknown"])[0],
|
|
"album": song.get("album", ["Unknown"])[0],
|
|
"length": str(song.info.length),
|
|
}
|
|
self.assertEqual(game_data["to_delete"][attrs["filename"]], filepath)
|
|
|
|
content = "<song_added xmlns='' %s/>" % " ".join(
|
|
["%s='%s'" % (attr, attrs[attr]) for attr in attrs]
|
|
)
|
|
sent = self.host.get_sent_message(profile_index)
|
|
self.assert_equal_xml(
|
|
sent.toXml(), self._expected_message(REFEREE_FULL, "normal", content)
|
|
)
|
|
|
|
reject_song = len(game_data["queue"]) >= plugin.QUEUE_LIMIT
|
|
no_upload = len(game_data["queue"]) + 1 >= plugin.QUEUE_LIMIT
|
|
play_next = (
|
|
not game_data["playing"]
|
|
and len(game_data["queue"]) + 1 == plugin.QUEUE_TO_START
|
|
)
|
|
|
|
self._room_game_cmd(sent, profile_index) # queue unchanged or +1
|
|
if reject_song:
|
|
self._reject_song_cb(profile_index)
|
|
return
|
|
if no_upload:
|
|
self._no_upload_cb()
|
|
self._preload_cb(attrs, profile_index)
|
|
self.playlist.append(attrs["filename"])
|
|
if play_next:
|
|
self._play_next_song_cb() # queue -1
|
|
|
|
def _room_game_cmd(self, sent, from_index=0, call=[]):
|
|
"""Process a command. It is also possible to call this method as
|
|
_room_game_cmd(sent, call) instead of _room_game_cmd(sent, from_index, call).
|
|
If from index is a list, it is assumed that it is containing the value
|
|
for call and from_index will take its default value.
|
|
@param sent: the sent message that we need to process
|
|
@param from_index: index of the message sender
|
|
@param call: list containing the name of the expected bridge call
|
|
followed by its arguments, or empty list if no call is expected
|
|
"""
|
|
if isinstance(from_index, list):
|
|
call = from_index
|
|
from_index = 0
|
|
|
|
sent["from"] = ROOM_JID.full() + "/" + self.plugin_0045.get_nick(0, from_index)
|
|
recipient = JID(sent["to"]).resource
|
|
|
|
# The message could have been sent to a room user (room_jid + '/' + nick),
|
|
# but when it is received, the 'to' attribute of the message has been
|
|
# changed to the recipient own JID. We need to simulate that here.
|
|
if recipient:
|
|
room = self.plugin_0045.get_room(0, 0)
|
|
sent["to"] = (
|
|
Const.JID_STR[0]
|
|
if recipient == room.nick
|
|
else room.roster[recipient].entity.full()
|
|
)
|
|
|
|
for index in range(0, len(Const.PROFILE)):
|
|
nick = self.plugin_0045.get_nick(0, index)
|
|
if nick:
|
|
if not recipient or nick == recipient:
|
|
if call and (
|
|
self.plugin.is_player(ROOM_JID, nick)
|
|
or call[0] == "radiocol_started"
|
|
):
|
|
args = copy.deepcopy(call)
|
|
args.append(Const.PROFILE[index])
|
|
self.host.bridge.expect_call(*args)
|
|
self.plugin.room_game_cmd(sent, Const.PROFILE[index])
|
|
|
|
def _sync_cb(self, sync_data, profile_index):
|
|
"""Synchronize one player when he joins a running game.
|
|
@param sync_data: result from self.plugin.getSyncData
|
|
@param profile_index: index of the profile to be synchronized
|
|
"""
|
|
for nick in sync_data:
|
|
expected = self._expected_message(
|
|
JID(ROOM_JID.userhost() + "/" + nick), "normal", sync_data[nick]
|
|
)
|
|
sent = self.host.get_sent_message(0)
|
|
self.assert_equal_xml(sent.toXml(), expected)
|
|
for elt in sync_data[nick]:
|
|
if elt.name == "preload":
|
|
self.host.bridge.expect_call(
|
|
"radiocol_preload",
|
|
ROOM_JID.full(),
|
|
elt["timestamp"],
|
|
elt["filename"],
|
|
elt["title"],
|
|
elt["artist"],
|
|
elt["album"],
|
|
elt["sender"],
|
|
Const.PROFILE[profile_index],
|
|
)
|
|
elif elt.name == "play":
|
|
self.host.bridge.expect_call(
|
|
"radiocol_play",
|
|
ROOM_JID.full(),
|
|
elt["filename"],
|
|
Const.PROFILE[profile_index],
|
|
)
|
|
elif elt.name == "no_upload":
|
|
self.host.bridge.expect_call(
|
|
"radiocol_no_upload",
|
|
ROOM_JID.full(),
|
|
Const.PROFILE[profile_index],
|
|
)
|
|
sync_data[nick]
|
|
self._room_game_cmd(sent, [])
|
|
|
|
def _join_room(self, room, nicks, player_index, sync=True):
|
|
"""Make a player join a room and update the list of nicks
|
|
@param room: wokkel.muc.Room instance from the referee perspective
|
|
@param nicks: list of the players which will be updated
|
|
@param player_index: profile index of the new player
|
|
@param sync: set to True to synchronize data
|
|
"""
|
|
user_nick = self.plugin_0045.join_room(0, player_index)
|
|
self.plugin.user_joined_trigger(room, room.roster[user_nick], PROFILE)
|
|
if player_index not in PLAYERS_INDICES:
|
|
# this user is actually not a player
|
|
self.assertFalse(self.plugin.is_player(ROOM_JID, user_nick))
|
|
to_jid, type_ = (JID(ROOM_JID.userhost() + "/" + user_nick), "normal")
|
|
else:
|
|
# this user is a player
|
|
self.assertTrue(self.plugin.is_player(ROOM_JID, user_nick))
|
|
nicks.append(user_nick)
|
|
to_jid, type_ = (ROOM_JID, "groupchat")
|
|
|
|
# Check that the message "players" has been sent by the referee
|
|
expected = self._expected_message(to_jid, type_, self._build_players(nicks))
|
|
sent = self.host.get_sent_message(0)
|
|
self.assert_equal_xml(sent.toXml(), expected)
|
|
|
|
# Process the command with the profiles of each room users
|
|
self._room_game_cmd(
|
|
sent,
|
|
[
|
|
"radiocol_started",
|
|
ROOM_JID.full(),
|
|
REFEREE_FULL.full(),
|
|
nicks,
|
|
[plugin.QUEUE_TO_START, plugin.QUEUE_LIMIT],
|
|
],
|
|
)
|
|
|
|
if sync:
|
|
self._sync_cb(self.plugin._get_sync_data(ROOM_JID, [user_nick]), player_index)
|
|
|
|
def _leave_room(self, room, nicks, player_index):
|
|
"""Make a player leave a room and update the list of nicks
|
|
@param room: wokkel.muc.Room instance from the referee perspective
|
|
@param nicks: list of the players which will be updated
|
|
@param player_index: profile index of the new player
|
|
"""
|
|
user_nick = self.plugin_0045.get_nick(0, player_index)
|
|
user = room.roster[user_nick]
|
|
self.plugin_0045.leave_room(0, player_index)
|
|
self.plugin.user_left_trigger(room, user, PROFILE)
|
|
nicks.remove(user_nick)
|
|
|
|
def _upload_song(self, song_index, profile_index):
|
|
"""Upload the song of index song_index (modulo self.songs size) from the profile of index profile_index.
|
|
|
|
@param song_index: index of the song or None to test with non existing file
|
|
@param profile_index: index of the uploader's profile
|
|
"""
|
|
if song_index is None:
|
|
dst_filepath = str(uuid.uuid1())
|
|
expect_io_error = True
|
|
else:
|
|
song_index = song_index % len(self.songs)
|
|
src_filename = self.songs[song_index]
|
|
dst_filepath = "/tmp/%s%s" % (uuid.uuid1(), os.path.splitext(src_filename)[1])
|
|
shutil.copy(self.sound_dir + src_filename, dst_filepath)
|
|
expect_io_error = False
|
|
|
|
try:
|
|
d = self.plugin.radiocol_song_added(
|
|
REFEREE_FULL, dst_filepath, Const.PROFILE[profile_index]
|
|
)
|
|
except IOError:
|
|
self.assertTrue(expect_io_error)
|
|
return
|
|
|
|
self.assertFalse(expect_io_error)
|
|
cb = lambda defer: self._add_song_cb(defer, dst_filepath, profile_index)
|
|
|
|
def eb(failure):
|
|
if not isinstance(failure, Failure):
|
|
self.fail("Adding a song which is not OGG nor MP3 should fail!")
|
|
self.assertEqual(failure.value.__class__, exceptions.DataError)
|
|
|
|
if src_filename.endswith(".ogg") or src_filename.endswith(".mp3"):
|
|
d.addCallbacks(cb, cb)
|
|
else:
|
|
d.addCallbacks(eb, eb)
|
|
|
|
def test_init(self):
|
|
self.reinit()
|
|
self.assertEqual(self.plugin.invite_mode, self.plugin.FROM_PLAYERS)
|
|
self.assertEqual(self.plugin.wait_mode, self.plugin.FOR_NONE)
|
|
self.assertEqual(self.plugin.join_mode, self.plugin.INVITED)
|
|
self.assertEqual(self.plugin.ready_mode, self.plugin.FORCE)
|
|
|
|
def test_game(self):
|
|
self.reinit()
|
|
|
|
# create game
|
|
self.plugin.prepare_room(OTHER_PLAYERS, ROOM_JID, PROFILE)
|
|
self.assertTrue(self.plugin._game_exists(ROOM_JID, True))
|
|
room = self.plugin_0045.get_room(0, 0)
|
|
nicks = [self.plugin_0045.get_nick(0, 0)]
|
|
|
|
sent = self.host.get_sent_message(0)
|
|
self.assert_equal_xml(
|
|
sent.toXml(),
|
|
self._expected_message(ROOM_JID, "groupchat", self._build_players(nicks)),
|
|
)
|
|
self._room_game_cmd(
|
|
sent,
|
|
[
|
|
"radiocol_started",
|
|
ROOM_JID.full(),
|
|
REFEREE_FULL.full(),
|
|
nicks,
|
|
[plugin.QUEUE_TO_START, plugin.QUEUE_LIMIT],
|
|
],
|
|
)
|
|
|
|
self._join_room(room, nicks, 1) # player joins
|
|
self._join_room(room, nicks, 4) # user not playing joins
|
|
|
|
song_index = 0
|
|
self._upload_song(
|
|
song_index, 0
|
|
) # ogg or mp3 file should exist in sat_media/test/song
|
|
self._upload_song(None, 0) # non existing file
|
|
|
|
# another songs are added by Const.JID[1] until the radio starts + 1 to fill the queue
|
|
# when the first song starts + 1 to be rejected because the queue is full
|
|
for song_index in range(1, plugin.QUEUE_TO_START + 1):
|
|
self._upload_song(song_index, 1)
|
|
|
|
self.plugin.play_next(Const.MUC[0], PROFILE) # simulate the end of the first song
|
|
self._play_next_song_cb()
|
|
self._upload_song(
|
|
song_index, 1
|
|
) # now the song is accepted and the queue is full again
|
|
|
|
self._join_room(room, nicks, 3) # new player joins
|
|
|
|
self.plugin.play_next(Const.MUC[0], PROFILE) # the second song finishes
|
|
self._play_next_song_cb()
|
|
self._upload_song(0, 3) # the player who recently joined re-upload the first file
|
|
|
|
self._leave_room(room, nicks, 1) # one player leaves
|
|
self._join_room(room, nicks, 1) # and join again
|
|
|
|
self.plugin.play_next(Const.MUC[0], PROFILE) # empty the queue
|
|
self._play_next_song_cb()
|
|
self.plugin.play_next(Const.MUC[0], PROFILE)
|
|
self._play_next_song_cb()
|
|
|
|
for filename in self.playlist:
|
|
self.plugin.delete_file("/tmp/" + filename)
|
|
|
|
return defer.succeed(None)
|
|
|
|
def tearDown(self, *args, **kwargs):
|
|
"""Clean the reactor"""
|
|
helpers.SatTestCase.tearDown(self, *args, **kwargs)
|
|
for delayed_call in reactor.getDelayedCalls():
|
|
delayed_call.cancel()
|