← Back to branch summary

~ubuntu-branches/ubuntu/saucy/migrate/saucy-proposed

« back to all changes in this revision

Viewing changes to test/changeset/test_constraint.py

  • Committer: Bazaar Package Importer
  • Author(s): Jan Dittberner
  • Date: 2010-07-12 00:24:57 UTC
  • mfrom: (1.1.5 upstream) (2.1.8 sid)
  • Revision ID: james.westby@ubuntu.com-20100712002457-4j2fdmco4u9kqzm5
Upload to unstable.

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/changeset/test_constraint.py
1
 
from sqlalchemy import *
2
 
from sqlalchemy.util import *
3
 
from test import fixture
4
 
from migrate.changeset import *
5
 
 
6
 
class TestConstraint(fixture.DB):
7
 
    level=fixture.DB.CONNECT
8
 
    def _setup(self, url):
9
 
        super(TestConstraint, self)._setup(url)
10
 
        self._create_table()
11
 
    def _teardown(self):
12
 
        if hasattr(self,'table') and self.engine.has_table(self.table.name):
13
 
            self.table.drop()
14
 
        super(TestConstraint, self)._teardown()
15
 
 
16
 
    def _create_table(self):
17
 
        self._connect(self.url)
18
 
        self.meta = MetaData(self.engine)
19
 
        self.table = Table('mytable',self.meta,
20
 
            Column('id',Integer),
21
 
            Column('fkey',Integer),
22
 
            mysql_engine='InnoDB'
23
 
        )
24
 
        if self.engine.has_table(self.table.name):
25
 
            self.table.drop()
26
 
        self.table.create()
27
 
        #self.assertEquals(self.table.primary_key,[])
28
 
        self.assertEquals(len(self.table.primary_key),0)
29
 
        self.assert_(isinstance(self.table.primary_key,
30
 
            schema.PrimaryKeyConstraint),self.table.primary_key.__class__)
31
 
    def _define_pk(self,*cols):
32
 
        # Add a pk by creating a PK constraint
33
 
        pk = PrimaryKeyConstraint(table=self.table, *cols)
34
 
        self.assertEquals(list(pk.columns),list(cols))
35
 
        if self.url.startswith('oracle'):
36
 
            # Can't drop Oracle PKs without an explicit name
37
 
            pk.name = 'fgsfds'
38
 
        pk.create()
39
 
        self.refresh_table()
40
 
        if not self.url.startswith('sqlite'):
41
 
            self.assertEquals(list(self.table.primary_key),list(cols))
42
 
        #self.assert_(self.table.primary_key.name is not None)
43
 
 
44
 
        # Drop the PK constraint
45
 
        if not self.url.startswith('oracle'):
46
 
            # Apparently Oracle PK names aren't introspected
47
 
            pk.name = self.table.primary_key.name
48
 
        pk.drop()
49
 
        self.refresh_table()
50
 
        #self.assertEquals(list(self.table.primary_key),list())
51
 
        self.assertEquals(len(self.table.primary_key),0)
52
 
        self.assert_(isinstance(self.table.primary_key,
53
 
            schema.PrimaryKeyConstraint),self.table.primary_key.__class__)
54
 
        return pk
55
 
 
56
 
    @fixture.usedb(not_supported='sqlite')
57
 
    def test_define_fk(self):
58
 
        """FK constraints can be defined, created, and dropped"""
59
 
        # FK target must be unique
60
 
        pk = PrimaryKeyConstraint(self.table.c.id, table=self.table)
61
 
        pk.create()
62
 
        # Add a FK by creating a FK constraint
63
 
        self.assertEquals(self.table.c.fkey.foreign_keys._list, [])
64
 
        fk = ForeignKeyConstraint([self.table.c.fkey],[self.table.c.id], table=self.table)
65
 
        self.assert_(self.table.c.fkey.foreign_keys._list is not [])
66
 
        self.assertEquals(list(fk.columns), [self.table.c.fkey])
67
 
        self.assertEquals([e.column for e in fk.elements],[self.table.c.id])
68
 
        self.assertEquals(list(fk.referenced),[self.table.c.id])
69
 
 
70
 
        if self.url.startswith('mysql'):
71
 
            # MySQL FKs need an index
72
 
            index = Index('index_name',self.table.c.fkey)
73
 
            index.create()
74
 
        if self.url.startswith('oracle'):
75
 
            # Oracle constraints need a name
76
 
            fk.name = 'fgsfds'
77
 
        print 'drop...'
78
 
        #self.engine.echo=True
79
 
        fk.create()
80
 
        #self.engine.echo=False
81
 
        print 'dropped'
82
 
        self.refresh_table()
83
 
        self.assert_(self.table.c.fkey.foreign_keys._list is not [])
84
 
 
85
 
        print 'drop...'
86
 
        #self.engine.echo=True
87
 
        fk.drop()
88
 
        #self.engine.echo=False
89
 
        print 'dropped'
90
 
        self.refresh_table()
91
 
        self.assertEquals(self.table.c.fkey.foreign_keys._list, [])
92
 
 
93
 
    @fixture.usedb()
94
 
    def test_define_pk(self):
95
 
        """PK constraints can be defined, created, and dropped"""
96
 
        self._define_pk(self.table.c.id)
97
 
 
98
 
    @fixture.usedb()
99
 
    def test_define_pk_multi(self):
100
 
        """Multicolumn PK constraints can be defined, created, and dropped"""
101
 
        #self.engine.echo=True
102
 
        self._define_pk(self.table.c.id,self.table.c.fkey)
103
 
 
104
 
 
105
 
class TestAutoname(fixture.DB):
106
 
    level=fixture.DB.CONNECT
107
 
 
108
 
    def _setup(self, url):
109
 
        super(TestAutoname, self)._setup(url)
110
 
        self._connect(self.url)
111
 
        self.meta = MetaData(self.engine)
112
 
        self.table = Table('mytable',self.meta,
113
 
            Column('id',Integer),
114
 
            Column('fkey',String(40)),
115
 
        )
116
 
        if self.engine.has_table(self.table.name):
117
 
            self.table.drop()
118
 
        self.table.create()
119
 
        
120
 
    def _teardown(self):
121
 
        if hasattr(self,'table') and self.engine.has_table(self.table.name):
122
 
            self.table.drop()
123
 
        super(TestAutoname, self)._teardown()
124
 
        
125
 
    @fixture.usedb(not_supported='oracle')
126
 
    def test_autoname(self):
127
 
        """Constraints can guess their name if none is given"""
128
 
        # Don't supply a name; it should create one
129
 
        cons = PrimaryKeyConstraint(self.table.c.id)
130
 
        cons.create()
131
 
        self.refresh_table()
132
 
        if not self.url.startswith('sqlite'):
133
 
            self.assertEquals(list(cons.columns),list(self.table.primary_key))
134
 
 
135
 
        # Remove the name, drop the constraint; it should succeed
136
 
        cons.name = None
137
 
        cons.drop()
138
 
        self.refresh_table()
139
 
        self.assertEquals(list(),list(self.table.primary_key))