3
Tests for L{epsilon.structlike}.
8
from twisted.trial import unittest
9
from twisted.internet.threads import deferToThread
10
from twisted.internet.defer import gatherResults
11
from epsilon.structlike import record
14
class MyRecord(record('something somethingElse')):
16
A sample record subclass.
21
class StructLike(unittest.TestCase):
22
def _testme(self, TestStruct):
24
self.assertEquals(x.x, 1)
25
self.assertEquals(x.y, 2)
26
self.assertEquals(x.z, 3)
28
y = TestStruct('3', '2', '1')
29
self.assertEquals(y.x, '3')
30
self.assertEquals(y.y, '2')
31
self.assertEquals(y.z, '1')
33
z = TestStruct(z='z', x='x', y='y')
34
self.assertEquals(z.x, 'x')
35
self.assertEquals(z.y, 'y')
36
self.assertEquals(z.z, 'z')
39
self.assertEquals(a.x, 'abc')
40
self.assertEquals(a.y, 2)
41
self.assertEquals(a.z, 3)
43
b = TestStruct(y='123')
44
self.assertEquals(b.x, 1)
45
self.assertEquals(b.y, '123')
46
self.assertEquals(b.z, 3)
48
def testWithPositional(self):
49
self._testme(record('x y z', x=1, y=2, z=3))
51
def testWithPositionalSubclass(self):
52
class RecordSubclass(record('x y z', x=1, y=2, z=3)):
54
self._testme(RecordSubclass)
56
def testWithoutPositional(self):
57
self._testme(record(x=1, y=2, z=3))
59
def testWithoutPositionalSubclass(self):
60
class RecordSubclass(record(x=1, y=2, z=3)):
62
self._testme(RecordSubclass)
64
def testBreakRecord(self):
65
self.assertRaises(TypeError, record)
66
self.assertRaises(TypeError, record, 'a b c', a=1, c=2)
67
self.assertRaises(TypeError, record, 'a b', c=2)
68
self.assertRaises(TypeError, record, 'a b', a=1)
70
def testUndeclared(self):
74
self.assertEquals(r.foo, 2)
76
def testCreateWithNoValuesAndNoDefaults(self):
78
self.assertRaises(TypeError, R)
80
def testUnknownArgs(self):
82
Test that passing in unknown keyword and / or positional arguments to a
83
record's initializer causes TypeError to be raised.
86
self.assertRaises(TypeError, R, x=5, y=6)
87
self.assertRaises(TypeError, R, 5, 6)
90
def test_typeStringRepresentation(self):
92
'Record' types should have a name which provides information about the
95
R = record('xyz abc def')
96
self.assertEquals(R.__name__, "Record<xyz abc def>")
99
def test_instanceStringRepresentation(self):
101
'Record' instances should provide a string representation which
102
provides information about the values contained in their slots.
104
obj = MyRecord(something=1, somethingElse=2)
105
self.assertEquals(repr(obj), 'MyRecord(something=1, somethingElse=2)')
108
def test_instanceStringRepresentationNesting(self):
110
Nested L{Record} instances should have nested string representations.
112
obj = MyRecord(something=1, somethingElse=2)
113
objRepr = 'MyRecord(something=1, somethingElse=2)'
115
repr(MyRecord(obj, obj)),
116
'MyRecord(something=%s, somethingElse=%s)' % (objRepr, objRepr))
119
def test_instanceStringRepresentationRecursion(self):
121
'Record' instances should provide a repr that displays 'ClassName(...)'
122
when it would otherwise infinitely recurse.
124
obj = MyRecord(something=1, somethingElse=2)
125
obj.somethingElse = obj
127
repr(obj), 'MyRecord(something=1, somethingElse=MyRecord(...))')
130
def test_instanceStringRepresentationUnhashableRecursion(self):
132
'Record' instances should display 'ClassName(...)' even for unhashable
135
obj = MyRecord(something=1, somethingElse=[])
136
obj.somethingElse.append(obj)
138
repr(obj), 'MyRecord(something=1, somethingElse=[MyRecord(...)])')
141
def test_threadLocality(self):
143
An 'Record' repr()'d in two separate threads at the same time should
144
look the same (i.e. the repr state tracking for '...' should be
147
class StickyRepr(object):
149
This has a __repr__ which will block until a separate thread
150
notifies it that it should return. We use this to create a race
155
self.set = threading.Event()
156
self.wait = threading.Event()
163
mr = MyRecord(something=1, somethingElse=r)
164
d = deferToThread(repr, mr)
166
# First we wait for the first thread doing a repr() to enter its
169
# OK, now it's blocked. Let's make sure that subsequent calls to
170
# this repr() won't block.
172
# Do it! This is a concurrent repr().
174
# Now we're done, wake up the other repr and let it complete.
177
d2 = deferToThread(otherRepr)
179
def done((thread1repr, thread2repr)):
180
knownGood = 'MyRecord(something=1, somethingElse=sticky)'
181
# self.assertEquals(thread1repr, thread2repr)
182
self.assertEquals(thread1repr, knownGood)
183
self.assertEquals(thread2repr, knownGood)
184
return gatherResults([d, d2]).addCallback(done)
added
removed
Epsilon/epsilon/test/test_structlike.py
