import optparse import sys import time import subprocess import os import sleekxmpp import imp import logging #logging.basicConfig(level=logging.DEBUG, #format='%(levelname)-8s %(message)s') def registerXMPPAccount(user, password): responder = sleekxmpp.ClientXMPP(user, password) responder.register_plugin('xep_0030') # Service Discovery responder.register_plugin('xep_0077') responder['feature_mechanisms'].unencrypted_plain = True if responder.connect(("127.0.0.1", 5222)): responder.process(block=False) else: print "connect() failed" sys.exit(1) class BaseTest: def __init__(self, config, server_mode, room): self.config = config self.server_mode = server_mode self.room = room self.responder_jid = "responder@localhost" self.client_jid = "client@localhost" self.responder_password = "password" self.client_password = "password" self.client_room = room self.responder_room = room self.client_nick = "client" self.responder_nick = "responder" self.responder_roompassword = "" def skip_test(self, test): return False def start(self, Client, Responder): os.system("../../spectrum/src/spectrum2 -n ./" + self.config + " > spectrum2.log &") self.pre_test() time.sleep(1) responder = Responder(self.responder_jid, self.responder_password, self.responder_room, self.responder_roompassword, self.responder_nick) responder.register_plugin('xep_0030') # Service Discovery responder.register_plugin('xep_0045') # Multi-User Chat responder.register_plugin('xep_0199') # XMPP Ping responder.register_plugin('xep_0153') responder.register_plugin('xep_0054') responder['feature_mechanisms'].unencrypted_plain = True to = ("127.0.0.1", 5223) if self.responder_password != "password": to = () if responder.connect(to): responder.process(block=False) else: print "connect() failed" os.system("killall spectrum2") self.post_test() sys.exit(1) client = Client(self.client_jid, self.client_password, self.client_room, self.client_nick) client.register_plugin('xep_0030') # Service Discovery client.register_plugin('xep_0045') # Multi-User Chat client.register_plugin('xep_0199') # XMPP Ping client.register_plugin('xep_0153') client.register_plugin('xep_0054') client['feature_mechanisms'].unencrypted_plain = True time.sleep(1) to = ("127.0.0.1", 5223) if self.responder_password != "password": to = ("127.0.0.1", 5222) if client.connect(to): client.process(block=False) else: print "connect() failed" os.system("killall spectrum2") self.post_test() sys.exit(1) max_time = 60 while not client.finished and not responder.finished and max_time > 0: time.sleep(1) max_time -= 1 client.disconnect() responder.disconnect() os.system("killall spectrum2") self.post_test() ret = True tests = [] tests += client.tests.values() tests += responder.tests.values() for v in tests: if v[1]: print v[0] + ": PASSED" else: print v[0] + ": FAILED" ret = False if not ret: os.system("cat spectrum2.log") return ret class LibcommuniServerModeSingleServerConf(BaseTest): def __init__(self): BaseTest.__init__(self, "../libcommuni/irc_test.cfg", True, "#channel@localhost") self.directory = "../libcommuni/" def skip_test(self, test): if test in ["muc_join_nickname_used.py"]: return True return False def pre_test(self): os.system("ngircd -f ../libcommuni/ngircd.conf &") def post_test(self): os.system("killall ngircd 2>/dev/null") os.system("killall spectrum2_libcommuni_backend 2>/dev/null") class LibcommuniServerModeConf(BaseTest): def __init__(self): BaseTest.__init__(self, "../libcommuni/irc_test2.cfg", True, "#channel%localhost@localhost") self.directory = "../libcommuni/" def pre_test(self): os.system("ngircd -f ../libcommuni/ngircd.conf &") def post_test(self): os.system("killall ngircd 2>/dev/null") os.system("killall spectrum2_libcommuni_backend 2>/dev/null") class JabberServerModeConf(BaseTest): def __init__(self): BaseTest.__init__(self, "../libpurple_jabber/jabber_test.cfg", True, "room%conference.localhost@localhostxmpp") self.directory = "../libpurple_jabber/" self.client_jid = "client%localhost@localhostxmpp" self.responder_jid = "responder%localhost@localhostxmpp" def skip_test(self, test): if test in ["muc_whois.py", "muc_change_topic.py"]: return True return False def pre_test(self): os.system("prosody --config ../libpurple_jabber/prosody.cfg.lua >prosody.log &") time.sleep(3) os.system("../../spectrum_manager/src/spectrum2_manager -c ../libpurple_jabber/manager.conf localhostxmpp register client%localhost@localhostxmpp client@localhost password 2>/dev/null >/dev/null") os.system("../../spectrum_manager/src/spectrum2_manager -c ../libpurple_jabber/manager.conf localhostxmpp register responder%localhost@localhostxmpp responder@localhost password 2>/dev/null >/dev/null") def post_test(self): os.system("killall lua-5.1 2>/dev/null") os.system("killall spectrum2_libpurple_backend 2>/dev/null") class JabberSlackServerModeConf(BaseTest): def __init__(self): BaseTest.__init__(self, "../slack_jabber/jabber_slack_test.cfg", True, "room%conference.localhost@localhostxmpp") self.directory = "../slack_jabber/" self.client_jid = "client@localhost" self.client_room = "room@conference.localhost" self.responder_jid = "owner@spectrum2tests.xmpp.slack.com" self.responder_password = "spectrum2tests.e2zJwtKjLhLmt14VsMKq" self.responder_room = "spectrum2_room@conference.spectrum2tests.xmpp.slack.com" self.responder_nick = "owner" self.responder_roompassword = "spectrum2tests.e2zJwtKjLhLmt14VsMKq" def skip_test(self, test): os.system("cp ../slack_jabber/slack.sql .") if test.find("bad_password") != -1: print "Changing password to 'badpassword'" os.system("sqlite3 slack.sql \"UPDATE users SET password='badpassword' WHERE id=1\"") return False def pre_test(self): os.system("prosody --config ../slack_jabber/prosody.cfg.lua > prosody.log &") def post_test(self): os.system("killall lua-5.1 2>/dev/null") os.system("killall spectrum2_libpurple_backend 2>/dev/null") class TwitterServerModeConf(BaseTest): def __init__(self): BaseTest.__init__(self, "../twitter/twitter_test.cfg", True, "") self.directory = "../twitter/" self.client_password = "testpass123" def skip_test(self, test): os.system("cp ../twitter/twitter.sql .") def pre_test(self): pass def post_test(self): os.system("killall spectrum2_twitter_backend 2>/dev/null") configurations = [] configurations.append(LibcommuniServerModeSingleServerConf()) configurations.append(LibcommuniServerModeConf()) #configurations.append(JabberServerModeConf()) ##configurations.append(JabberSlackServerModeConf()) #configurations.append(TwitterServerModeConf()) exitcode = 0 for conf in configurations: for f in os.listdir(conf.directory): if not f.endswith(".py") or f == "start.py": continue if len(sys.argv) == 2 and sys.argv[1] != f: continue print conf.__class__.__name__ + ": Starting " + f + " test ..." test = imp.load_source('test', conf.directory + f) if conf.skip_test(f): print "Skipped." continue #try: ret = conf.start(test.Client, test.Responder) #except: #ret = False if not ret: exitcode = -1 sys.exit(exitcode)