sopel/test/test_irc.py

148 lines
3.5 KiB
Python
Raw Normal View History

2017-11-22 19:26:40 -05:00
# coding=utf8
"""Tests for message formatting"""
from __future__ import unicode_literals
import pytest
import asynchat
import os
import shutil
import socket
import select
import tempfile
import threading
import time
import asyncore
from sopel import irc
from sopel.tools import stderr, Identifier
import sopel.config as conf
HOST = '127.0.0.1'
SERVER_QUIT = 'QUIT'
class BasicServer(asyncore.dispatcher):
def __init__(self, address, handler):
asyncore.dispatcher.__init__(self)
self.response_handler = handler
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.bind(address)
self.address = self.socket.getsockname()
self.listen(1)
return
def handle_accept(self):
# Called when a client connects to our socket
client_info = self.accept()
BasicHandler(sock=client_info[0], handler=self.response_handler)
self.handle_close()
return
def handle_close(self):
self.close()
class BasicHandler(asynchat.async_chat):
ac_in_buffer_size = 512
ac_out_buffer_size = 512
def __init__(self, sock, handler):
self.received_data = []
asynchat.async_chat.__init__(self, sock)
self.handler_function = handler
self.set_terminator(b'\n')
return
def collect_incoming_data(self, data):
self.received_data.append(data.decode('utf-8'))
def found_terminator(self):
self._process_command()
def _process_command(self):
command = ''.join(self.received_data)
response = self.handler_function(self, command)
self.push(':fake.server {}\n'.format(response).encode())
self.received_data = []
def start_server(rpl_function=None):
def rpl_func(msg):
print(msg)
return msg
if rpl_function is None:
rpl_function = rpl_func
address = ('localhost', 0) # let the kernel give us a port
server = BasicServer(address, rpl_function)
return server
@pytest.fixture
def bot(request):
cfg_dir = tempfile.mkdtemp()
print(cfg_dir)
filename = tempfile.mkstemp(dir=cfg_dir)[1]
os.mkdir(os.path.join(cfg_dir, 'modules'))
def fin():
print('teardown config file')
shutil.rmtree(cfg_dir)
request.addfinalizer(fin)
def gen(data):
with open(filename, 'w') as fileo:
fileo.write(data)
cfg = conf.Config(filename)
irc_bot = irc.Bot(cfg)
irc_bot.config = cfg
return irc_bot
return gen
def test_bot_init(bot):
test_bot = bot(
'[core]\n'
'owner=Baz\n'
'nick=Foo\n'
'user=Bar\n'
'name=Sopel\n'
)
assert test_bot.nick == Identifier('Foo')
assert test_bot.user == 'Bar'
assert test_bot.name == 'Sopel'
def basic_irc_replies(server, msg):
if msg.startswith('NICK'):
return '001 Foo :Hello'
elif msg.startswith('USER'):
# Quit here because good enough
server.close()
elif msg.startswith('PING'):
return 'PONG{}'.format(msg.replace('PING','',1))
elif msg.startswith('CAP'):
return 'CAP * :'
elif msg.startswith('QUIT'):
server.close()
else:
return '421 {} :Unknown command'.format(msg)
def test_bot_connect(bot):
test_bot = bot(
'[core]\n'
'owner=Baz\n'
'nick=Foo\n'
'user=Bar\n'
'name=Sopel\n'
'host=127.0.0.1\n'
'timeout=10\n'
)
s = start_server(basic_irc_replies)
# Do main run
test_bot.run(HOST, s.address[1])