Add new testing changes

This commit is contained in:
Shauna 2014-08-01 13:00:24 -04:00
parent bbde4ae19a
commit 3b9b498376
2 changed files with 151 additions and 68 deletions

105
newbot.py
View File

@ -1,13 +1,7 @@
# Welcome to WelcomeBot. Find source, documentation, etc here: https://github.com/shaunagm/WelcomeBot/ Licensed https://creativecommons.org/licenses/by-sa/2.0/
# Import some necessary libraries.
import socket
import sys
import time
import csv
import Queue
import random
import re
import socket, sys, time, csv, Queue, random, re, pdb
from threading import Thread
# Some basic variables used to configure the bot.
@ -49,14 +43,28 @@ class Bot(object):
# Defines a newcomer object
class NewComer(object):
def __init__(self, nick, bot):
self.nick = nick
self.born = time.time()
bot.newcomers.append(self)
def around_for(self):
return time.time() - self.born
def __init__(self, nick, bot):
self.nick = nick
self.born = time.time()
bot.newcomers.append(self)
def around_for(self):
return time.time() - self.born
#########################
### FAKE FUNCTION WE ARE REMOVING LATER ###
#########################
class fake_ircsock(object):
def send(self, msg):
self.sent_message = msg
def fake_irc_start():
global ircsock
ircsock = fake_ircsock()
#########################
@ -65,8 +73,7 @@ class NewComer(object):
# Creates a socket that will be used to send and receive messages,
# then connects the socket to an IRC server and joins the channel.
def irc_start():
global ircsock
def irc_start(): # pragma: no cover (this excludes this function from testing)
ircsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ircsock.connect((server, 6667)) # Here we connect to server using port 6667.
ircsock.send("USER {0} {0} {0} :This is http://openhatch.org/'s greeter bot"
@ -75,16 +82,16 @@ def irc_start():
ircsock.send("JOIN {} \n".format(channel)) # Joins channel
# Creates a separate thread for incoming messages (to combat concurrency issues)
def thread_start():
def thread_start(): # pragma: no cover (this excludes this function from testing)
global q
q = Queue.Queue() # Creates a Queue that will hold the incoming messages.
t = Thread(target=msg_handler) # Creates a separate thread running msg_hander function (below)
q = queue.Queue() # Creates a Queue that will hold the incoming messages.
t = Thread(target=msg_handler()) # Creates a separate thread running msg_hander function (below)
t.daemon = True
t.start()
# Reads the messages from the server and adds them to the Queue and prints
# them to the console. This function will be run in a thread, see below.
def msg_handler():
def msg_handler(): # pragma: no cover (this excludes this function from testing)
while True:
new_msg = ircsock.recv(2048) # receive data from the server
new_msg = new_msg.strip('\n\r') # removing any unnecessary linebreaks
@ -106,7 +113,7 @@ def get_regex(options):
### General Functions ###
#########################
# This welcomes the "person" passed to it.
# Welcomes the "person" passed to it.
def welcome_nick(newcomer):
ircsock.send("PRIVMSG {0} :Welcome {1}! The channel is pretty quiet "
"right now, so I though I'd say hello, and ping some people "
@ -114,10 +121,10 @@ def welcome_nick(newcomer):
"while, try emailing us at hello@openhatch.org or just try "
"coming back later. FYI, you're now on my list of known "
"nicknames, so I won't bother you again."
"\n".format(channel, newcomer, greeter_string("and")))
"\n".format(channel, newcomer, greeter_string("and", channel_greeters)))
# Checks and manages the status of newcomers.
def process_newcomers(bot, newcomerlist=[], welcome=1):
def process_newcomers(bot, newcomerlist, welcome=1):
for person in newcomerlist:
if welcome == 1:
welcome_nick(person.nick)
@ -125,17 +132,19 @@ def process_newcomers(bot, newcomerlist=[], welcome=1):
bot.newcomers.remove(person)
# Checks for messages.
def check_messages():
ircmsg = q.get() # get the next msg in the queue
actor = ircmsg.split(":")[1].split("!")[0] # and get the nick of the msg sender
return ircmsg, actor # Assumes that the above parsing works. :/
def parse_messages(ircmsg):
try:
actor = ircmsg.split(":")[1].split("!")[0] # and get the nick of the msg sender
return ircmsg, actor
except:
return None, None
# This function parses messages and responds to them appropriately.
# Parses messages and responds to them appropriately.
def message_response(bot, ircmsg, actor):
# if someone other than a newcomer speaks into the channel
if ircmsg.find("PRIVMSG " + channel) != -1 and actor not in [i.nick for i in bot.newcomers]:
process_newcomers(bot.newcomers,welcome=0) # Process/check newcomers without welcoming them
process_newcomers(bot,bot.newcomers,welcome=0) # Process/check newcomers without welcoming them
# if someone (other than the bot) joins the channel
if ircmsg.find("JOIN " + channel) != -1 and actor != botnick:
@ -168,11 +177,11 @@ def message_response(bot, ircmsg, actor):
### Bot Response Functions (called by message_response()) ###
#############################################################
# This function responds to a user that inputs "Hello Mybot".
# Responds to a user that inputs "Hello Mybot".
def bot_hello(greeting, actor):
ircsock.send("PRIVMSG {0} :{1} {2}\n".format(channel, greeting, actor))
# This function explains what the bot is when queried.
# Explains what the bot is when queried.
def bot_help():
ircsock.send("PRIVMSG {} :I'm a bot! I'm from here <https://github"
".com/shaunagm/oh-irc-bot>. You can change my behavior by "
@ -180,22 +189,20 @@ def bot_help():
".\n".format(channel))
# Returns a grammatically correct string of the channel_greeters.
def greeter_string(conjunction):
greeters = ""
if len(channel_greeters) > 2:
for name in channel_greeters[:-1]:
greeters += "{}, ".format(name)
greeters += "{0} {1}".format(conjunction, channel_greeters[-1])
elif len(channel_greeters) == 2:
greeters = "{0} {1} {2}".format(channel_greeters[0], conjunction,
channel_greeters[1])
def greeter_string(conjunction, greeters):
greeterstring = ""
if len(greeters) > 2:
for name in greeters[:-1]:
greeterstring += "{}, ".format(name)
greeterstring += "{0} {1}".format(conjunction, greeters[-1])
elif len(greeters) == 2:
greeterstring = "{0} {1} {2}".format(greeters[0], conjunction,
greeters[1])
else:
greeters = channel_greeters[0]
return greeters
greeterstring = greeters[0]
return greeterstring
# This function is used to change the wait time from the channel.
# It confirms that the attempt is allowed and then returns the requested value.
# If the attempt is not allowed, a message is sent to help
# Changes the wait time from the channel.
def wait_time_change(actor, ircmsg):
for admin in channel_greeters:
if actor == admin:
@ -206,7 +213,7 @@ def wait_time_change(actor, ircmsg):
return int(finder.group())
ircsock.send("PRIVMSG {0} :{1} you are not authorized to make that "
"change. Please contact one of the channel greeters, like {2}, for "
"assistance.\n".format(channel, actor, greeter_string("or")))
"assistance.\n".format(channel, actor, greeter_string("or", channel_greeters)))
# Responds to server Pings.
def pong():
@ -218,13 +225,15 @@ def pong():
##########################
def main():
irc_start()
irc_start()
thread_start()
WelcomeBot = Bot()
while 1: # Loop forever
process_newcomers(WelcomeBot, [i for i in WelcomeBot.newcomers if i.around_for() > WelcomeBot.wait_time])
if q.empty() == 0: # If the queue is not empty...
message_response(WelcomeBot, *check_messages()) # Checks for messages, gets a message text and actor, and responds
ircmsg, actor = parse_messages(q.get()) # parse the next msg in the queue
if ircmsg is not None: # If we were able to parse it
message_response(WelcomeBot, ircmsg, actor) # Respond to the parsed message
if __name__ == "__main__": # This line tells the interpreter to only execute main() if the program is being run, not imported.
sys.exit(main())

View File

@ -4,6 +4,7 @@ import csv
import unittest
import newbot
import time
import pdb
class TestBotClass(unittest.TestCase):
@ -73,10 +74,6 @@ class TestNewComerClass(unittest.TestCase):
time.sleep(0.01)
self.assertAlmostEqual(newComer.around_for(), .01, places=2)
#
# Not sure how to test irc_start, thread_start or msg_handler yet.
#
class TestProcessNewcomers(unittest.TestCase):
def setUp(self):
@ -93,29 +90,106 @@ class TestProcessNewcomers(unittest.TestCase):
def test_check_new_known_nicks(self):
newbot.process_newcomers(self.bot, [i for i in self.bot.newcomers if i.around_for() > self.bot.wait_time], welcome=0)
self.assertEqual(self.bot.known_nicks,[['Alice'],['Bob'],['Harry'],['Hermione']])
def test_welcome_nick(self):
newbot.fake_irc_start()
newbot.process_newcomers(self.bot, [i for i in self.bot.newcomers if i.around_for() > self.bot.wait_time], welcome=1)
self.assertEqual(newbot.ircsock.sent_message, "PRIVMSG #openhatch-bots :Welcome Hermione! The channel is pretty quiet right now, so I though I'd say hello, and ping some people (like shauna) that you're here. If no one responds for a while, try emailing us at hello@openhatch.org or just try coming back later. FYI, you're now on my list of known nicknames, so I won't bother you again.\n")
def tearDown(self):
with open('test_nicks.csv', 'w') as csv_file:
csv_file.write('Alice\nBob\n')
## Should be a test of the welcome=1/welcome=0 functionality here, but not sure how to do that yet since welcome() calls ircsock.
class TestParseMessages(unittest.TestCase):
def test_good_string(self):
ircmsg, actor = newbot.parse_messages(":vader!darth@darkside.org PRIVMSG #deathstar : I find your lack of faith disturbing")
self.assertEqual([ircmsg, actor], [':vader!darth@darkside.org PRIVMSG #deathstar : I find your lack of faith disturbing', 'vader'])
def test_bad_string(self):
ircmsg, actor = newbot.parse_messages("we should probably replace this with a bad string more likely to occur")
self.assertEqual([ircmsg, actor], [None, None])
class TestMessageResponse(unittest.TestCase):
def setUp(self):
self.bot = newbot.Bot('test_nicks.csv')
newbot.NewComer('Chappe', self.bot)
def test_newcomer_speaking(self):
newbot.message_response(self.bot,"~q@r.m.us PRIVMSG #openhatch-bots :hah","Chappe") # Standard message by newcomer
nicklist = [i.nick for i in self.bot.newcomers] # Makes a list of newcomers nicks for easy asserting
self.assertEqual(nicklist, ['Chappe'])
def test_oldtimer_speaking(self):
newbot.message_response(self.bot,"~q@r.m.us PRIVMSG #openhatch-bots :hah","Alice") # Standard message by oldtimer
nicklist = [i.nick for i in self.bot.newcomers] # Makes a list of newcomers nicks for easy asserting
self.assertEqual(nicklist, [])
def test_join(self):
newbot.message_response(self.bot,"JOIN #openhatch-bots right now!","Shauna") # Replace with actual ping message ALSO argh the channel variable might mess things up if folks change it, which they very well might :/ (Also true for tests below.) I think maybe use the format.() style that is used for wait_change etc
self.assertEqual(self.bot.newcomers[1].nick,'Shauna')
def test_part(self):
newbot.message_response(self.bot,"JOIN #openhatch-bots right now!","Shauna") # Replace with actual ping message ALSO argh the channel variable might mess things up if folks change it, which they very well might :/ (Also true for tests below.)
self.assertEqual(len(self.bot.newcomers), 2)
newbot.message_response(self.bot,"PART #openhatch-bots","Shauna") # Replace with actual ping message ALSO argh the channel variable might mess things up :/
self.assertEqual(len(self.bot.newcomers), 1)
def test_hello(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"PRIVMSG sup WelcomeBot2","Shauna") # The botnick may also be changed. :(
self.assertTrue(hasattr(newbot.ircsock, 'sent_message')) # Fails because sent_message is never actually created
self.assertIn(newbot.ircsock.sent_message, ["PRIVMSG #openhatch-bots :hello Shauna\n", "PRIVMSG #openhatch-bots :hi Shauna\n", "PRIVMSG #openhatch-bots :hey Shauna\n", "PRIVMSG #openhatch-bots :yo Shauna\n", "PRIVMSG #openhatch-bots :sup Shauna\n"])
def test_help(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"PRIVMSG info WelcomeBot2","Shauna") # The botnick may also be changed. :(
self.assertTrue(hasattr(newbot.ircsock, 'sent_message')) # Fails because sent_message is never actually created
self.assertEqual(newbot.ircsock.sent_message, "PRIVMSG #openhatch-bots :I'm a bot! I'm from here <https://github.com/shaunagm/oh-irc-bot>. You can change my behavior by submitting a pull request or by talking to shauna.\n")
def test_wait_time_from_admin(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"WelcomeBot2 --wait-time 40","shauna") # Channel-greeters may also be changed. :(
self.assertEqual(newbot.ircsock.sent_message, "PRIVMSG #openhatch-bots :shauna the wait time is changing to 40 seconds.\n")
def test_wait_time_from_non_admin(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"WelcomeBot2 --wait-time 40","Impostor") # Channel-greeters may also be changed. :(
self.assertEqual(newbot.ircsock.sent_message, "PRIVMSG #openhatch-bots :Impostor you are not authorized to make that change. Please contact one of the channel greeters, like shauna, for assistance.\n")
def test_pong(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"PING :","Shauna") # Replace this with actual ping message
self.assertEqual(newbot.ircsock.sent_message,"PONG :pingis\n")
def test_bad_pong(self):
newbot.fake_irc_start()
newbot.message_response(self.bot,"PING!!! :","Shauna") # Replace this with actual ping message
self.assertFalse(hasattr(newbot.ircsock, 'sent_message')) # Fails because sent_message is never actually created
def tearDown(self):
with open('test_nicks.csv', 'w') as csv_file:
csv_file.write('Alice\nBob\n')
#
# Not sure how to test check_messages.
#
class TestMessageResponse(unittest.TestCase):
class TestGreeterString(unittest.TestCase):
def setUp(self):
pass
self.bot = newbot.Bot('test_nicks.csv')
def test_one_greeter(self):
greeterstring = newbot.greeter_string("and", ['shauna'])
self.assertEqual(greeterstring, "shauna")
def test_two_greeters(self):
greeters = newbot.greeter_string("and", ['shauna','sauna'])
self.assertEqual(greeters, "shauna and sauna")
def tearDown(self):
pass
def test_three_greeters(self):
greeters = newbot.greeter_string("and", ['shauna','sauna','megafauna'])
self.assertEqual(greeters, "shauna, sauna, and megafauna")
class TestProcessNewcomers(unittest.TestCase):
def setUp(self):
pass
def tearDown(self):
pass
# Runs all the unit-tests
if __name__ == '__main__':
unittest.main()