← Back to branch summary

~launchpad-committers/storm/lp

« back to all changes in this revision

Viewing changes to storm/tests/databases/base.py

  • Committer: Colin Watson
  • Date: 2020-03-18 17:45:51 UTC
  • mfrom: (386.34.109 storm)
  • Revision ID: cjwatson@canonical.com-20200318174551-z51ycxqkxn4m70ui
Merge Storm 0.23 release.

Show diffs side-by-side

added added

removed removed

Lines of Context:
storm/tests/databases/base.py
32
32
 
33
33
from storm.uri import URI
34
34
from storm.expr import Select, Column, SQLToken, SQLRaw, Count, Alias
35
 
from storm.variables import (Variable, PickleVariable, RawStrVariable,
 
35
from storm.variables import (Variable, PickleVariable, BytesVariable,
36
36
                             DecimalVariable, DateTimeVariable, DateVariable,
37
37
                             TimeVariable, TimeDeltaVariable)
38
38
from storm.database import *
151
151
        result = self.connection.execute(u"SELECT title FROM test")
152
152
        self.assertTrue(isinstance(result, Result))
153
153
        row = result.get_one()
154
 
        self.assertEquals(row, ("Title 10",))
 
154
        self.assertEqual(row, ("Title 10",))
155
155
        self.assertTrue(isinstance(row[0], six.text_type))
156
156
 
157
157
    def test_execute_params(self):
176
176
 
177
177
    def test_get_one(self):
178
178
        result = self.connection.execute("SELECT * FROM test ORDER BY id")
179
 
        self.assertEquals(result.get_one(), (10, "Title 10"))
 
179
        self.assertEqual(result.get_one(), (10, "Title 10"))
180
180
 
181
181
    def test_get_all(self):
182
182
        result = self.connection.execute("SELECT * FROM test ORDER BY id")
183
 
        self.assertEquals(result.get_all(),
184
 
                          [(10, "Title 10"), (20, "Title 20")])
 
183
        self.assertEqual(result.get_all(),
 
184
                         [(10, "Title 10"), (20, "Title 20")])
185
185
 
186
186
    def test_iter(self):
187
187
        result = self.connection.execute("SELECT * FROM test ORDER BY id")
188
 
        self.assertEquals([item for item in result],
189
 
                          [(10, "Title 10"), (20, "Title 20")])
 
188
        self.assertEqual([item for item in result],
 
189
                         [(10, "Title 10"), (20, "Title 20")])
190
190
 
191
191
    def test_simultaneous_iter(self):
192
192
        result1 = self.connection.execute("SELECT * FROM test "
195
195
                                          "ORDER BY id DESC")
196
196
        iter1 = iter(result1)
197
197
        iter2 = iter(result2)
198
 
        self.assertEquals(next(iter1), (10, "Title 10"))
199
 
        self.assertEquals(next(iter2), (20, "Title 20"))
200
 
        self.assertEquals(next(iter1), (20, "Title 20"))
201
 
        self.assertEquals(next(iter2), (10, "Title 10"))
 
198
        self.assertEqual(next(iter1), (10, "Title 10"))
 
199
        self.assertEqual(next(iter2), (20, "Title 20"))
 
200
        self.assertEqual(next(iter1), (20, "Title 20"))
 
201
        self.assertEqual(next(iter2), (10, "Title 10"))
202
202
        self.assertRaises(StopIteration, next, iter1)
203
203
        self.assertRaises(StopIteration, next, iter2)
204
204
 
210
210
        expr = result.get_insert_identity(primary_key, primary_variables)
211
211
        select = Select(Column("title", SQLToken("test")), expr)
212
212
        result = self.connection.execute(select)
213
 
        self.assertEquals(result.get_one(), ("Title 30",))
 
213
        self.assertEqual(result.get_one(), ("Title 30",))
214
214
 
215
215
    def test_get_insert_identity_composed(self):
216
216
        result = self.connection.execute("INSERT INTO test (title) "
221
221
        expr = result.get_insert_identity(primary_key, primary_variables)
222
222
        select = Select(Column("title", SQLToken("test")), expr)
223
223
        result = self.connection.execute(select)
224
 
        self.assertEquals(result.get_one(), ("Title 30",))
 
224
        self.assertEqual(result.get_one(), ("Title 30",))
225
225
 
226
226
    def test_datetime(self):
227
227
        value = datetime(1977, 4, 5, 12, 34, 56, 78)
232
232
        result.set_variable(variable, result.get_one()[0])
233
233
        if not self.supports_microseconds:
234
234
            value = value.replace(microsecond=0)
235
 
        self.assertEquals(variable.get(), value)
 
235
        self.assertEqual(variable.get(), value)
236
236
 
237
237
    def test_date(self):
238
238
        value = date(1977, 4, 5)
241
241
        result = self.connection.execute("SELECT d FROM datetime_test")
242
242
        variable = DateVariable()
243
243
        result.set_variable(variable, result.get_one()[0])
244
 
        self.assertEquals(variable.get(), value)
 
244
        self.assertEqual(variable.get(), value)
245
245
 
246
246
    def test_time(self):
247
247
        value = time(12, 34, 56, 78)
252
252
        result.set_variable(variable, result.get_one()[0])
253
253
        if not self.supports_microseconds:
254
254
            value = value.replace(microsecond=0)
255
 
        self.assertEquals(variable.get(), value)
 
255
        self.assertEqual(variable.get(), value)
256
256
 
257
257
    def test_timedelta(self):
258
258
        value = timedelta(12, 34, 56)
261
261
        result = self.connection.execute("SELECT td FROM datetime_test")
262
262
        variable = TimeDeltaVariable()
263
263
        result.set_variable(variable, result.get_one()[0])
264
 
        self.assertEquals(variable.get(), value)
 
264
        self.assertEqual(variable.get(), value)
265
265
 
266
266
    def test_pickle(self):
267
267
        value = {"a": 1, "b": 2}
271
271
        result = self.connection.execute("SELECT b FROM bin_test")
272
272
        variable = PickleVariable()
273
273
        result.set_variable(variable, result.get_one()[0])
274
 
        self.assertEquals(variable.get(), value)
 
274
        self.assertEqual(variable.get(), value)
275
275
 
276
276
    def test_binary(self):
277
277
        """Ensure database works with high bits and embedded zeros."""
279
279
        self.connection.execute("INSERT INTO bin_test (b) VALUES (?)",
280
280
                                (value,))
281
281
        result = self.connection.execute("SELECT b FROM bin_test")
282
 
        variable = RawStrVariable()
 
282
        variable = BytesVariable()
283
283
        result.set_variable(variable, result.get_one()[0])
284
 
        self.assertEquals(variable.get(), value)
 
284
        self.assertEqual(variable.get(), value)
285
285
 
286
286
    def test_binary_ascii(self):
287
287
        """Some databases like pysqlite2 may return unicode for strings."""
288
288
        self.connection.execute("INSERT INTO bin_test VALUES (10, 'Value')")
289
289
        result = self.connection.execute("SELECT b FROM bin_test")
290
 
        variable = RawStrVariable()
 
290
        variable = BytesVariable()
291
291
        # If the following doesn't raise a TypeError we're good.
292
292
        result.set_variable(variable, result.get_one()[0])
293
 
        self.assertEquals(variable.get(), b"Value")
 
293
        self.assertEqual(variable.get(), b"Value")
294
294
 
295
295
    def test_order_by_group_by(self):
296
296
        self.connection.execute("INSERT INTO test VALUES (100, 'Title 10')")
299
299
        title = Column("title", "test")
300
300
        expr = Select(Count(id), group_by=title, order_by=Count(id))
301
301
        result = self.connection.execute(expr)
302
 
        self.assertEquals(result.get_all(), [(1,), (3,)])
 
302
        self.assertEqual(result.get_all(), [(1,), (3,)])
303
303
 
304
304
    def test_set_decimal_variable_from_str_column(self):
305
305
        self.connection.execute("INSERT INTO test VALUES (40, '40.5')")
312
312
        variable.set("40.5", from_db=True)
313
313
        self.connection.execute("INSERT INTO test VALUES (40, ?)", (variable,))
314
314
        result = self.connection.execute("SELECT title FROM test WHERE id=40")
315
 
        self.assertEquals(result.get_one()[0], "40.5")
 
315
        self.assertEqual(result.get_one()[0], "40.5")
316
316
 
317
317
    def test_quoting(self):
318
318
        # FIXME "with'quote" should be in the list below, but it doesn't
322
322
            expr = Select(reserved_name,
323
323
                          tables=Alias(Select(Alias(1, reserved_name))))
324
324
            result = self.connection.execute(expr)
325
 
            self.assertEquals(result.get_one(), (1,))
 
325
            self.assertEqual(result.get_one(), (1,))
326
326
 
327
327
    def test_concurrent_behavior(self):
328
328
        """The default behavior should be to handle transactions in isolation.
342
342
        connection2 = self.database.connect()
343
343
        try:
344
344
            result = connection1.execute("SELECT title FROM test WHERE id=10")
345
 
            self.assertEquals(result.get_one(), ("Title 10",))
 
345
            self.assertEqual(result.get_one(), ("Title 10",))
346
346
            try:
347
347
                connection2.execute("UPDATE test SET title='Title 100' "
348
348
                                    "WHERE id=10")
349
349
                connection2.commit()
350
350
            except OperationalError as e:
351
 
                self.assertEquals(str(e), "database is locked") # SQLite blocks
 
351
                self.assertEqual(str(e), "database is locked") # SQLite blocks
352
352
            result = connection1.execute("SELECT title FROM test WHERE id=10")
353
 
            self.assertEquals(result.get_one(), ("Title 10",))
 
353
            self.assertEqual(result.get_one(), ("Title 10",))
354
354
        finally:
355
355
            connection1.rollback()
356
356
 
376
376
    def test_wb_result_get_one_goes_through_from_database(self):
377
377
        result = self.connection.execute("SELECT one, two FROM number")
378
378
        result.from_database = self.from_database
379
 
        self.assertEquals(result.get_one(), (2, 3))
 
379
        self.assertEqual(result.get_one(), (2, 3))
380
380
 
381
381
    def test_wb_result_get_all_goes_through_from_database(self):
382
382
        result = self.connection.execute("SELECT one, two FROM number")
383
383
        result.from_database = self.from_database
384
 
        self.assertEquals(result.get_all(), [(2, 3)])
 
384
        self.assertEqual(result.get_all(), [(2, 3)])
385
385
 
386
386
    def test_wb_result_iter_goes_through_from_database(self):
387
387
        result = self.connection.execute("SELECT one, two FROM number")
388
388
        result.from_database = self.from_database
389
 
        self.assertEquals(next(iter(result)), (2, 3))
 
389
        self.assertEqual(next(iter(result)), (2, 3))
390
390
 
391
391
    def test_rowcount_insert(self):
392
392
        # All supported backends support rowcount, so far.
393
393
        result = self.connection.execute(
394
394
            "INSERT INTO test VALUES (999, '999')")
395
 
        self.assertEquals(result.rowcount, 1)
 
395
        self.assertEqual(result.rowcount, 1)
396
396
 
397
397
    def test_rowcount_delete(self):
398
398
        # All supported backends support rowcount, so far.
399
399
        result = self.connection.execute("DELETE FROM test")
400
 
        self.assertEquals(result.rowcount, 2)
 
400
        self.assertEqual(result.rowcount, 2)
401
401
 
402
402
    def test_rowcount_update(self):
403
403
        # All supported backends support rowcount, so far.
404
404
        result = self.connection.execute(
405
405
            "UPDATE test SET title='whatever'")
406
 
        self.assertEquals(result.rowcount, 2)
 
406
        self.assertEqual(result.rowcount, 2)
407
407
 
408
408
    def test_expr_startswith(self):
409
409
        self.connection.execute("INSERT INTO test VALUES (30, '!!_%blah')")
412
412
        title = Column("title", SQLToken("test"))
413
413
        expr = Select(id, title.startswith(u"!!_%"))
414
414
        result = list(self.connection.execute(expr))
415
 
        self.assertEquals(result, [(30,)])
 
415
        self.assertEqual(result, [(30,)])
416
416
 
417
417
    def test_expr_endswith(self):
418
418
        self.connection.execute("INSERT INTO test VALUES (30, 'blah_%!!')")
421
421
        title = Column("title", SQLToken("test"))
422
422
        expr = Select(id, title.endswith(u"_%!!"))
423
423
        result = list(self.connection.execute(expr))
424
 
        self.assertEquals(result, [(30,)])
 
424
        self.assertEqual(result, [(30,)])
425
425
 
426
426
    def test_expr_contains_string(self):
427
427
        self.connection.execute("INSERT INTO test VALUES (30, 'blah_%!!x')")
430
430
        title = Column("title", SQLToken("test"))
431
431
        expr = Select(id, title.contains_string(u"_%!!"))
432
432
        result = list(self.connection.execute(expr))
433
 
        self.assertEquals(result, [(30,)])
 
433
        self.assertEqual(result, [(30,)])
434
434
 
435
435
    def test_block_access(self):
436
436
        """Access to the connection is blocked by block_access()."""