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
|
import sys
import time
import threading
from sleekxmpp.test import *
class TestStreamPubsub(SleekTest):
"""
Test using the XEP-0030 plugin.
"""
def setUp(self):
self.stream_start()
def tearDown(self):
self.stream_close()
def testCreateInstantNode(self):
"""Test creating an instant node"""
t = threading.Thread(name='create_node',
target=self.xmpp['xep_0060'].create_node,
args=('pubsub.example.com', None))
t.start()
self.send("""
<iq type="set" id="1" to="pubsub.example.com">
<pubsub xmlns="http://jabber.org/protocol/pubsub">
<create />
</pubsub>
</iq>
""")
self.recv("""
<iq type="result" id="1"
to="tester@localhost" from="pubsub.example.com">
<pubsub xmlns="http://jabber.org/protocol/pubsub">
<create node="25e3d37dabbab9541f7523321421edc5bfeb2dae" />
</pubsub>
</iq>
""")
t.join()
def testCreateNodeNoConfig(self):
"""Test creating a node without a config"""
t = threading.Thread(name='create_node',
target=self.xmpp['xep_0060'].create_node,
args=('pubsub.example.com', 'princely_musings'))
t.start()
self.send("""
<iq type="set" id="1" to="pubsub.example.com">
<pubsub xmlns="http://jabber.org/protocol/pubsub">
<create node="princely_musings" />
</pubsub>
</iq>
""")
self.recv("""
<iq type="result" id="1"
to="tester@localhost" from="pubsub.example.com" />
""")
t.join()
def testCreateNodeConfig(self):
"""Test creating a node with a config"""
form = self.xmpp['xep_0004'].stanza.Form()
form['type'] = 'submit'
form.add_field(var='pubsub#access_model', value='whitelist')
t = threading.Thread(name='create_node',
target=self.xmpp['xep_0060'].create_node,
args=('pubsub.example.com', 'princely_musings'),
kwargs={'config': form})
t.start()
self.send("""
<iq type="set" id="1" to="pubsub.example.com">
<pubsub xmlns="http://jabber.org/protocol/pubsub">
<create node="princely_musings" />
<configure>
<x xmlns="jabber:x:data" type="submit">
<field var="pubsub#access_model">
<value>whitelist</value>
</field>
<field var="FORM_TYPE">
<value>http://jabber.org/protocol/pubsub#node_config</value>
</field>
</x>
</configure>
</pubsub>
</iq>
""")
self.recv("""
<iq type="result" id="1"
to="tester@localhost" from="pubsub.example.com" />
""")
t.join()
def testDeleteNode(self):
"""Test deleting a node"""
t = threading.Thread(name='delete_node',
target=self.xmpp['xep_0060'].delete_node,
args=('pubsub.example.com', 'some_node'))
t.start()
self.send("""
<iq type="set" to="pubsub.example.com" id="1">
<pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
<delete node="some_node" />
</pubsub>
</iq>
""")
self.recv("""
<iq type="result" id="1"
to="tester@localhost" from="pubsub.example.com" />
""")
t.join()
suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamPubsub)
|
