← Back to branch summary

~vishvananda/nova/network-refactor

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_reflector.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_reflector.py
 
1
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for twisted.enterprise reflectors.
 
6
"""
 
7
 
 
8
import random
 
9
 
 
10
from twisted.internet import reactor, interfaces, defer
 
11
from twisted.enterprise.row import RowObject
 
12
from twisted.enterprise.reflector import EQUAL
 
13
from twisted.enterprise.sqlreflector import SQLReflector
 
14
from twisted.enterprise import util
 
15
from twisted.test.test_adbapi import makeSQLTests
 
16
from twisted.trial.util import suppress as suppressWarning
 
17
from twisted.trial import unittest
 
18
 
 
19
 
 
20
tableName = "testTable"
 
21
childTableName = "childTable"
 
22
 
 
23
 
 
24
class TestRow(RowObject):
 
25
    rowColumns = [("key_string",      "varchar"),
 
26
                  ("col2",            "int"),
 
27
                  ("another_column",  "varchar"),
 
28
                  ("Column4",         "varchar"),
 
29
                  ("column_5_",       "int")]
 
30
    rowKeyColumns = [("key_string", "varchar")]
 
31
    rowTableName  = tableName
 
32
 
 
33
 
 
34
class ChildRow(RowObject):
 
35
    rowColumns    = [("childId",  "int"),
 
36
                     ("foo",      "varchar"),
 
37
                     ("test_key", "varchar"),
 
38
                     ("stuff",    "varchar"),
 
39
                     ("gogogo",   "int"),
 
40
                     ("data",     "varchar")]
 
41
    rowKeyColumns = [("childId", "int")]
 
42
    rowTableName  = childTableName
 
43
    rowForeignKeys = [(tableName,
 
44
                       [("test_key","varchar")],
 
45
                       [("key_string","varchar")],
 
46
                       None, 1)]
 
47
 
 
48
 
 
49
main_table_schema = """
 
50
CREATE TABLE testTable (
 
51
  key_string     varchar(64),
 
52
  col2           integer,
 
53
  another_column varchar(64),
 
54
  Column4        varchar(64),
 
55
  column_5_      integer
 
56
)
 
57
"""
 
58
 
 
59
child_table_schema = """
 
60
CREATE TABLE childTable (
 
61
  childId        integer,
 
62
  foo            varchar(64),
 
63
  test_key       varchar(64),
 
64
  stuff          varchar(64),
 
65
  gogogo         integer,
 
66
  data           varchar(64)
 
67
)
 
68
"""
 
69
 
 
70
 
 
71
def randomizeRow(row, nulls_ok=True, trailing_spaces_ok=True):
 
72
    values = {}
 
73
    for name, type in row.rowColumns:
 
74
        if util.getKeyColumn(row, name):
 
75
            values[name] = getattr(row, name)
 
76
            continue
 
77
        elif nulls_ok and random.randint(0, 9) == 0:
 
78
            value = None # null
 
79
        elif type == 'int':
 
80
            value = random.randint(-10000, 10000)
 
81
        else:
 
82
            if random.randint(0, 9) == 0:
 
83
                value = ''
 
84
            else:
 
85
                value = ''.join(map(lambda i:chr(random.randrange(32,127)),
 
86
                                    xrange(random.randint(1, 64))))
 
87
            if not trailing_spaces_ok:
 
88
                value = value.rstrip()
 
89
        setattr(row, name, value)
 
90
        values[name] = value
 
91
    return values
 
92
 
 
93
 
 
94
def rowMatches(row, values):
 
95
    for name, type in row.rowColumns:
 
96
        if getattr(row, name) != values[name]:
 
97
            print ("Mismatch on column %s: |%s| (row) |%s| (values)" %
 
98
                   (name, getattr(row, name), values[name]))
 
99
            return False
 
100
    return True
 
101
 
 
102
 
 
103
rowObjectSuppression = suppressWarning(
 
104
    message="twisted.enterprise.row is deprecated since Twisted 8.0",
 
105
    category=DeprecationWarning)
 
106
 
 
107
 
 
108
reflectorSuppression = suppressWarning(
 
109
    message="twisted.enterprise.reflector is deprecated since Twisted 8.0",
 
110
    category=DeprecationWarning)
 
111
 
 
112
 
 
113
class ReflectorTestBase:
 
114
    """
 
115
    Base class for testing reflectors.
 
116
 
 
117
    @ivar reflector: The reflector created during setup.
 
118
    """
 
119
 
 
120
    if interfaces.IReactorThreads(reactor, None) is None:
 
121
        skip = "No thread support, no reflector tests"
 
122
 
 
123
    count = 100 # a parameter used for running iterative tests
 
124
 
 
125
    def randomizeRow(self, row):
 
126
        return randomizeRow(row, self.nulls_ok, self.trailing_spaces_ok)
 
127
 
 
128
    def extraSetUp(self):
 
129
        """
 
130
        Create and store a reference to a SQL reflector for use by the tests.
 
131
        """
 
132
        d = self.createReflector()
 
133
        d.addCallback(self._cbSetUp)
 
134
        return d
 
135
 
 
136
    def _cbSetUp(self, reflector):
 
137
        self.reflector = reflector
 
138
 
 
139
    def tearDown(self):
 
140
        return self.destroyReflector()
 
141
 
 
142
    def destroyReflector(self):
 
143
        pass
 
144
 
 
145
    def test_reflector(self):
 
146
        """
 
147
        Full featured tests of reflector.
 
148
        """
 
149
        # create one row to work with
 
150
        row = TestRow()
 
151
        row.assignKeyAttr("key_string", "first")
 
152
        values = self.randomizeRow(row)
 
153
 
 
154
        # save it
 
155
        d = self.reflector.insertRow(row)
 
156
 
 
157
        def _loadBack(_):
 
158
            # now load it back in
 
159
            whereClause = [("key_string", EQUAL, "first")]
 
160
            d = self.reflector.loadObjectsFrom(tableName,
 
161
                                               whereClause=whereClause)
 
162
            return d.addCallback(self.gotData)
 
163
 
 
164
        def _getParent(_):
 
165
            # make sure it came back as what we saved
 
166
            self.failUnless(len(self.data) == 1, "no row")
 
167
            parent = self.data[0]
 
168
            self.failUnless(rowMatches(parent, values), "no match")
 
169
            return parent
 
170
 
 
171
        d.addCallback(_loadBack)
 
172
        d.addCallback(_getParent)
 
173
        d.addCallback(self._cbTestReflector)
 
174
        return d
 
175
    test_reflector.suppress = [rowObjectSuppression, reflectorSuppression]
 
176
 
 
177
    def _cbTestReflector(self, parent):
 
178
        # create some child rows
 
179
        test_values = {}
 
180
        inserts = []
 
181
        child_values = {}
 
182
        for i in range(0, self.num_iterations):
 
183
            row = ChildRow()
 
184
            row.assignKeyAttr("childId", i)
 
185
            values = self.randomizeRow(row)
 
186
            values['test_key'] = row.test_key = "first"
 
187
            child_values[i] = values
 
188
            inserts.append(self.reflector.insertRow(row))
 
189
            row = None
 
190
        #del inserts
 
191
        d = defer.gatherResults(inserts)
 
192
        values = [None]
 
193
 
 
194
        def _loadObjects(_):
 
195
            d = self.reflector.loadObjectsFrom(childTableName, parentRow=parent)
 
196
            return d.addCallback(self.gotData)
 
197
 
 
198
        def _checkLoadObjects(_):
 
199
            self.failUnless(len(self.data) == self.num_iterations,
 
200
                            "no rows on query")
 
201
            self.failUnless(len(parent.childRows) == self.num_iterations,
 
202
                            "did not load child rows: %d" % len(parent.childRows))
 
203
            for child in parent.childRows:
 
204
                self.failUnless(rowMatches(child, child_values[child.childId]),
 
205
                                "child %d does not match" % child.childId)
 
206
 
 
207
        def _checkLoadObjects2(_):
 
208
            self.failUnless(len(self.data) == self.num_iterations,
 
209
                            "no rows on query")
 
210
            self.failUnless(len(parent.childRows) == self.num_iterations,
 
211
                            "child rows added twice!: %d" % len(parent.childRows))
 
212
 
 
213
        def _changeParent(_):
 
214
            # now change the parent
 
215
            values[0] = self.randomizeRow(parent)
 
216
            return self.reflector.updateRow(parent)
 
217
 
 
218
        def _loadBack(_):
 
219
            # now load it back in
 
220
            whereClause = [("key_string", EQUAL, "first")]
 
221
            d = self.reflector.loadObjectsFrom(tableName, whereClause=whereClause)
 
222
            return d.addCallback(self.gotData)
 
223
 
 
224
        def _checkLoadBack(_):
 
225
            # make sure it came back as what we saved
 
226
            self.failUnless(len(self.data) == 1, "no row")
 
227
            parent = self.data[0]
 
228
            self.failUnless(rowMatches(parent, values[0]), "no match")
 
229
            # save parent
 
230
            test_values[parent.key_string] = values[0]
 
231
            parent = None
 
232
 
 
233
        def _saveMoreTestRows(_):
 
234
            # save some more test rows
 
235
            ds = []
 
236
            for i in range(0, self.num_iterations):
 
237
                row = TestRow()
 
238
                row.assignKeyAttr("key_string", "bulk%d"%i)
 
239
                test_values[row.key_string] = self.randomizeRow(row)
 
240
                ds.append(self.reflector.insertRow(row))
 
241
            return defer.gatherResults(ds)
 
242
 
 
243
        def _loadRowsBack(_):
 
244
            # now load them all back in
 
245
            d = self.reflector.loadObjectsFrom("testTable")
 
246
            return d.addCallback(self.gotData)
 
247
 
 
248
        def _checkRowsBack(_):
 
249
            # make sure they are the same
 
250
            self.failUnless(len(self.data) == self.num_iterations + 1,
 
251
                            "query did not get rows")
 
252
            for row in self.data:
 
253
                self.failUnless(rowMatches(row, test_values[row.key_string]),
 
254
                                "child %s does not match" % row.key_string)
 
255
 
 
256
        def _changeRows(_):
 
257
            # now change them all
 
258
            ds = []
 
259
            for row in self.data:
 
260
                test_values[row.key_string] = self.randomizeRow(row)
 
261
                ds.append(self.reflector.updateRow(row))
 
262
            d = defer.gatherResults(ds)
 
263
            return d.addCallback(_cbChangeRows)
 
264
 
 
265
        def _cbChangeRows(_):
 
266
            self.data = None
 
267
 
 
268
        def _deleteRows(_):
 
269
            # now delete them
 
270
            ds = []
 
271
            for row in self.data:
 
272
                ds.append(self.reflector.deleteRow(row))
 
273
            d = defer.gatherResults(ds)
 
274
            return d.addCallback(_cbChangeRows)
 
275
 
 
276
        def _checkRowsDeleted(_):
 
277
            self.failUnless(len(self.data) == 0, "rows were not deleted")
 
278
 
 
279
        d.addCallback(_loadObjects)
 
280
        d.addCallback(_checkLoadObjects)
 
281
        d.addCallback(_loadObjects)
 
282
        d.addCallback(_checkLoadObjects2)
 
283
        d.addCallback(_changeParent)
 
284
        d.addCallback(_loadBack)
 
285
        d.addCallback(_checkLoadBack)
 
286
        d.addCallback(_saveMoreTestRows)
 
287
        d.addCallback(_loadRowsBack)
 
288
        d.addCallback(_checkRowsBack)
 
289
        d.addCallback(_changeRows)
 
290
        d.addCallback(_loadRowsBack)
 
291
        d.addCallback(_checkRowsBack)
 
292
        d.addCallback(_deleteRows)
 
293
        d.addCallback(_loadRowsBack)
 
294
        d.addCallback(_checkRowsDeleted)
 
295
        return d
 
296
 
 
297
 
 
298
    def test_saveAndDelete(self):
 
299
        """
 
300
        Create a row and then try to delete it.
 
301
        """
 
302
        # create one row to work with
 
303
        row = TestRow()
 
304
        row.assignKeyAttr("key_string", "first")
 
305
        values = self.randomizeRow(row)
 
306
        # save it
 
307
        d = self.reflector.insertRow(row)
 
308
        def _deleteRow(_):
 
309
            # delete it
 
310
            return self.reflector.deleteRow(row)
 
311
        d.addCallback(_deleteRow)
 
312
        return d
 
313
    test_saveAndDelete.suppress = [rowObjectSuppression, reflectorSuppression]
 
314
 
 
315
 
 
316
    def gotData(self, data):
 
317
        self.data = data
 
318
 
 
319
 
 
320
ReflectorTestBase.timeout = 30.0
 
321
 
 
322
 
 
323
class SQLReflectorTestBase(ReflectorTestBase):
 
324
    """
 
325
    Base class for the SQL reflector.
 
326
    """
 
327
 
 
328
    def createReflector(self):
 
329
        self.startDB()
 
330
        self.dbpool = self.makePool()
 
331
        self.dbpool.start()
 
332
 
 
333
        if self.can_clear:
 
334
            d = self.dbpool.runOperation('DROP TABLE testTable')
 
335
            d.addCallback(lambda _:
 
336
                          self.dbpool.runOperation('DROP TABLE childTable'))
 
337
            d.addErrback(lambda _: None)
 
338
        else:
 
339
            d = defer.succeed(None)
 
340
 
 
341
        d.addCallback(lambda _: self.dbpool.runOperation(main_table_schema))
 
342
        d.addCallback(lambda _: self.dbpool.runOperation(child_table_schema))
 
343
        reflectorClass = self.escape_slashes and SQLReflector \
 
344
                         or NoSlashSQLReflector
 
345
        d.addCallback(lambda _:
 
346
                      reflectorClass(self.dbpool, [TestRow, ChildRow]))
 
347
        return d
 
348
 
 
349
    def destroyReflector(self):
 
350
        d = self.dbpool.runOperation('DROP TABLE testTable')
 
351
        d.addCallback(lambda _:
 
352
                      self.dbpool.runOperation('DROP TABLE childTable'))
 
353
        def close(_):
 
354
            self.dbpool.close()
 
355
            self.stopDB()
 
356
        d.addCallback(close)
 
357
        return d
 
358
 
 
359
 
 
360
class DeprecationTestCase(unittest.TestCase):
 
361
    """
 
362
    Test various deprecations of twisted.enterprise.
 
363
    """
 
364
 
 
365
    def test_rowDeprecation(self):
 
366
        """
 
367
        Test deprecation of L{RowObject}.
 
368
        """
 
369
        def wrapper():
 
370
            return TestRow()
 
371
        self.assertWarns(DeprecationWarning,
 
372
            "twisted.enterprise.row is deprecated since Twisted 8.0",
 
373
            __file__,
 
374
            wrapper)
 
375
 
 
376
    def test_reflectorDeprecation(self):
 
377
        """
 
378
        Test deprecation of L{SQLReflector}.
 
379
        """
 
380
        def wrapper():
 
381
            return SQLReflector(None, ())
 
382
        from twisted.enterprise import sqlreflector
 
383
        self.assertWarns(DeprecationWarning,
 
384
            "twisted.enterprise.reflector is deprecated since Twisted 8.0",
 
385
            sqlreflector.__file__,
 
386
            wrapper)
 
387
 
 
388
 
 
389
# GadflyReflectorTestCase SQLiteReflectorTestCase PyPgSQLReflectorTestCase
 
390
# PsycopgReflectorTestCase MySQLReflectorTestCase FirebirdReflectorTestCase
 
391
makeSQLTests(SQLReflectorTestBase, 'ReflectorTestCase', globals())
 
392
 
 
393
 
 
394
class NoSlashSQLReflector(SQLReflector):
 
395
    """
 
396
    An sql reflector that only escapes single quotes.
 
397
    """
 
398
 
 
399
    def escape_string(self, text):
 
400
        return text.replace("'", "''")
 
401