from collections import defaultdict from log import log import main import time class IrcUser(): def __init__(self, nick, ts, uid, ident='null', host='null', realname='PyLink dummy client', realhost='null', ip='0.0.0.0', modes=set()): self.nick = nick self.ts = ts self.uid = uid self.ident = ident self.host = host self.realhost = realhost self.ip = ip self.realname = realname self.modes = modes self.identified = False self.channels = set() def __repr__(self): return repr(self.__dict__) class IrcServer(): """PyLink IRC Server class. uplink: The SID of this IrcServer instance's uplink. This is set to None for the main PyLink PseudoServer! name: The name of the server. internal: Whether the server is an internal PyLink PseudoServer. """ def __init__(self, uplink, name, internal=False): self.uplink = uplink self.users = [] self.internal = internal self.name = name.lower() self.has_bursted = False def __repr__(self): return repr(self.__dict__) class IrcChannel(): def __init__(self): self.users = set() self.modes = set() self.topic = '' self.ts = int(time.time()) self.prefixmodes = {'ops': set(), 'halfops': set(), 'voices': set(), 'owners': set(), 'admins': set()} def __repr__(self): return repr(self.__dict__) def removeuser(self, target): for s in self.prefixmodes.values(): s.discard(target) self.users.discard(target) class ProtocolError(Exception): pass global testconf testconf = {'bot': { 'nick': 'PyLink', 'user': 'pylink', 'realname': 'PyLink Service Client', 'loglevel': 'DEBUG', }, 'servers': {'unittest': { 'ip': '0.0.0.0', 'port': 7000, 'recvpass': "abcd", 'sendpass': "abcd", 'protocol': "null", 'hostname': "pylink.unittest", 'sid': "9PY", 'channels': ["#pylink"], }, }, } class FakeIRC(main.Irc): def connect(self): self.messages = [] self.hookargs = [] self.hookmsgs = [] self.socket = None def run(self, data): """Queues a message to the fake IRC server.""" log.debug('<- ' + data) self.proto.handle_events(self, data) def send(self, data): self.messages.append(data) log.debug('-> ' + data) def takeMsgs(self): """Returns a list of messages sent by the protocol module since the last takeMsgs() call, so we can track what has been sent.""" msgs = self.messages self.messages = [] return msgs def takeCommands(self, msgs): """Returns a list of commands parsed from the output of takeMsgs().""" sidprefix = ':' + self.sid commands = [] for m in msgs: args = m.split() if m.startswith(sidprefix): commands.append(args[1]) else: commands.append(args[0]) return commands def takeHooks(self): """Returns a list of hook arguments sent by the protocol module since the last takeHooks() call.""" hookmsgs = self.hookmsgs self.hookmsgs = [] return hookmsgs @staticmethod def dummyhook(irc, source, command, parsed_args): """Dummy function to bind to hooks.""" irc.hookmsgs.append(parsed_args) class FakeProto(): """Dummy protocol module for testing purposes.""" @staticmethod def handle_events(irc, data): pass @staticmethod def connect(irc): pass