summaryrefslogtreecommitdiff
path: root/conn_tests/test_pubsubserver.py
blob: 1750636ff810b27933754e3d73dfb7527d9841c2 (plain)
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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import logging
import sleekxmpp
from optparse import OptionParser
from xml.etree import cElementTree as ET
import os
import time
import sys
import thread
import unittest
import sleekxmpp.plugins.xep_0004
try:
	import configparser
except ImportError:
	import ConfigParser as configparser
try:
	import queue
except ImportError:
	import Queue as queue

class TestClient(sleekxmpp.ClientXMPP):
	def __init__(self, jid, password):
		sleekxmpp.ClientXMPP.__init__(self, jid, password)
		self.add_event_handler("session_start", self.start)
		#self.add_event_handler("message", self.message)
		self.waitforstart = queue.Queue()
	
	def start(self, event):
		self.getRoster()
		self.sendPresence()
		self.waitforstart.put(True)


class TestPubsubServer(unittest.TestCase):
	statev = {}

	def __init__(self, *args, **kwargs):
		unittest.TestCase.__init__(self, *args, **kwargs)

	def setUp(self):
		pass

	def test001getdefaultconfig(self):
		"""Get the default node config"""
		result = self.xmpp1['xep_0060'].getNodeConfig(self.pshost)
		self.statev['defaultconfig'] = result
		self.failUnless(isinstance(result, sleekxmpp.plugins.xep_0004.Form))
	
	def test002createdefaultnode(self):
		"""Create a node without config"""
		self.failUnless(self.xmpp1['xep_0060'].create_node(self.pshost, 'testnode1'))

	def test003deletenode(self):
		"""Delete recently created node"""
		self.failUnless(self.xmpp1['xep_0060'].deleteNode(self.pshost, 'testnode1'))
	
	def test004createnode(self):
		"""Create a node with a config"""
		self.statev['defaultconfig'].field['pubsub#access_model'].setValue('open')
		self.statev['defaultconfig'].field['pubsub#notify_retract'].setValue(True)
		self.failUnless(self.xmpp1['xep_0060'].create_node(self.pshost, 'testnode2', self.statev['defaultconfig']))
	
	def test005reconfigure(self):
		"""Retrieving node config and reconfiguring"""
		nconfig = self.xmpp1['xep_0060'].getNodeConfig(self.pshost, 'testnode2')
		self.failUnless(nconfig, "No configuration returned")
		#print("%s == %s" % (nconfig.getValues(), self.statev['defaultconfig'].getValues()))
		self.failUnless(nconfig.getValues() == self.statev['defaultconfig'].getValues(), "Configuration does not match")
		self.failUnless(self.xmpp1['xep_0060'].setNodeConfig(self.pshost, 'testnode2', nconfig))

	def test006subscribetonode(self):
		"""Subscribe to node from account 2"""
		self.failUnless(self.xmpp2['xep_0060'].subscribe(self.pshost, "testnode2"))
	
	def test007publishitem(self):
		"""Publishing item"""
		item = ET.Element('{http://netflint.net/protocol/test}test')
		result = self.xmpp1['xep_0060'].setItem(self.pshost, "testnode2", (('test_node1', item),))
		self.failUnless(result)
		#need to add check for update
	
	def test999cleanup(self):
		"Cleaning up"
		self.failUnless(self.xmpp1['xep_0060'].deleteNode(self.pshost, 'testnode2'))


if __name__ == '__main__':
	#parse command line arguements
	optp = OptionParser()
	optp.add_option('-q','--quiet', help='set logging to ERROR', action='store_const', dest='loglevel', const=logging.ERROR, default=logging.INFO)
	optp.add_option('-d','--debug', help='set logging to DEBUG', action='store_const', dest='loglevel', const=logging.DEBUG, default=logging.INFO)
	optp.add_option('-v','--verbose', help='set logging to COMM', action='store_const', dest='loglevel', const=5, default=logging.INFO)
	optp.add_option("-c","--config", dest="configfile", default="config.xml", help="set config file to use")
	optp.add_option("-n","--nodenum", dest="nodenum", default="1", help="set node number to use")
	optp.add_option("-p","--pubsub", dest="pubsub", default="1", help="set pubsub host to use")
	opts,args = optp.parse_args()
	
	logging.basicConfig(level=opts.loglevel, format='%(levelname)-8s %(message)s')

	#load xml config
	logging.info("Loading config file: %s" % opts.configfile)
	config = configparser.RawConfigParser()
	config.read(opts.configfile)
	
	#init
	logging.info("Account 1 is %s" % config.get('account1', 'jid'))
	xmpp1 = TestClient(config.get('account1','jid'), config.get('account1','pass'))
	logging.info("Account 2 is %s" % config.get('account2', 'jid'))
	xmpp2 = TestClient(config.get('account2','jid'), config.get('account2','pass'))
	
	xmpp1.registerPlugin('xep_0004')
	xmpp1.registerPlugin('xep_0030')
	xmpp1.registerPlugin('xep_0060')
	xmpp1.registerPlugin('xep_0199')
	xmpp2.registerPlugin('xep_0004')
	xmpp2.registerPlugin('xep_0030')
	xmpp2.registerPlugin('xep_0060')
	xmpp2.registerPlugin('xep_0199')

	if not config.get('account1', 'server'):
		# we don't know the server, but the lib can probably figure it out
		xmpp1.connect() 
	else:
		xmpp1.connect((config.get('account1', 'server'), 5222))
	xmpp1.process(threaded=True)
	
	#init
	if not config.get('account2', 'server'):
		# we don't know the server, but the lib can probably figure it out
		xmpp2.connect() 
	else:
		xmpp2.connect((config.get('account2', 'server'), 5222))
	xmpp2.process(threaded=True)

	TestPubsubServer.xmpp1 = xmpp1
	TestPubsubServer.xmpp2 = xmpp2
	TestPubsubServer.pshost = 'pubsub.recon'
	xmpp1.waitforstart.get(True)
	xmpp2.waitforstart.get(True)
	testsuite = unittest.TestLoader().loadTestsFromTestCase(TestPubsubServer)

	alltests_suite = unittest.TestSuite([testsuite])
	result = unittest.TextTestRunner(verbosity=2).run(alltests_suite)
	xmpp1.disconnect()
	xmpp2.disconnect()