2015-12-19 00:11:57 +00:00
|
|
|
import socket
|
|
|
|
import unittest
|
|
|
|
|
2015-12-19 07:43:45 +00:00
|
|
|
from .irc_utils import message_parser
|
|
|
|
|
2015-12-19 00:11:57 +00:00
|
|
|
class _IrcTestCase(unittest.TestCase):
|
|
|
|
controllerClass = None # Will be set by __main__.py
|
|
|
|
|
2015-12-19 08:30:50 +00:00
|
|
|
def getLine(self):
|
|
|
|
raise NotImplementedError()
|
|
|
|
def getMessage(self, filter_pred=None):
|
|
|
|
"""Gets a message and returns it. If a filter predicate is given,
|
|
|
|
fetches messages until the predicate returns a False on a message,
|
|
|
|
and returns this message."""
|
|
|
|
while True:
|
|
|
|
msg = message_parser.parse_message(self.getLine())
|
|
|
|
if not filter_pred or filter_pred(msg):
|
|
|
|
return msg
|
|
|
|
|
2015-12-19 08:03:08 +00:00
|
|
|
class BaseClientTestCase(_IrcTestCase):
|
|
|
|
"""Basic class for client tests. Handles spawning a client and getting
|
|
|
|
messages from it."""
|
2015-12-19 00:11:57 +00:00
|
|
|
def setUp(self):
|
|
|
|
self.controller = self.controllerClass()
|
|
|
|
self._setUpServer()
|
|
|
|
def tearDown(self):
|
|
|
|
del self.controller
|
|
|
|
self.conn_file.close()
|
|
|
|
self.conn.close()
|
|
|
|
self.server.close()
|
|
|
|
|
|
|
|
def _setUpServer(self):
|
|
|
|
"""Creates the server and make it listen."""
|
|
|
|
self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
|
|
self.server.bind(('', 0)) # Bind any free port
|
|
|
|
self.server.listen(1)
|
|
|
|
def acceptClient(self):
|
|
|
|
"""Make the server accept a client connection. Blocking."""
|
|
|
|
(self.conn, addr) = self.server.accept()
|
2015-12-19 08:30:50 +00:00
|
|
|
self.conn_file = self.conn.makefile(newline='\r\n',
|
|
|
|
encoding='utf8')
|
2015-12-19 00:11:57 +00:00
|
|
|
|
|
|
|
def getLine(self):
|
2015-12-19 08:30:50 +00:00
|
|
|
line = self.conn_file.readline()
|
|
|
|
if self.show_io:
|
|
|
|
print('C: {}'.format(line.strip()))
|
|
|
|
return line
|
|
|
|
def sendLine(self, line):
|
|
|
|
assert self.conn.sendall(line.encode()) is None
|
|
|
|
if not line.endswith('\r\n'):
|
|
|
|
assert self.conn.sendall(b'\r\n') is None
|
|
|
|
print('S: {}'.format(line.strip()))
|
2015-12-19 08:03:08 +00:00
|
|
|
|
2015-12-19 08:30:50 +00:00
|
|
|
class ClientNegociationHelper:
|
2015-12-19 08:03:08 +00:00
|
|
|
"""Helper class for tests handling capabilities negociation."""
|
|
|
|
def readCapLs(self):
|
|
|
|
(hostname, port) = self.server.getsockname()
|
|
|
|
self.controller.run(
|
|
|
|
hostname=hostname,
|
|
|
|
port=port,
|
|
|
|
authentication=None,
|
|
|
|
)
|
|
|
|
self.acceptClient()
|
|
|
|
m = self.getMessage()
|
|
|
|
self.assertEqual(m.command, 'CAP',
|
|
|
|
'First message is not CAP LS.')
|
2015-12-19 08:30:50 +00:00
|
|
|
if m.params == ['LS']:
|
2015-12-19 08:03:08 +00:00
|
|
|
self.protocol_version = 301
|
2015-12-19 08:30:50 +00:00
|
|
|
elif m.params == ['LS', '302']:
|
2015-12-19 08:03:08 +00:00
|
|
|
self.protocol_version = 302
|
|
|
|
else:
|
2015-12-19 08:30:50 +00:00
|
|
|
raise AssertionError('Unknown CAP params: {}'
|
2015-12-19 08:03:08 +00:00
|
|
|
.format(m.params))
|
2015-12-19 08:30:50 +00:00
|
|
|
|
|
|
|
def userNickPredicate(self, msg):
|
|
|
|
"""Predicate to be used with getMessage to handle NICK/USER
|
|
|
|
transparently."""
|
|
|
|
if msg.command == 'NICK':
|
|
|
|
self.assertEqual(len(msg.params), 1, msg)
|
|
|
|
self.nick = msg.params[0]
|
|
|
|
return False
|
|
|
|
elif msg.command == 'USER':
|
|
|
|
self.assertEqual(len(msg.params), 4, msg)
|
|
|
|
self.nick = msg.params
|
|
|
|
return False
|
|
|
|
else:
|
|
|
|
return True
|
|
|
|
|