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
145
146
147
148
149
150
151
|
import os
import sys
import unittest
import uuid
sys.path.insert(0, os.pardir)
sys.path.append(os.getcwd())
try:
import stompy
except ImportError:
stompy = None
Frame = StompMessage = StompBackend = object
else:
from carrot.backends.pystomp import Message as StompMessage
from carrot.backends.pystomp import Backend as StompBackend
from stompy.frame import Frame
from carrot.connection import BrokerConnection
from carrot.messaging import Publisher, Consumer
from tests.utils import test_stomp_connection_args, STOMP_QUEUE
from carrot.serialization import encode
_no_stompy_msg = "* stompy (python-stomp) not installed. " \
+ "Will not execute related tests."
_no_stompy_msg_emitted = False
def stompy_or_None():
def emit_no_stompy_msg():
global _no_stompy_msg_emitted
if not _no_stompy_msg_emitted:
sys.stderr.write("\n" + _no_stompy_msg + "\n")
_no_stompy_msg_emitted = True
if stompy is None:
emit_no_stompy_msg()
return None
return stompy
def create_connection():
return BrokerConnection(backend_cls=StompBackend,
**test_stomp_connection_args())
def create_backend():
return create_connection().create_backend()
class MockFrame(Frame):
def mock(self, command=None, headers=None, body=None):
self.command = command
self.headers = headers
self.body = body
return self
class TestStompMessage(unittest.TestCase):
def test_message(self):
if not stompy_or_None():
return
b = create_backend()
self.assertTrue(b)
message_body = "George Constanza"
delivery_tag = str(uuid.uuid4())
frame = MockFrame().mock(body=message_body, headers={
"message-id": delivery_tag,
"content_type": "text/plain",
"content_encoding": "utf-8",
})
m1 = StompMessage(backend=b, frame=frame)
m2 = StompMessage(backend=b, frame=frame)
m3 = StompMessage(backend=b, frame=frame)
self.assertEquals(m1.body, message_body)
self.assertEquals(m1.delivery_tag, delivery_tag)
#m1.ack()
self.assertRaises(NotImplementedError, m2.reject)
self.assertRaises(NotImplementedError, m3.requeue)
class TestPyStompMessaging(unittest.TestCase):
def setUp(self):
if stompy_or_None():
self.conn = create_connection()
self.queue = STOMP_QUEUE
self.exchange = STOMP_QUEUE
self.routing_key = STOMP_QUEUE
def create_consumer(self, **options):
return Consumer(connection=self.conn,
queue=self.queue, exchange=self.exchange,
routing_key=self.routing_key, **options)
def create_publisher(self, **options):
return Publisher(connection=self.conn,
exchange=self.exchange,
routing_key=self.routing_key, **options)
def test_backend(self):
if not stompy_or_None():
return
publisher = self.create_publisher()
consumer = self.create_consumer()
for i in range(100):
publisher.send({"foo%d" % i: "bar%d" % i})
publisher.close()
discarded = consumer.discard_all()
self.assertEquals(discarded, 100)
publisher.close()
consumer.close()
publisher = self.create_publisher()
for i in range(100):
publisher.send({"foo%d" % i: "bar%d" % i})
consumer = self.create_consumer()
for i in range(100):
while True:
message = consumer.fetch()
if message:
break
self.assertTrue("foo%d" % i in message.payload)
message.ack()
publisher.close()
consumer.close()
consumer = self.create_consumer()
discarded = consumer.discard_all()
self.assertEquals(discarded, 0)
def create_raw_message(self, publisher, body, delivery_tag):
content_type, content_encoding, payload = encode(body)
frame = MockFrame().mock(body=payload, headers={
"message-id": delivery_tag,
"content-type": content_type,
"content-encoding": content_encoding,
})
return frame
|
