← Back to branch summary

~lifeless/storm/bug-620615

« back to all changes in this revision

Viewing changes to tests/store/base.py

  • Committer: Gustavo Niemeyer
  • Date: 2006-05-25 20:05:42 UTC
  • Revision ID: gustavo@niemeyer.net-20060525200541-c171f5100ab0a30f
- Implemented MySQL database support! Store tests all pass!
- Splitted out the store tests into per-database files.
- Implemented DateTime, Date, and Time kinds, and support for
  all backends.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/store/base.py
1
1
import gc
2
 
import os
3
 
 
4
 
from storm.databases.sqlite import SQLite
5
 
from storm.databases.postgres import Postgres
6
2
 
7
3
from storm.properties import get_obj_info
8
4
from storm.references import Reference
12
8
from storm.kinds import StrKind, IntKind
13
9
from storm.store import *
14
10
 
15
 
from tests.helper import *
16
 
 
17
11
 
18
12
class Test(object):
19
13
    __table__ = "test", "id"
48
42
    title = Property(kind=DecorateKind())
49
43
 
50
44
 
51
 
class StoreTest(TestHelper):
52
 
 
53
 
    helpers = [MakePath]
 
45
class StoreTest(object):
54
46
 
55
47
    def setUp(self):
56
 
        TestHelper.setUp(self)
57
48
        self.create_database()
58
49
        self.drop_tables()
59
50
        self.create_tables()
65
56
        self.drop_sample_data()
66
57
        self.drop_tables()
67
58
        self.drop_database()
68
 
        TestHelper.tearDown(self)
69
59
 
70
60
    def create_database(self):
71
61
        raise NotImplementedError
96
86
        connection = self.database.connect()
97
87
        try:
98
88
            connection.execute("DROP TABLE test")
 
89
            connection.commit()
 
90
        except:
 
91
            connection.rollback()
 
92
        try:
99
93
            connection.execute("DROP TABLE other")
100
94
            connection.commit()
101
95
        except:
107
101
    def get_items(self):
108
102
        # Bypass the store to avoid flushing.
109
103
        connection = self.store._connection
110
 
        result = connection.execute("SELECT * FROM test DATABASE ORDER BY id")
 
104
        result = connection.execute("SELECT * FROM test ORDER BY id")
111
105
        return list(result)
112
106
 
113
107
    def get_committed_items(self):
114
108
        connection = self.database.connect()
115
 
        result = connection.execute("SELECT * FROM test DATABASE ORDER BY id")
 
109
        result = connection.execute("SELECT * FROM test ORDER BY id")
116
110
        return list(result)
117
111
 
118
112
 
1241
1235
 
1242
1236
        self.assertEquals(obj.id, 21)
1243
1237
        self.assertEquals(obj.title, "to_kind(Title 20)")
1244
 
 
1245
 
 
1246
 
class SQLiteStoreTest(StoreTest):
1247
 
 
1248
 
    helpers = [MakePath]
1249
 
 
1250
 
    def create_database(self):
1251
 
        self.database = SQLite(self.make_path())
1252
 
 
1253
 
    def create_tables(self):
1254
 
        connection = self.database.connect()
1255
 
        connection.execute("CREATE TABLE test "
1256
 
                           "(id INTEGER PRIMARY KEY,"
1257
 
                           " title VARCHAR DEFAULT 'Default Title')")
1258
 
        connection.execute("CREATE TABLE other "
1259
 
                           "(id INTEGER PRIMARY KEY,"
1260
 
                           " test_id INTEGER,"
1261
 
                           " other_title VARCHAR)")
1262
 
        connection.commit()
1263
 
 
1264
 
    def drop_tables(self):
1265
 
        pass
1266
 
 
1267
 
 
1268
 
class PostgresStoreTest(StoreTest):
1269
 
 
1270
 
    def is_supported(self):
1271
 
        return bool(os.environ.get("STORM_POSTGRES_DBNAME"))
1272
 
 
1273
 
    def create_database(self):
1274
 
        self.database = Postgres(os.environ["STORM_POSTGRES_DBNAME"])
1275
 
 
1276
 
    def create_tables(self):
1277
 
        connection = self.database.connect()
1278
 
        connection.execute("CREATE TABLE test "
1279
 
                           "(id SERIAL PRIMARY KEY,"
1280
 
                           " title VARCHAR DEFAULT 'Default Title')")
1281
 
        connection.execute("CREATE TABLE other "
1282
 
                           "(id SERIAL PRIMARY KEY,"
1283
 
                           " test_id INTEGER,"
1284
 
                           " other_title VARCHAR)")
1285
 
        connection.commit()
1286
 
 
1287
 
del StoreTest