Files
@ ab4740c7b632
Branch filter:
Location: libtransport.git/spectrum/src/tests/start.py
ab4740c7b632
2.8 KiB
text/x-python
Communi: Support /whois (or .whois) command
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | import optparse
import sys
import time
import subprocess
import os
import sleekxmpp
import imp
class BaseTest:
def __init__(self, config, server_mode, room):
self.config = config
self.server_mode = server_mode
self.room = room
def start(self, Client, Responder):
self.pre_test()
os.system("../spectrum2 -n ./" + self.config + " > spectrum2.log &")
time.sleep(1)
responder = Responder("responder@localhost", "password", self.room, "responder")
responder.register_plugin('xep_0030') # Service Discovery
responder.register_plugin('xep_0045') # Multi-User Chat
responder.register_plugin('xep_0199') # XMPP Ping
responder['feature_mechanisms'].unencrypted_plain = True
if responder.connect():
responder.process(block=False)
else:
print "connect() failed"
sys.exit(1)
client = Client("client@localhost", "password", self.room, "client")
client.register_plugin('xep_0030') # Service Discovery
client.register_plugin('xep_0045') # Multi-User Chat
client.register_plugin('xep_0199') # XMPP Ping
client['feature_mechanisms'].unencrypted_plain = True
time.sleep(2)
if client.connect():
client.process(block=False)
else:
print "connect() failed"
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, "irc_test.cfg", True, "#channel@localhost")
def pre_test(self):
os.system("ngircd -f 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, "irc_test2.cfg", True, "#channel%localhost@localhost")
def pre_test(self):
os.system("ngircd -f ngircd.conf &")
def post_test(self):
os.system("killall ngircd 2>/dev/null")
os.system("killall spectrum2_libcommuni_backend 2>/dev/null")
configurations = []
configurations.append(LibcommuniServerModeSingleServerConf())
configurations.append(LibcommuniServerModeConf())
exitcode = 0
for conf in configurations:
for f in os.listdir("."):
if not f.endswith(".py") or f == "start.py":
continue
print conf.__class__.__name__ + ": Starting " + f + " test ..."
test = imp.load_source('test', './' + f)
try:
ret = conf.start(test.Client, test.Responder)
except:
ret = False
if not ret:
exitcode = -1
sys.exit(exitcode)
|