start implementation of the IRC plugin
not functional yet, but this has some of the operations implemented and under test Signed-off-by: Brian S. Stephan <bss@incorporeal.org>
This commit is contained in:
parent
497619e0ff
commit
2730978663
GPG Key ID:
3DE06D3180895FCB
139
idlerpg/ircplugin.py
Normal file
139
idlerpg/ircplugin.py
Normal file
@ -0,0 +1,139 @@
|
||||
"""IRC support for managing and executing the IdleRPG game.
|
||||
|
||||
SPDX-FileCopyrightText: © 2024 Brian S. Stephan <bss@incorporeal.org>
|
||||
SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
"""
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
import irc.client
|
||||
from django.db.utils import IntegrityError
|
||||
from ircbot.lib import Plugin
|
||||
|
||||
from idlerpg.models import Character, Game
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IdleRPG(Plugin):
|
||||
"""Run a game of IdleRPG."""
|
||||
|
||||
SLEEP_BETWEEN_LEVEL_CHECKS = 1
|
||||
|
||||
LOGIN_COMMAND_PATTERN = r'^LOGIN\s+(?P<name>\S+)\s+(?P<password>\S+)$'
|
||||
LOGOUT_COMMAND_PATTERN = r'^LOGOUT$'
|
||||
REGISTER_COMMAND_PATTERN = r'^REGISTER\s+(?P<name>\S+)\s+(?P<password>\S+)\s+(?P<char_class>.*)$'
|
||||
REMOVEME_COMMAND_PATTERN = r'^REMOVEME$'
|
||||
STATUS_COMMAND_PATTERN = r'^(STATUS|WHOAMI)$'
|
||||
|
||||
def __init__(self, bot, connection, event):
|
||||
"""Initialize miscellaneous state stuff."""
|
||||
self.seen_hostmasks = set()
|
||||
super().__init__(bot, connection, event)
|
||||
|
||||
def handle_join(self, connection, event):
|
||||
"""Track a character as online if their player joins the game channel."""
|
||||
self.seen_hostmasks.add(event.source)
|
||||
logger.info("Added %s to the set of seen hostmasks.", event.source)
|
||||
|
||||
def handle_kick_penalty(self, connection, event):
|
||||
"""Penalize characters for their user getting kicked from the channel."""
|
||||
self.seen_hostmasks.discard(event.source)
|
||||
logger.info("Removed %s from the set of seen hostmasks.", event.source)
|
||||
return self._handle_generic_penalty_and_logout(event.source, event.target, 250,
|
||||
"getting kicked from the game channel",
|
||||
'time_penalized_kicked')
|
||||
|
||||
def handle_nick_penalty(self, connection, event):
|
||||
"""Penalize characters for changing their nick while in the channel."""
|
||||
# TODO: figure out how to update the character and seen hostmasks
|
||||
return self._handle_generic_penalty_and_logout(event.source, event.target, 30,
|
||||
"changing their nick",
|
||||
'time_penalized_nick_change')
|
||||
|
||||
def handle_part_penalty(self, connection, event):
|
||||
"""Penalize characters for their user parting."""
|
||||
self.seen_hostmasks.discard(event.source)
|
||||
logger.info("Removed %s from the set of seen hostmasks.", event.source)
|
||||
return self._handle_generic_penalty_and_logout(event.source, event.target, 200,
|
||||
"parting the game channel",
|
||||
'time_penalized_part')
|
||||
|
||||
def handle_message_penalty(self, connection, event):
|
||||
"""Pay attention to messages in the game channel, and penalize talkers."""
|
||||
hostmask = event.source
|
||||
|
||||
logger.debug("looking for %s to try to penalize them", hostmask)
|
||||
try:
|
||||
character = Character.objects.get(enabled=True, hostmask=hostmask)
|
||||
logger.debug("found character %s", character)
|
||||
message = event.arguments[0]
|
||||
penalty = character.penalize(len(message), "sending a privmsg to the game channel")
|
||||
character.time_penalized_privmsg += penalty
|
||||
character.save()
|
||||
except Character.DoesNotExist:
|
||||
logger.debug("no character found for %s", hostmask)
|
||||
return
|
||||
|
||||
def handle_quit_penalty(self, connection, event):
|
||||
"""Penalize characters for their user quitting IRC."""
|
||||
self.seen_hostmasks.discard(event.source)
|
||||
logger.info("Removed %s from the set of seen hostmasks.", event.source)
|
||||
return self._handle_generic_penalty_and_logout(event.source, event.target, 20,
|
||||
"quitting IRC",
|
||||
'time_penalized_quit')
|
||||
|
||||
def handle_login(self, connection, event, match):
|
||||
"""Log in a character when requested, assuming they're online."""
|
||||
hostmask = event.source
|
||||
|
||||
nick = irc.client.NickMask(hostmask).nick
|
||||
try:
|
||||
character = Character.objects.get(enabled=True, name=match.group('name'))
|
||||
character.check_password(match.group('password'))
|
||||
except (Character.DoesNotExist, ValueError):
|
||||
return self.bot.reply(event, "The requested character does not exist, or has been disabled, "
|
||||
"or your password does not match.")
|
||||
|
||||
if hostmask not in self.seen_hostmasks:
|
||||
return self.bot.reply(event, f"Please join {character.game.channel.name} before logging in.")
|
||||
|
||||
if character.status != Character.CHARACTER_STATUS_OFFLINE:
|
||||
return self.bot.reply(event, f"Cannot log in {character.name}, either they already are, "
|
||||
"or they are in a state that cannot be modified.")
|
||||
|
||||
character.log_in(match.group('password'), hostmask)
|
||||
character.save()
|
||||
self.bot.reply(None, f"{character}, is now online from nickname {nick}. "
|
||||
f"Next level at {character.next_level_str()}.",
|
||||
explicit_target=character.game.channel.name)
|
||||
return self.bot.reply(event, f"{character} has been successfully logged in.")
|
||||
|
||||
def _handle_generic_penalty_and_logout(self, hostmask: str, channel: str, penalty: int, reason: str,
|
||||
penalty_log_attr: str):
|
||||
"""Penalize a character and log them out, for a provided reason.
|
||||
|
||||
Args:
|
||||
hostmask: the hostmask for the character in question
|
||||
channel: the game channel the event occurred in
|
||||
penalty: the penalty to apply
|
||||
reason: the reason for the penalty
|
||||
"""
|
||||
logger.debug("looking for %s to try to penalize them", hostmask)
|
||||
try:
|
||||
character = Character.objects.get(enabled=True, hostmask=hostmask)
|
||||
logger.debug("character found for %s", hostmask)
|
||||
seconds = character.penalize(penalty, reason)
|
||||
log = getattr(character, penalty_log_attr)
|
||||
setattr(character, penalty_log_attr, log + seconds)
|
||||
try:
|
||||
character.log_out()
|
||||
except ValueError:
|
||||
logger.debug("tried to log out %s but they already were", character)
|
||||
character.save()
|
||||
except Character.DoesNotExist:
|
||||
logger.debug("no character found for %s", hostmask)
|
||||
|
||||
|
||||
plugin = IdleRPG
|
||||
257
tests/test_idlerpg_ircplugin.py
Normal file
257
tests/test_idlerpg_ircplugin.py
Normal file
@ -0,0 +1,257 @@
|
||||
"""Test IdleRPG character operations.
|
||||
|
||||
SPDX-FileCopyrightText: © 2024 Brian S. Stephan <bss@incorporeal.org>
|
||||
SPDX-License-Identifier: AGPL-3.0-or-later
|
||||
"""
|
||||
import logging
|
||||
import re
|
||||
import unittest.mock as mock
|
||||
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
|
||||
from ircbot.models import IrcChannel, IrcServer
|
||||
from idlerpg.ircplugin import IdleRPG
|
||||
from idlerpg.models import Character, Game
|
||||
|
||||
|
||||
class IrcPluginTest(TestCase):
|
||||
"""Test the IRC integration."""
|
||||
|
||||
fixtures = ['tests/fixtures/simple_character.json']
|
||||
|
||||
def setUp(self):
|
||||
"""Create common objects."""
|
||||
self.mock_bot = mock.MagicMock()
|
||||
self.mock_connection = mock.MagicMock()
|
||||
|
||||
self.mock_connection.get_nickname.return_value = 'test_bot'
|
||||
self.mock_connection.server_config = IrcServer.objects.get(pk=1)
|
||||
|
||||
self.plugin = IdleRPG(self.mock_bot, self.mock_connection, mock.MagicMock())
|
||||
self.game = Game.objects.get(pk=1)
|
||||
|
||||
def test_kick_penalty(self):
|
||||
"""Test that if a character is kicked from the game channel, they get penalized."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_kick_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testkickpen', self.game, 'test',
|
||||
'bss!bss@test_kick_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out') as mock_log_out:
|
||||
self.plugin.handle_kick_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_called_with(250, "getting kicked from the game channel")
|
||||
mock_log_out.assert_called()
|
||||
test_char = Character.objects.get(name='testkickpen')
|
||||
assert test_char.time_penalized_kicked == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_generic_penalty_no_penalty_for_unknown(self):
|
||||
"""Test that a kick from an unknown hostmask is ignored."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_kick_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out') as mock_log_out:
|
||||
self.plugin.handle_kick_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_not_called()
|
||||
mock_log_out.assert_not_called()
|
||||
|
||||
def test_nick_penalty(self):
|
||||
"""Test that if a user changes their nick, they get penalized."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_nick_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testnickpen', self.game, 'test',
|
||||
'bss!bss@test_nick_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out') as mock_log_out:
|
||||
self.plugin.handle_nick_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_called_with(30, "changing their nick")
|
||||
mock_log_out.assert_called()
|
||||
test_char = Character.objects.get(name='testnickpen')
|
||||
assert test_char.time_penalized_nick_change == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_part_penalty(self):
|
||||
"""Test that if a character parts from the game channel, they get penalized."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_part_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testpartpen', self.game, 'test',
|
||||
'bss!bss@test_part_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out') as mock_log_out:
|
||||
self.plugin.handle_part_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_called_with(200, "parting the game channel")
|
||||
mock_log_out.assert_called()
|
||||
test_char = Character.objects.get(name='testpartpen')
|
||||
assert test_char.time_penalized_part == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_pubmsg_penalty(self):
|
||||
"""Test that if a character speaks in the game channel, they get penalized."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.arguments = ['this is a test message']
|
||||
mock_event.source = 'bss!bss@test_pubmsg_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testpubmsgpen', self.game, 'test',
|
||||
'bss!bss@test_pubmsg_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
self.plugin.handle_message_penalty(self.mock_connection, mock_event)
|
||||
|
||||
# 22 is the len of the message
|
||||
mock_penalize.assert_called_with(22, "sending a privmsg to the game channel")
|
||||
test_char = Character.objects.get(name='testpubmsgpen')
|
||||
assert test_char.time_penalized_privmsg == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_pubmsg_no_penalty_for_unknown(self):
|
||||
"""Test that a speakerwith an unknown hostmask is ignored."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.arguments = ['this is a test message']
|
||||
mock_event.source = 'bss!bss@some_user'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
self.plugin.handle_message_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_not_called()
|
||||
|
||||
def test_join(self):
|
||||
"""Test that joiners update the seen hostmasks."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.arguments = ['this is a test message']
|
||||
mock_event.source = 'test!test@test_join'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
assert 'test!test@test_join' not in self.plugin.seen_hostmasks
|
||||
self.plugin.handle_join(self.mock_connection, mock_event)
|
||||
assert 'test!test@test_join' in self.plugin.seen_hostmasks
|
||||
|
||||
def test_quit_penalty(self):
|
||||
"""Test that if a character quits from the game channel, they get penalized."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_quit_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testquitpen', self.game, 'test',
|
||||
'bss!bss@test_quit_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out') as mock_log_out:
|
||||
self.plugin.handle_quit_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_called_with(20, "quitting IRC")
|
||||
mock_log_out.assert_called()
|
||||
test_char = Character.objects.get(name='testquitpen')
|
||||
assert test_char.time_penalized_quit == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_quit_penalty_even_if_logged_out(self):
|
||||
"""Test that the penalty applies even if the character is already logged out."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_quit_penalty'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
# make a character so as to not disturb other tests
|
||||
test_char = Character.objects.register('testquitpen', self.game, 'test',
|
||||
'bss!bss@test_quit_penalty', 'tester')
|
||||
with mock.patch('idlerpg.models.Character.penalize', return_value=5) as mock_penalize:
|
||||
with mock.patch('idlerpg.models.Character.log_out', side_effect=ValueError) as mock_log_out:
|
||||
self.plugin.handle_quit_penalty(self.mock_connection, mock_event)
|
||||
|
||||
mock_penalize.assert_called_with(20, "quitting IRC")
|
||||
mock_log_out.assert_called()
|
||||
test_char = Character.objects.get(name='testquitpen')
|
||||
assert test_char.time_penalized_quit == 5
|
||||
|
||||
test_char.delete()
|
||||
|
||||
def test_login_character_not_found(self):
|
||||
"""Test the LOGIN command sent to the bot."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_login'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
match = re.match(IdleRPG.LOGIN_COMMAND_PATTERN, 'LOGIN nope bss')
|
||||
self.plugin.handle_login(self.mock_connection, mock_event, match)
|
||||
|
||||
self.mock_bot.reply.assert_called_once_with(
|
||||
mock_event,
|
||||
"The requested character does not exist, or has been disabled, or your password does not match."
|
||||
)
|
||||
|
||||
def test_login_bad_password(self):
|
||||
"""Test the LOGIN command sent to the bot."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_login'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
match = re.match(IdleRPG.LOGIN_COMMAND_PATTERN, 'LOGIN bss nope')
|
||||
self.plugin.handle_login(self.mock_connection, mock_event, match)
|
||||
|
||||
self.mock_bot.reply.assert_called_once_with(
|
||||
mock_event,
|
||||
"The requested character does not exist, or has been disabled, or your password does not match."
|
||||
)
|
||||
|
||||
def test_login_not_seen(self):
|
||||
"""Test the LOGIN command sent to the bot."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_login'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
match = re.match(IdleRPG.LOGIN_COMMAND_PATTERN, 'LOGIN bss bss')
|
||||
self.plugin.handle_login(self.mock_connection, mock_event, match)
|
||||
|
||||
self.mock_bot.reply.assert_called_once_with(mock_event, "Please join #test before logging in.")
|
||||
|
||||
def test_login_not_logged_off(self):
|
||||
"""Test the LOGIN command sent to the bot."""
|
||||
mock_event = mock.MagicMock()
|
||||
mock_event.source = 'bss!bss@test_login'
|
||||
mock_event.target = '#test'
|
||||
mock_event.recursing = False
|
||||
|
||||
self.plugin.seen_hostmasks.add('bss!bss@test_login')
|
||||
match = re.match(IdleRPG.LOGIN_COMMAND_PATTERN, 'LOGIN bss bss')
|
||||
self.plugin.handle_login(self.mock_connection, mock_event, match)
|
||||
|
||||
self.mock_bot.reply.assert_called_once_with(
|
||||
mock_event,
|
||||
"Cannot log in bss, either they already are, or they are in a state that cannot be modified."
|
||||
)
|
||||
|
||||
self.plugin.seen_hostmasks.remove('bss!bss@test_login')
|
||||
Loading…
Reference in New Issue
Block a user