- cmake -DCMAKE_BUILD_TYPE=Debug -DENABLE_TESTS=ON . && make && ./src/libtransport_test && make spectrum2 && cd ./spectrum/src/tests && python -V && python start.py

import optparse

import sys

import time

import subprocess

import os

import sleekxmpp

class Responder(sleekxmpp.ClientXMPP):

	def __init__(self, jid, password, room, nick):

		sleekxmpp.ClientXMPP.__init__(self, jid, password)

		self.room = room

		self.nick = nick

		self.add_event_handler("session_start", self.start)

		self.add_event_handler("groupchat_message", self.muc_message)

		self.tests = {}

	def muc_message(self, msg):

		if msg['mucnick'] != self.nick:

			self.send_message(mto=msg['from'].bare,

							mbody="echo %s" % msg['body'],

							mtype='groupchat')

	def start(self, event):

		self.plugin['xep_0045'].joinMUC(self.room, self.nick, wait=True)

class Client(sleekxmpp.ClientXMPP):

	def __init__(self, jid, password, room, nick):

		sleekxmpp.ClientXMPP.__init__(self, jid, password)

		self.room = room

		self.nick = nick

		self.add_event_handler("session_start", self.start)

		self.add_event_handler("groupchat_message", self.muc_message)

		self.finished = False

		self.tests = {}

		self.tests["echo_received"] = ["libcommuni: Send and receive messages", False]

	def muc_message(self, msg):

		if msg['mucnick'] != self.nick:

			if msg['body'] == "echo abc":

				self.tests["echo_received"][1] = True

				self.finished = True

	def start(self, event):

		self.getRoster()

		self.sendPresence()

		self.plugin['xep_0045'].joinMUC(self.room, self.nick, wait=True)

		self.send_message(mto=self.room, mbody="abc", mtype='groupchat')

import optparse

import sys

import time

import subprocess

import os

import sleekxmpp

import imp

def single_test(Client, Responder):

	os.system("../spectrum2 -n ./irc_test.cfg > spectrum2.log &")

	os.system("ngircd -f ngircd.conf &")

	time.sleep(1)

	responder = Responder("responder@localhost", "password", "#channel@localhost", "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", "#channel@localhost", "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 max_time > 0:

		time.sleep(1)

		max_time -= 1

	client.disconnect()

	responder.disconnect()

	os.system("killall spectrum2")

	os.system("killall ngircd")

	os.system("killall spectrum2_libcommuni_backend 2>/dev/null")

	ret = True

	for v in client.tests.values():

		if v[1]:

			print v[0] + ": PASSED"

		else:

			print v[0] + ": FAILED"

			ret = False

	if not ret:

		os.system("cat spectrum2.log")

	return ret

exitcode = 0

for f in os.listdir("."):

	if not f.endswith(".py") or f == "start.py":

		continue

	print "Starting " + f + " test ..."

	test = imp.load_source('test', './' + f)

	ret = single_test(test.Client, test.Responder)

	if not ret:

		exitcode = -1

sys.exit(exitcode)