1177
1176
raise NotImplementedError
1180
class TransportTestProviderAdapter(object):
1181
"""A tool to generate a suite testing all transports for a single test.
1183
This is done by copying the test once for each transport and injecting
1184
the transport_class and transport_server classes into each copy. Each copy
1185
is also given a new id() to make it easy to identify.
1188
def adapt(self, test):
1189
result = unittest.TestSuite()
1190
for klass, server_factory in self._test_permutations():
1191
new_test = deepcopy(test)
1192
new_test.transport_class = klass
1193
new_test.transport_server = server_factory
1194
def make_new_test_id():
1195
new_id = "%s(%s)" % (new_test.id(), server_factory.__name__)
1196
return lambda: new_id
1197
new_test.id = make_new_test_id()
1198
result.addTest(new_test)
1201
def get_transport_test_permutations(self, module):
1202
"""Get the permutations module wants to have tested."""
1203
if getattr(module, 'get_test_permutations', None) is None:
1204
raise AssertionError("transport module %s doesn't provide get_test_permutations()"
1206
##warning("transport module %s doesn't provide get_test_permutations()"
1207
## % module.__name__)
1209
return module.get_test_permutations()
1211
def _test_permutations(self):
1212
"""Return a list of the klass, server_factory pairs to test."""
1214
for module in _get_transport_modules():
1216
result.extend(self.get_transport_test_permutations(reduce(getattr,
1217
(module).split('.')[1:],
1218
__import__(module))))
1219
except errors.DependencyNotPresent, e:
1220
# Continue even if a dependency prevents us
1221
# from running this test
1226
1179
class TransportLogger(object):
1227
1180
"""Adapt a transport to get clear logging data on api calls.