← Back to branch summary

~sidnei/storm/mssql-support

« back to all changes in this revision

Viewing changes to tests/databases/base.py

  • Committer: Gustavo Niemeyer
  • Date: 2007-08-07 23:15:09 UTC
  • mfrom: (156.1.8 preset-primary-key)
  • Revision ID: gustavo@niemeyer.net-20070807231509-iw58uiygzuef5wth
Merged preset-primary-key branch [r=radix,jkakar] [f=126080]

This branch introduces a new hook which allows backends to modify
variables which are part of the primary key of an object before
they are inserted in the database.  Also, lazy values may now be
used in the primary key, which means that it's possible to set these
primary variables to SQL expressions if needed.

Other minor changes include an optimization in Result: it will
preset the raw cursor's arraysize to a better value if it's found
to be 1.  Result.__iter__() was also fixed to parse rows through
the from_database hook.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/databases/base.py
65
65
        raise NotImplementedError
66
66
 
67
67
    def create_sample_data(self):
 
68
        self.connection.execute("INSERT INTO number VALUES (1, 2, 3)")
68
69
        self.connection.execute("INSERT INTO test VALUES (10, 'Title 10')")
69
70
        self.connection.execute("INSERT INTO test VALUES (20, 'Title 20')")
70
71
        self.connection.commit()
73
74
        pass
74
75
 
75
76
    def drop_tables(self):
76
 
        try:
77
 
            self.connection.execute("DROP TABLE test")
78
 
            self.connection.commit()
79
 
        except:
80
 
            self.connection.rollback()
81
 
        try:
82
 
            self.connection.execute("DROP TABLE datetime_test")
83
 
            self.connection.commit()
84
 
        except:
85
 
            self.connection.rollback()
86
 
        try:
87
 
            self.connection.execute("DROP TABLE bin_test")
88
 
            self.connection.commit()
89
 
        except:
90
 
            self.connection.rollback()
 
77
        for table in ["number", "test", "datetime_test", "bin_test"]:
 
78
            try:
 
79
                self.connection.execute("DROP TABLE " + table)
 
80
                self.connection.commit()
 
81
            except:
 
82
                self.connection.rollback()
91
83
 
92
84
    def drop_database(self):
93
85
        pass
138
130
        self.assertTrue(isinstance(row[0], unicode))
139
131
 
140
132
    def test_execute_params(self):
141
 
        result = self.connection.execute("SELECT 1 FROM (SELECT 1) AS ALIAS "
 
133
        result = self.connection.execute("SELECT one FROM number "
142
134
                                         "WHERE 1=?", (1,))
143
135
        self.assertTrue(result.get_one())
144
 
        result = self.connection.execute("SELECT 1 FROM (SELECT 1) AS ALIAS "
 
136
        result = self.connection.execute("SELECT one FROM number "
145
137
                                         "WHERE 1=?", (2,))
146
138
        self.assertFalse(result.get_one())
147
139
 
148
140
    def test_execute_empty_params(self):
149
 
        result = self.connection.execute("SELECT 1", ())
 
141
        result = self.connection.execute("SELECT one FROM number", ())
150
142
        self.assertTrue(result.get_one())
151
143
 
152
144
    def test_execute_expression(self):
330
322
        finally:
331
323
            connection1.rollback()
332
324
 
 
325
    def from_database(self, row):
 
326
        return [int(item)+1 for item in row]
 
327
 
 
328
    def test_wb_result_get_one_goes_through_from_database(self):
 
329
        result = self.connection.execute("SELECT one, two FROM number")
 
330
        result._from_database = self.from_database
 
331
        self.assertEquals(result.get_one(), (2, 3))
 
332
 
 
333
    def test_wb_result_get_all_goes_through_from_database(self):
 
334
        result = self.connection.execute("SELECT one, two FROM number")
 
335
        result._from_database = self.from_database
 
336
        self.assertEquals(result.get_all(), [(2, 3)])
 
337
 
 
338
    def test_wb_result_iter_goes_through_from_database(self):
 
339
        result = self.connection.execute("SELECT one, two FROM number")
 
340
        result._from_database = self.from_database
 
341
        self.assertEquals(iter(result).next(), (2, 3))
 
342
 
333
343
 
334
344
class UnsupportedDatabaseTest(object):
335
345