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
|
# This file is part of Moksha.
# Copyright (C) 2008-2010 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
""" Utilities to make writing tests easier """
import functools
import socket
old_school = False
try:
# For python-2.6
import unittest2 as unittest
old_school = True
except ImportError:
import unittest
from moksha.common.lib.helpers import get_moksha_appconfig
def crosstest(method):
@functools.wraps(method)
def wrapper(self):
for setup, name in self._setUp():
def _inner(name):
if not old_school:
setup()
else:
try:
setup()
except unittest.SkipTest:
# For python-2.6
# nose doesn't know how to catch this, so we just
# return. The test "passes" which is incorrect. It
# should be skipped.
return
try:
return method(self)
finally:
self._tearDown()
yield _inner, name
return wrapper
config_sets = {
'stomp': {
"moksha.livesocket": "True",
"moksha.livesocket.backend": "stomp",
"orbited_host": "localhost",
"orbited_port": "9000",
"orbited_scheme": "http",
"stomp_broker": "localhost",
"stomp_port": "61613",
"stomp_user": "guest",
"stomp_pass": "guest",
},
'amqp': {
"moksha.livesocket": "True",
"moksha.livesocket.backend": "amqp",
"orbited_host": "localhost",
"orbited_port": "9000",
"orbited_scheme": "http",
"amqp_broker": "guest/guest@localhost",
"amqp_broker_host": "localhost",
"amqp_broker_port": "5672",
"amqp_broker_user": "guest",
"amqp_broker_pass": "guest",
"amqp_broker_ssl": "False",
},
'zeromq': {
"moksha.livesocket": "True",
"moksha.livesocket.backend": "websocket",
"moksha.livesocket.websocket.port": "9998",
"zmq_enabled": "True",
"zmq_publish_endpoints": "tcp://*:6543",
"zmq_subscribe_endpoints": "tcp://127.0.0.1:6543",
"zmq_strict": "True",
},
}
flash_keys = []
for name, config_set in config_sets.items():
flash_keys.extend(config_set.keys())
flash_keys = list(set(flash_keys))
def should_skip_config_set(name, config_set):
if name == 'stomp':
address = (config_set['stomp_broker'],
config_set['stomp_port'])
# If we can connect, then run tests. If we can't, then don't.
try:
s = socket.create_connection(address)
s.close()
return False
except:
return True
elif name == 'amqp':
address = (config_set['amqp_broker_host'],
config_set['amqp_broker_port'])
# If we can't import qpid, then bail... but also:
# If we can connect, then run tests. If we can't, then don't.
try:
import qpid
s = socket.create_connection(address)
s.close()
return False
except:
return True
elif name == 'zeromq':
return False
else:
raise ValueError("Unknown config set name.")
def make_setup_functions(kernel):
for name, config_set in config_sets.items():
def __setup():
if should_skip_config_set(name, config_set):
raise unittest.SkipTest("%s is not available." % name)
config = get_moksha_appconfig()
for key in flash_keys:
if key in config:
del config[key]
for key, value in config_set.items():
config[key] = value
kernel(config)
yield __setup, name
|
