2
# Copyright (c) 2008 Canonical
4
# Written by James Henstridge <jamesh@canonical.com>
6
# This file is part of Storm Object Relational Mapper.
8
# Storm is free software; you can redistribute it and/or modify
9
# it under the terms of the GNU Lesser General Public License as
10
# published by the Free Software Foundation; either version 2.1 of
11
# the License, or (at your option) any later version.
13
# Storm is distributed in the hope that it will be useful,
14
# but WITHOUT ANY WARRANTY; without even the implied warranty of
15
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16
# GNU Lesser General Public License for more details.
18
# You should have received a copy of the GNU Lesser General Public License
19
# along with this program. If not, see <http://www.gnu.org/licenses/>.
25
from django.conf import settings
28
have_django_and_transaction = False
30
have_django_and_transaction = True
31
from storm.django import stores
32
from storm.zope.zstorm import global_zstorm, StoreDataManager
34
from tests.helper import TestHelper
37
class DjangoBackendTests(object):
39
def is_supported(self):
40
return have_django_and_transaction and self.get_store_uri() is not None
43
super(DjangoBackendTests, self).setUp()
44
settings.configure(STORM_STORES={})
45
settings.MIDDLEWARE_CLASSES += (
46
"storm.django.middleware.ZopeTransactionMiddleware",)
48
settings.DATABASE_ENGINE = "storm.django.backend"
49
settings.DATABASE_NAME = "django"
50
settings.STORM_STORES["django"] = self.get_store_uri()
51
stores.have_configured_stores = False
57
settings._target = None
58
global_zstorm._reset()
59
stores.have_configured_stores = False
60
transaction.manager.free(transaction.get())
61
super(DjangoBackendTests, self).tearDown()
63
def get_store_uri(self):
64
raise NotImplementedError
66
def get_wrapper_class(self):
67
raise NotImplementedError
69
def create_tables(self):
70
raise NotImplementedError
72
def drop_tables(self):
73
raise NotImplementedError
75
def test_create_wrapper(self):
76
from storm.django.backend import base
77
wrapper = base.DatabaseWrapper(**settings.DATABASE_OPTIONS)
78
self.assertTrue(isinstance(wrapper, self.get_wrapper_class()))
80
# The wrapper uses the same database connection as the store.
81
store = stores.get_store("django")
82
self.assertEqual(store._connection._raw_connection, wrapper.connection)
84
def _isInTransaction(self, store):
85
"""Check if a Store is part of the current transaction."""
86
for dm in transaction.get()._resources:
87
if isinstance(dm, StoreDataManager) and dm._store is store:
91
def assertInTransaction(self, store):
92
"""Check that the given store is joined to the transaction."""
93
self.assertTrue(self._isInTransaction(store),
94
"%r should be joined to the transaction" % store)
96
def test_using_wrapper_joins_transaction(self):
97
from storm.django.backend import base
98
wrapper = base.DatabaseWrapper(**settings.DATABASE_OPTIONS)
99
cursor = wrapper.cursor()
100
cursor.execute("SELECT 1")
101
self.assertInTransaction(stores.get_store("django"))
103
def test_commit(self):
104
from storm.django.backend import base
105
wrapper = base.DatabaseWrapper(**settings.DATABASE_OPTIONS)
106
cursor = wrapper.cursor()
107
cursor.execute("INSERT INTO django_test (title) VALUES ('foo')")
110
cursor = wrapper.cursor()
111
cursor.execute("SELECT title FROM django_test")
112
result = cursor.fetchall()
113
self.assertEqual(len(result), 1)
114
self.assertEqual(result[0][0], "foo")
116
def test_rollback(self):
117
from storm.django.backend import base
118
wrapper = base.DatabaseWrapper(**settings.DATABASE_OPTIONS)
119
cursor = wrapper.cursor()
120
cursor.execute("INSERT INTO django_test (title) VALUES ('foo')")
123
cursor = wrapper.cursor()
124
cursor.execute("SELECT title FROM django_test")
125
result = cursor.fetchall()
126
self.assertEqual(len(result), 0)
129
class PostgresDjangoBackendTests(DjangoBackendTests, TestHelper):
131
def get_store_uri(self):
132
return os.environ.get("STORM_POSTGRES_URI")
134
def get_wrapper_class(self):
135
from storm.django.backend import base
136
return base.PostgresStormDatabaseWrapper
138
def create_tables(self):
139
store = stores.get_store("django")
140
store.execute("CREATE TABLE django_test ("
141
" id SERIAL PRIMARY KEY,"
145
def drop_tables(self):
146
store = stores.get_store("django")
147
store.execute("DROP TABLE django_test")
151
class MySQLDjangoBackendTests(DjangoBackendTests, TestHelper):
153
def get_store_uri(self):
154
return os.environ.get("STORM_MYSQL_URI")
156
def get_wrapper_class(self):
157
from storm.django.backend import base
158
return base.MySQLStormDatabaseWrapper
160
def create_tables(self):
161
store = stores.get_store("django")
162
store.execute("CREATE TABLE django_test ("
163
" id INT AUTO_INCREMENT PRIMARY KEY,"
164
" title TEXT) ENGINE=InnoDB")
167
def drop_tables(self):
168
store = stores.get_store("django")
169
store.execute("DROP TABLE django_test")
added
removed
tests/django/backend.py
