~ibid-core/ibid/old-trunk-1.6 : revision 893

1

2

# Released under terms of the MIT/X/Expat Licence. See COPYING for details.

3

from inspect import stack

4

5

from twisted.trial import unittest

6

7

from ibid import event

8

9

class TestEvent(unittest.TestCase):

10

def _ev(self, source='fakesource', type='testmessage'):

11

return event.Event(source, type)

12

13

def assertByteStringWarning(self, count=1):

14

"Check that addresponse raised an error about using bytestrings"

15

caller = getattr(self, stack()[1][3])

16

warnings = self.flushWarnings(offendingFunctions=[caller])

17

self.assertEqual(len(warnings), count)

18

for i in range(count):

19

self.assertTrue('byte string' in warnings[i]['message'],

20

'Byte-String response should provoke a warning')

21

22

def assertListSubstitutionWarning(self, count=1):

23

"Check that addresponse raised an error about substituting lists"

24

caller = getattr(self, stack()[1][3])

25

warnings = self.flushWarnings(offendingFunctions=[caller])

26

self.assertEqual(len(warnings), count)

27

for i in range(count):

28

self.assertTrue('single item or dict' in warnings[i]['message'],

29

'Byte-String response should provoke a warning')

30

31

def test_empty_event(self):

32

"Events contain some default data."

33

ev = self._ev()

34

self.assertEqual('fakesource', ev.source)

35

self.assertEqual('testmessage', ev.type)

36

self.assertEqual([], ev.responses)

37

self.assertEqual({}, ev.sender)

38

self.assertEqual(False, ev.processed)

39

40

def test_attr(self):

41

"Attibutes and indexed keys are equivalent."

42

ev = self._ev()

43

self.assertRaises(AttributeError, lambda: ev.foo)

44

self.assertRaises(KeyError, lambda: ev['foo'])

45

ev.foo = 'bar'

46

self.assertEqual('bar', ev.foo)

47

self.assertEqual('bar', ev['foo'])

48

self.assertRaises(AttributeError, lambda: ev.bar)

49

self.assertRaises(KeyError, lambda: ev['bar'])

50

ev['bar'] = 'foo'

51

self.assertEqual('foo', ev.bar)

52

self.assertEqual('foo', ev['bar'])

53

54

def test_None_response(self):

55

"None is an invalid response."

56

ev = self._ev()

57

self.assertEqual([], ev.responses)

58

self.assertRaises(Exception, lambda: ev.addresponse(None))

59

self.assertEqual([], ev.responses)

60

61

def test_str_response(self):

62

"String responses become appropriate structures."

63

ev = self._ev()

64

self.assertEqual([], ev.responses)

65

ev.addresponse('foo')

66

self.assertEqual([{'reply': 'foo',

67

'target': None,

68

'source': 'fakesource',

69

'address': True,

70

'conflate': True}], ev.responses)

71

self.assertEqual(True, ev.processed)

72

self.assertByteStringWarning()

73

74

def test_str_response_twice(self):

75

"Two responses are separate."

76

ev = self._ev()

77

self.assertEqual([], ev.responses)

78

ev.addresponse('foo')

79

ev.addresponse('bar')

80

self.assertEqual([{'reply': 'foo',

81

'target': None,

82

'source': 'fakesource',

83

'address': True,

84

'conflate': True},

85

{'reply': 'bar',

86

'target': None,

87

'source': 'fakesource',

88

'address': True,

89

'conflate': True}], ev.responses)

90

self.assertEqual(True, ev.processed)

91

self.assertByteStringWarning(2)

92

93

def test_str_response_unprocessed(self):

94

"Responses don't have to mark the event as processed."

95

ev = self._ev()

96

self.assertEqual([], ev.responses)

97

ev.addresponse('foo', processed=False)

98

self.assertEqual([{'reply': 'foo',

99

'target': None,

100

'source': 'fakesource',

101

'address': True,

102

'conflate': True}], ev.responses)

103

self.assertEqual(False, ev.processed)

104

self.assertByteStringWarning(1)

105

106

def test_str_response_processed_unprocessed(self):

107

"processed=False doesn't clear the processed flag."

108

ev = self._ev()

109

self.assertEqual([], ev.responses)

110

ev.processed = True

111

ev.addresponse('foo', processed=False)

112

self.assertEqual([{'reply': 'foo',

113

'target': None,

114

'source': 'fakesource',

115

'address': True,

116

'conflate': True}], ev.responses)

117

self.assertEqual(True, ev.processed)

118

self.assertByteStringWarning(1)

119

120

def test_str_response_with_channel(self):

121

"Events from a channel send their responses back there."

122

ev = self._ev()

123

ev.channel = '#chan'

124

self.assertEqual([], ev.responses)

125

ev.addresponse('foo')

126

self.assertEqual([{'reply': 'foo',

127

'target': '#chan',

128

'source': 'fakesource',

129

'address': True,

130

'conflate': True}], ev.responses)

131

self.assertByteStringWarning(1)

132

133

def test_unicode_response(self):

134

"Unicode responses behave the same as bytestrings."

135

ev = self._ev()

136

self.assertEqual([], ev.responses)

137

ev.addresponse(u'foo')

138

self.assertEqual([{'reply': 'foo',

139

'target': None,

140

'source': 'fakesource',

141

'address': True,

142

'conflate': True}], ev.responses)

143

144

def test_unicode_params_response(self):

145

"Responses can contain parameters for dict-string interpolation."

146

ev = self._ev()

147

self.assertEqual([], ev.responses)

148

ev.addresponse(u'foo %(name)s', {'name': 'bar'})

149

self.assertEqual([{'reply': 'foo bar',

150

'target': None,

151

'source': 'fakesource',

152

'address': True,

153

'conflate': True}], ev.responses)

154

155

def test_unicode_tuple_params_response(self):

156

"Responses can contain parameters for tuple-string interpolation."

157

ev = self._ev()

158

self.assertEqual([], ev.responses)

159

ev.addresponse(u'foo %s', ('bar',))

160

self.assertEqual([{'reply': 'foo bar',

161

'target': None,

162

'source': 'fakesource',

163

'address': True,

164

'conflate': True}], ev.responses)

165

self.assertListSubstitutionWarning()

166

167

def test_simple_dict_response(self):

168

"Dicts are valid response values."

169

ev = self._ev()

170

self.assertEqual([], ev.responses)

171

ev.addresponse({'reply': 'foo'})

172

self.assertEqual([{'reply': 'foo',

173

'target': None,

174

'source': 'fakesource',

175

'address': True,

176

'conflate': True}], ev.responses)

177

178

def test_complex_dict_response(self):

179

"Dict responses can override event defaults."

180

ev = self._ev()

181

self.assertEqual([], ev.responses)

182

ev.addresponse({'reply': 'foo',

183

'target': 'mytarget',

184

'source': 'mysource',

185

'address': False,

186

'conflate': False,

187

'mykey': 'myvalue'})

188

self.assertEqual([{'reply': 'foo',

189

'target': 'mytarget',

190

'source': 'mysource',

191

'address': False,

192

'conflate': False,

193

'mykey': 'myvalue'}], ev.responses)

194

195

def test_str_kwargs_response(self):

196

"Keyword arguments to addresponse override event defaults."

197

ev = self._ev()

198

self.assertEqual([], ev.responses)

199

ev.addresponse('foo', bar='baz')

200

self.assertEqual([{'reply': 'foo',

201

'target': None,

202

'source': 'fakesource',

203

'address': True,

204

'conflate': True,

205

'bar': 'baz'}], ev.responses)

206

self.assertByteStringWarning(1)

207

208

def test_complex_dict_with_kwargs_response(self):

209

"Keyword arguments to addresponse override response dict values."

210

ev = self._ev()

211

self.assertEqual([], ev.responses)

212

ev.addresponse({'reply': 'r1',

213

'target': 't1',

214

'source': 's1',

215

'address': 'a1',

216

'conflate': 'c1'})

217

ev.addresponse({'reply': 'r1',

218

'target': 't1',

219

'source': 's1',

220

'address': 'a1',

221

'conflate': 'c1'},

222

reply='r2',

223

target='t2',

224

source='s2',

225

address='a2',

226

conflate='c2')

227

self.assertEqual([{'reply': 'r1',

228

'target': 't1',

229

'source': 's1',

230

'address': 'a1',

231

'conflate': 'c1'},

232

{'reply': 'r2',

233

'target': 't2',

234

'source': 's2',

235

'address': 'a2',

236

'conflate': 'c2'}], ev.responses)

237

238

# vi: set et sta sw=4 ts=4: