← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/sql/test_metadata.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/sql/test_metadata.py
1
 
from test.lib.testing import assert_raises
2
 
from test.lib.testing import assert_raises_message
3
 
from test.lib.testing import emits_warning
 
1
from sqlalchemy.testing import assert_raises
 
2
from sqlalchemy.testing import assert_raises_message
 
3
from sqlalchemy.testing import emits_warning
4
4
 
5
5
import pickle
6
6
from sqlalchemy import Integer, String, UniqueConstraint, \
7
7
    CheckConstraint, ForeignKey, MetaData, Sequence, \
8
8
    ForeignKeyConstraint, ColumnDefault, Index, event,\
9
 
    events, Unicode
10
 
from test.lib.schema import Table, Column
 
9
    events, Unicode, types as sqltypes
 
10
from sqlalchemy.testing.schema import Table, Column
11
11
from sqlalchemy import schema, exc
12
12
import sqlalchemy as tsa
13
 
from test.lib import fixtures
14
 
from test.lib import testing
15
 
from test.lib.testing import ComparesTables, AssertsCompiledSQL
16
 
from test.lib.testing import eq_, is_
 
13
from sqlalchemy.testing import fixtures
 
14
from sqlalchemy import testing
 
15
from sqlalchemy.testing import ComparesTables, AssertsCompiledSQL
 
16
from sqlalchemy.testing import eq_, is_
17
17
 
18
18
class MetaDataTest(fixtures.TestBase, ComparesTables):
19
19
    def test_metadata_connect(self):
54
54
            Column(Integer(), ForeignKey('bat.blah'), doc="this is a col"),
55
55
            Column('bar', Integer(), ForeignKey('bat.blah'), primary_key=True,
56
56
                                                            key='bar'),
57
 
            Column('bar', Integer(), info={'foo':'bar'}),
 
57
            Column('bar', Integer(), info={'foo': 'bar'}),
58
58
        ]:
59
59
            c2 = col.copy()
60
60
            for attr in ('name', 'type', 'nullable',
148
148
    def test_dupe_tables(self):
149
149
        metadata = self.metadata
150
150
        Table('table1', metadata,
151
 
            Column('col1', Integer, primary_key=True),
152
 
            Column('col2', String(20)))
 
151
                Column('col1', Integer, primary_key=True),
 
152
                Column('col2', String(20)))
153
153
 
154
154
        metadata.create_all()
155
155
        Table('table1', metadata, autoload=True)
156
156
        def go():
157
157
            Table('table1', metadata,
158
 
                Column('col1', Integer, primary_key=True),
159
 
                Column('col2', String(20)))
 
158
                    Column('col1', Integer, primary_key=True),
 
159
                    Column('col2', String(20)))
160
160
        assert_raises_message(
161
161
            tsa.exc.InvalidRequestError,
162
 
            "Table 'table1' is already defined for this "\
163
 
            "MetaData instance.  Specify 'extend_existing=True' "\
164
 
            "to redefine options and columns on an existing "\
165
 
            "Table object.",
 
162
                "Table 'table1' is already defined for this "
 
163
                "MetaData instance.  Specify 'extend_existing=True' "
 
164
                "to redefine options and columns on an existing "
 
165
                "Table object.",
166
166
            go
167
167
        )
168
168
 
349
349
        finally:
350
350
            meta.drop_all(testing.db)
351
351
 
 
352
    def test_col_key_fk_parent_tometadata(self):
 
353
        # test #2643
 
354
        m1 = MetaData()
 
355
        a = Table('a', m1, Column('x', Integer))
 
356
        b = Table('b', m1, Column('x', Integer, ForeignKey('a.x'), key='y'))
 
357
        assert b.c.y.references(a.c.x)
 
358
 
 
359
        m2 = MetaData()
 
360
        b2 = b.tometadata(m2)
 
361
        a2 = a.tometadata(m2)
 
362
        assert b2.c.y.references(a2.c.x)
 
363
 
 
364
 
352
365
    def test_pickle_metadata_sequence_restated(self):
353
366
        m1 = MetaData()
354
367
        Table('a', m1,
531
544
 
532
545
        for i, (name, metadata, schema, quote_schema,
533
546
                        exp_schema, exp_quote_schema) in enumerate([
534
 
            ('t1', m1, None, None, 'sch1', None),
535
 
            ('t2', m1, 'sch2', None, 'sch2', None),
536
 
            ('t3', m1, 'sch2', True, 'sch2', True),
537
 
            ('t4', m1, 'sch1', None, 'sch1', None),
538
 
            ('t1', m2, None, None, 'sch1', True),
539
 
            ('t2', m2, 'sch2', None, 'sch2', None),
540
 
            ('t3', m2, 'sch2', True, 'sch2', True),
541
 
            ('t4', m2, 'sch1', None, 'sch1', None),
542
 
            ('t1', m3, None, None, 'sch1', False),
543
 
            ('t2', m3, 'sch2', None, 'sch2', None),
544
 
            ('t3', m3, 'sch2', True, 'sch2', True),
545
 
            ('t4', m3, 'sch1', None, 'sch1', None),
546
 
            ('t1', m4, None, None, None, None),
547
 
            ('t2', m4, 'sch2', None, 'sch2', None),
548
 
            ('t3', m4, 'sch2', True, 'sch2', True),
549
 
            ('t4', m4, 'sch1', None, 'sch1', None),
550
 
        ]):
 
547
                                ('t1', m1, None, None, 'sch1', None),
 
548
                                ('t2', m1, 'sch2', None, 'sch2', None),
 
549
                                ('t3', m1, 'sch2', True, 'sch2', True),
 
550
                                ('t4', m1, 'sch1', None, 'sch1', None),
 
551
                                ('t1', m2, None, None, 'sch1', True),
 
552
                                ('t2', m2, 'sch2', None, 'sch2', None),
 
553
                                ('t3', m2, 'sch2', True, 'sch2', True),
 
554
                                ('t4', m2, 'sch1', None, 'sch1', None),
 
555
                                ('t1', m3, None, None, 'sch1', False),
 
556
                                ('t2', m3, 'sch2', None, 'sch2', None),
 
557
                                ('t3', m3, 'sch2', True, 'sch2', True),
 
558
                                ('t4', m3, 'sch1', None, 'sch1', None),
 
559
                                ('t1', m4, None, None, None, None),
 
560
                                ('t2', m4, 'sch2', None, 'sch2', None),
 
561
                                ('t3', m4, 'sch2', True, 'sch2', True),
 
562
                                ('t4', m4, 'sch1', None, 'sch1', None),
 
563
                        ]):
551
564
            kw = {}
552
565
            if schema is not None:
553
566
                kw['schema'] = schema
555
568
                kw['quote_schema'] = quote_schema
556
569
            t = Table(name, metadata, **kw)
557
570
            eq_(t.schema, exp_schema, "test %d, table schema" % i)
558
 
            eq_(t.quote_schema, exp_quote_schema, "test %d, table quote_schema" % i)
 
571
            eq_(t.quote_schema, exp_quote_schema,
 
572
                            "test %d, table quote_schema" % i)
559
573
            seq = Sequence(name, metadata=metadata, **kw)
560
574
            eq_(seq.schema, exp_schema, "test %d, seq schema" % i)
561
 
            eq_(seq.quote_schema, exp_quote_schema, "test %d, seq quote_schema" % i)
 
575
            eq_(seq.quote_schema, exp_quote_schema,
 
576
                            "test %d, seq quote_schema" % i)
562
577
 
563
578
    def test_manual_dependencies(self):
564
579
        meta = MetaData()
667
682
            )
668
683
 
669
684
class TableTest(fixtures.TestBase, AssertsCompiledSQL):
 
685
    @testing.skip_if('mssql', 'different col format')
670
686
    def test_prefixes(self):
671
687
        from sqlalchemy import Table
672
688
        table1 = Table("temporary_table_1", MetaData(),
682
698
                      Column("col1", Integer),
683
699
                      prefixes=["VIRTUAL"])
684
700
        self.assert_compile(
685
 
          schema.CreateTable(table2),
686
 
          "CREATE VIRTUAL TABLE temporary_table_2 (col1 INTEGER)"
 
701
            schema.CreateTable(table2),
 
702
            "CREATE VIRTUAL TABLE temporary_table_2 (col1 INTEGER)"
687
703
        )
688
704
 
689
705
    def test_table_info(self):
735
751
        )
736
752
        is_(t._autoincrement_column, t.c.id)
737
753
 
 
754
class SchemaTypeTest(fixtures.TestBase):
 
755
    class MyType(sqltypes.SchemaType, sqltypes.TypeEngine):
 
756
        column = None
 
757
        table = None
 
758
        evt_targets = ()
 
759
 
 
760
        def _set_table(self, column, table):
 
761
            super(SchemaTypeTest.MyType, self)._set_table(column, table)
 
762
            self.column = column
 
763
            self.table = table
 
764
 
 
765
        def _on_table_create(self, target, bind, **kw):
 
766
            self.evt_targets += (target,)
 
767
 
 
768
    def test_independent_schema(self):
 
769
        m = MetaData()
 
770
        type_ = self.MyType(schema="q")
 
771
        t1 = Table('x', m, Column("y", type_), schema="z")
 
772
        eq_(t1.c.y.type.schema, "q")
 
773
 
 
774
    def test_inherit_schema(self):
 
775
        m = MetaData()
 
776
        type_ = self.MyType(schema="q", inherit_schema=True)
 
777
        t1 = Table('x', m, Column("y", type_), schema="z")
 
778
        eq_(t1.c.y.type.schema, "z")
 
779
 
 
780
    def test_independent_schema_enum(self):
 
781
        m = MetaData()
 
782
        type_ = sqltypes.Enum("a", schema="q")
 
783
        t1 = Table('x', m, Column("y", type_), schema="z")
 
784
        eq_(t1.c.y.type.schema, "q")
 
785
 
 
786
    def test_inherit_schema_enum(self):
 
787
        m = MetaData()
 
788
        type_ = sqltypes.Enum("a", "b", "c", schema="q", inherit_schema=True)
 
789
        t1 = Table('x', m, Column("y", type_), schema="z")
 
790
        eq_(t1.c.y.type.schema, "z")
 
791
 
 
792
    def test_tometadata_copy_type(self):
 
793
        m1 = MetaData()
 
794
 
 
795
        type_ = self.MyType()
 
796
        t1 = Table('x', m1, Column("y", type_))
 
797
 
 
798
        m2 = MetaData()
 
799
        t2 = t1.tometadata(m2)
 
800
 
 
801
        # metadata isn't set
 
802
        is_(t2.c.y.type.metadata, None)
 
803
 
 
804
        # our test type sets table, though
 
805
        is_(t2.c.y.type.table, t2)
 
806
 
 
807
    def test_tometadata_independent_schema(self):
 
808
        m1 = MetaData()
 
809
 
 
810
        type_ = self.MyType()
 
811
        t1 = Table('x', m1, Column("y", type_))
 
812
 
 
813
        m2 = MetaData()
 
814
        t2 = t1.tometadata(m2, schema="bar")
 
815
 
 
816
        eq_(t2.c.y.type.schema, None)
 
817
 
 
818
    def test_tometadata_inherit_schema(self):
 
819
        m1 = MetaData()
 
820
 
 
821
        type_ = self.MyType(inherit_schema=True)
 
822
        t1 = Table('x', m1, Column("y", type_))
 
823
 
 
824
        m2 = MetaData()
 
825
        t2 = t1.tometadata(m2, schema="bar")
 
826
 
 
827
        eq_(t1.c.y.type.schema, None)
 
828
        eq_(t2.c.y.type.schema, "bar")
 
829
 
 
830
    def test_tometadata_independent_events(self):
 
831
        m1 = MetaData()
 
832
 
 
833
        type_ = self.MyType()
 
834
        t1 = Table('x', m1, Column("y", type_))
 
835
 
 
836
        m2 = MetaData()
 
837
        t2 = t1.tometadata(m2)
 
838
 
 
839
        t1.dispatch.before_create(t1, testing.db)
 
840
        eq_(t1.c.y.type.evt_targets, (t1,))
 
841
        eq_(t2.c.y.type.evt_targets, ())
 
842
 
 
843
        t2.dispatch.before_create(t2, testing.db)
 
844
        t2.dispatch.before_create(t2, testing.db)
 
845
        eq_(t1.c.y.type.evt_targets, (t1,))
 
846
        eq_(t2.c.y.type.evt_targets, (t2, t2))
 
847
 
 
848
 
 
849
 
738
850
class SchemaTest(fixtures.TestBase, AssertsCompiledSQL):
739
851
 
740
852
    def test_default_schema_metadata_fk(self):
830
942
                          Unicode), autoload=True)
831
943
        assert_raises_message(
832
944
            exc.InvalidRequestError,
833
 
            "Table 'users' is already defined for this "\
 
945
                "Table 'users' is already defined for this "\
834
946
                "MetaData instance.",
835
947
            go
836
948
        )
1071
1183
        assert s1.c.a.references(t1.c.a)
1072
1184
        assert not s1.c.a.references(t1.c.b)
1073
1185
 
 
1186
    def test_invalid_composite_fk_check(self):
 
1187
        m = MetaData()
 
1188
        t1 = Table('t1', m, Column('x', Integer), Column('y', Integer),
 
1189
            ForeignKeyConstraint(['x', 'y'], ['t2.x', 't3.y'])
 
1190
        )
 
1191
        t2 = Table('t2', m, Column('x', Integer))
 
1192
        t3 = Table('t3', m, Column('y', Integer))
 
1193
 
 
1194
        assert_raises_message(
 
1195
            exc.ArgumentError,
 
1196
            r"ForeignKeyConstraint on t1\(x, y\) refers to "
 
1197
                "multiple remote tables: t2 and t3",
 
1198
            t1.join, t2
 
1199
        )
 
1200
        assert_raises_message(
 
1201
            exc.ArgumentError,
 
1202
            r"ForeignKeyConstraint on t1\(x, y\) refers to "
 
1203
                "multiple remote tables: t2 and t3",
 
1204
            t1.join, t3
 
1205
        )
 
1206
 
 
1207
        assert_raises_message(
 
1208
            exc.ArgumentError,
 
1209
            r"ForeignKeyConstraint on t1\(x, y\) refers to "
 
1210
                "multiple remote tables: t2 and t3",
 
1211
            schema.CreateTable(t1).compile
 
1212
        )
 
1213
 
 
1214
    def test_constraint_copied_to_proxy_ok(self):
 
1215
        m = MetaData()
 
1216
        t1 = Table('t1', m, Column('id', Integer, primary_key=True))
 
1217
        t2 = Table('t2', m, Column('id', Integer, ForeignKey('t1.id'),
 
1218
                        primary_key=True))
 
1219
 
 
1220
        s = tsa.select([t2])
 
1221
        t2fk = list(t2.c.id.foreign_keys)[0]
 
1222
        sfk = list(s.c.id.foreign_keys)[0]
 
1223
 
 
1224
        # the two FKs share the ForeignKeyConstraint
 
1225
        is_(
 
1226
            t2fk.constraint,
 
1227
            sfk.constraint
 
1228
        )
 
1229
 
 
1230
        # but the ForeignKeyConstraint isn't
 
1231
        # aware of the select's FK
 
1232
        eq_(
 
1233
            t2fk.constraint.elements,
 
1234
            [t2fk]
 
1235
        )
 
1236
 
 
1237
 
1074
1238
class ColumnDefinitionTest(AssertsCompiledSQL, fixtures.TestBase):
1075
1239
    """Test Column() construction."""
1076
1240
 
1217
1381
            getattr, select([t1.select().alias()]), 'c'
1218
1382
        )
1219
1383
 
 
1384
    def test_custom_create(self):
 
1385
        from sqlalchemy.ext.compiler import compiles, deregister
 
1386
 
 
1387
        @compiles(schema.CreateColumn)
 
1388
        def compile(element, compiler, **kw):
 
1389
            column = element.element
 
1390
 
 
1391
            if "special" not in column.info:
 
1392
                return compiler.visit_create_column(element, **kw)
 
1393
 
 
1394
            text = "%s SPECIAL DIRECTIVE %s" % (
 
1395
                    column.name,
 
1396
                    compiler.type_compiler.process(column.type)
 
1397
                )
 
1398
            default = compiler.get_column_default_string(column)
 
1399
            if default is not None:
 
1400
                text += " DEFAULT " + default
 
1401
 
 
1402
            if not column.nullable:
 
1403
                text += " NOT NULL"
 
1404
 
 
1405
            if column.constraints:
 
1406
                text += " ".join(
 
1407
                            compiler.process(const)
 
1408
                            for const in column.constraints)
 
1409
            return text
 
1410
 
 
1411
        t = Table('mytable', MetaData(),
 
1412
                Column('x', Integer, info={"special": True}, primary_key=True),
 
1413
                Column('y', String(50)),
 
1414
                Column('z', String(20), info={"special": True})
 
1415
            )
 
1416
 
 
1417
        self.assert_compile(
 
1418
            schema.CreateTable(t),
 
1419
            "CREATE TABLE mytable (x SPECIAL DIRECTIVE INTEGER "
 
1420
                "NOT NULL, y VARCHAR(50), "
 
1421
                "z SPECIAL DIRECTIVE VARCHAR(20), PRIMARY KEY (x))"
 
1422
        )
 
1423
 
 
1424
        deregister(schema.CreateColumn)
 
1425
 
1220
1426
class ColumnDefaultsTest(fixtures.TestBase):
1221
1427
    """test assignment of default fixures to columns"""
1222
1428
 
1229
1435
        assert c.server_default is target
1230
1436
        assert target.column is c
1231
1437
 
 
1438
    def test_onupdate_default_not_server_default_one(self):
 
1439
        target1 = schema.DefaultClause('y')
 
1440
        target2 = schema.DefaultClause('z')
 
1441
 
 
1442
        c = self._fixture(server_default=target1, server_onupdate=target2)
 
1443
        eq_(c.server_default.arg, 'y')
 
1444
        eq_(c.server_onupdate.arg, 'z')
 
1445
 
 
1446
    def test_onupdate_default_not_server_default_two(self):
 
1447
        target1 = schema.DefaultClause('y', for_update=True)
 
1448
        target2 = schema.DefaultClause('z', for_update=True)
 
1449
 
 
1450
        c = self._fixture(server_default=target1, server_onupdate=target2)
 
1451
        eq_(c.server_default.arg, 'y')
 
1452
        eq_(c.server_onupdate.arg, 'z')
 
1453
 
 
1454
    def test_onupdate_default_not_server_default_three(self):
 
1455
        target1 = schema.DefaultClause('y', for_update=False)
 
1456
        target2 = schema.DefaultClause('z', for_update=True)
 
1457
 
 
1458
        c = self._fixture(target1, target2)
 
1459
        eq_(c.server_default.arg, 'y')
 
1460
        eq_(c.server_onupdate.arg, 'z')
 
1461
 
 
1462
    def test_onupdate_default_not_server_default_four(self):
 
1463
        target1 = schema.DefaultClause('y', for_update=False)
 
1464
 
 
1465
        c = self._fixture(server_onupdate=target1)
 
1466
        is_(c.server_default, None)
 
1467
        eq_(c.server_onupdate.arg, 'y')
 
1468
 
1232
1469
    def test_server_default_keyword_as_schemaitem(self):
1233
1470
        target = schema.DefaultClause('y')
1234
1471
        c = self._fixture(server_default=target)
1340
1577
    def test_all_events(self):
1341
1578
        canary = []
1342
1579
        def before_attach(obj, parent):
1343
 
            canary.append("%s->%s" % (obj.__class__.__name__, parent.__class__.__name__))
 
1580
            canary.append("%s->%s" % (obj.__class__.__name__,
 
1581
                                parent.__class__.__name__))
1344
1582
 
1345
1583
        def after_attach(obj, parent):
1346
1584
            canary.append("%s->%s" % (obj.__class__.__name__, parent))
1375
1613
 
1376
1614
        def evt(target):
1377
1615
            def before_attach(obj, parent):
1378
 
                canary.append("%s->%s" % (target.__name__, parent.__class__.__name__))
 
1616
                canary.append("%s->%s" % (target.__name__,
 
1617
                                        parent.__class__.__name__))
1379
1618
 
1380
1619
            def after_attach(obj, parent):
1381
1620
                canary.append("%s->%s" % (target.__name__, parent))
1383
1622
            event.listen(target, "after_parent_attach", after_attach)
1384
1623
 
1385
1624
        for target in [
1386
 
            schema.ForeignKeyConstraint, schema.PrimaryKeyConstraint, schema.UniqueConstraint,
 
1625
            schema.ForeignKeyConstraint, schema.PrimaryKeyConstraint,
 
1626
            schema.UniqueConstraint,
1387
1627
            schema.CheckConstraint
1388
1628
        ]:
1389
1629
            evt(target)
1404
1644
        eq_(
1405
1645
            canary,
1406
1646
            [
1407
 
            'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t1',
1408
 
            'ForeignKeyConstraint->Table', 'ForeignKeyConstraint->t1',
1409
 
            'UniqueConstraint->Table', 'UniqueConstraint->t1',
1410
 
            'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t2',
1411
 
            'CheckConstraint->Table', 'CheckConstraint->t2',
1412
 
            'UniqueConstraint->Table', 'UniqueConstraint->t2'
 
1647
                'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t1',
 
1648
                'ForeignKeyConstraint->Table', 'ForeignKeyConstraint->t1',
 
1649
                'UniqueConstraint->Table', 'UniqueConstraint->t1',
 
1650
                'PrimaryKeyConstraint->Table', 'PrimaryKeyConstraint->t2',
 
1651
                'CheckConstraint->Table', 'CheckConstraint->t2',
 
1652
                'UniqueConstraint->Table', 'UniqueConstraint->t2'
1413
1653
            ]
1414
1654
        )
 
1655