← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/sql/test_types.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_types.py
1
1
# coding: utf-8
2
 
from test.lib.testing import eq_, assert_raises, assert_raises_message
 
2
from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
3
3
import decimal
4
 
import datetime, os, re
 
4
import datetime
 
5
import os
5
6
from sqlalchemy import *
6
 
from sqlalchemy import exc, types, util, schema, dialects
 
7
from sqlalchemy import exc, types, util, dialects
7
8
for name in dialects.__all__:
8
9
    __import__("sqlalchemy.dialects.%s" % name)
9
10
from sqlalchemy.sql import operators, column, table
10
 
from test.lib.testing import eq_
11
 
import sqlalchemy.engine.url as url
12
11
from sqlalchemy.engine import default
13
 
from test.lib.schema import Table, Column
14
 
from test.lib import *
15
 
from test.lib.util import picklers
16
 
from sqlalchemy.util.compat import decimal
17
 
from test.lib.util import round_decimal
18
 
from test.lib import fixtures
 
12
from sqlalchemy.testing.schema import Table, Column
 
13
from sqlalchemy import testing
 
14
from sqlalchemy.testing import AssertsCompiledSQL, AssertsExecutionResults, \
 
15
    engines, pickleable
 
16
from sqlalchemy.testing.util import picklers
 
17
from sqlalchemy.testing.util import round_decimal
 
18
from sqlalchemy.testing import fixtures
19
19
 
20
20
class AdaptTest(fixtures.TestBase):
21
21
    def _all_dialect_modules(self):
74
74
                (DATE, "DATE"),
75
75
                (TIME, ("TIME", "TIME WITHOUT TIME ZONE")),
76
76
                (CLOB, "CLOB"),
77
 
                (VARCHAR(10), ("VARCHAR(10)","VARCHAR(10 CHAR)")),
 
77
                (VARCHAR(10), ("VARCHAR(10)", "VARCHAR(10 CHAR)")),
78
78
                (NVARCHAR(10), ("NVARCHAR(10)", "NATIONAL VARCHAR(10)",
79
79
                                    "NVARCHAR2(10)")),
80
80
                (CHAR, "CHAR"),
122
122
                t1 = typ()
123
123
            for cls in [typ] + typ.__subclasses__():
124
124
                if not issubclass(typ, types.Enum) and \
125
 
                    issubclass(cls, types.Enum):
 
125
                        issubclass(cls, types.Enum):
126
126
                    continue
127
127
                t2 = t1.adapt(cls)
128
128
                assert t1 is not t2
242
242
                Column('Lar', LargeBinary()),
243
243
                Column('Pic', PickleType()),
244
244
                Column('Int', Interval()),
245
 
                Column('Enu', Enum('x','y','z', name="somename")),
 
245
                Column('Enu', Enum('x', 'y', 'z', name="somename")),
246
246
            ]
247
247
            for column_type in column_types:
248
 
                #print column_type
249
248
                meta = MetaData()
250
249
                Table('foo', meta, column_type)
251
 
                ct = loads(dumps(column_type))
252
 
                mt = loads(dumps(meta))
 
250
                loads(dumps(column_type))
 
251
                loads(dumps(meta))
253
252
 
254
253
 
255
254
class UserDefinedTest(fixtures.TablesTest, AssertsCompiledSQL):
284
283
    def test_typedecorator_impl(self):
285
284
        for impl_, exp, kw in [
286
285
            (Float, "FLOAT", {}),
287
 
            (Float, "FLOAT(2)", {'precision':2}),
288
 
            (Float(2), "FLOAT(2)", {'precision':4}),
 
286
            (Float, "FLOAT(2)", {'precision': 2}),
 
287
            (Float(2), "FLOAT(2)", {'precision': 4}),
289
288
            (Numeric(19, 2), "NUMERIC(19, 2)", {}),
290
289
        ]:
291
290
            for dialect_ in (dialects.postgresql, dialects.mssql, dialects.mysql):
303
302
                raw_dialect_impl = raw_impl.dialect_impl(dialect_)
304
303
                dec_dialect_impl = dec_type.dialect_impl(dialect_)
305
304
                eq_(dec_dialect_impl.__class__, MyType)
306
 
                eq_(raw_dialect_impl.__class__ , dec_dialect_impl.impl.__class__)
 
305
                eq_(raw_dialect_impl.__class__, dec_dialect_impl.impl.__class__)
307
306
 
308
307
                self.assert_compile(
309
308
                    MyType(**kw),
335
334
                Float().dialect_impl(pg).__class__
336
335
        )
337
336
 
 
337
    def test_type_decorator_repr(self):
 
338
        class MyType(TypeDecorator):
 
339
            impl = VARCHAR
 
340
 
 
341
        eq_(repr(MyType(45)), "MyType(length=45)")
 
342
 
338
343
    def test_user_defined_typedec_impl_bind(self):
339
344
        class TypeOne(types.TypeEngine):
340
345
            def bind_processor(self, dialect):
401
406
        t = Table('t', metadata, Column('data', String(50)))
402
407
        metadata.create_all()
403
408
 
404
 
        t.insert().values(data=type_coerce('d1BIND_OUT',MyType)).execute()
 
409
        t.insert().values(data=type_coerce('d1BIND_OUT', MyType)).execute()
405
410
 
406
411
        eq_(
407
412
            select([type_coerce(t.c.data, MyType)]).execute().fetchall(),
414
419
        )
415
420
 
416
421
        eq_(
 
422
            select([t.c.data, type_coerce(t.c.data, MyType)]).
 
423
                    alias().select().execute().fetchall(),
 
424
            [('d1', 'd1BIND_OUT')]
 
425
        )
 
426
 
 
427
        eq_(
417
428
            select([t.c.data, type_coerce(t.c.data, MyType)]).\
418
429
                        where(type_coerce(t.c.data, MyType) == 'd1BIND_OUT').\
419
430
                        execute().fetchall(),
441
452
            []
442
453
        )
443
454
 
 
455
        eq_(
 
456
            testing.db.scalar(
 
457
                select([type_coerce(literal('d1BIND_OUT'), MyType)])
 
458
            ),
 
459
            'd1BIND_OUT'
 
460
        )
 
461
 
444
462
    @classmethod
445
463
    def define_tables(cls, metadata):
446
464
        class MyType(types.UserDefinedType):
448
466
                return "VARCHAR(100)"
449
467
            def bind_processor(self, dialect):
450
468
                def process(value):
451
 
                    return "BIND_IN"+ value
 
469
                    return "BIND_IN" + value
452
470
                return process
453
471
            def result_processor(self, dialect, coltype):
454
472
                def process(value):
461
479
            impl = String
462
480
            def bind_processor(self, dialect):
463
481
                impl_processor = super(MyDecoratedType, self).bind_processor(dialect)\
464
 
                                        or (lambda value:value)
 
482
                                        or (lambda value: value)
465
483
                def process(value):
466
 
                    return "BIND_IN"+ impl_processor(value)
 
484
                    return "BIND_IN" + impl_processor(value)
467
485
                return process
468
486
            def result_processor(self, dialect, coltype):
469
487
                impl_processor = super(MyDecoratedType, self).result_processor(dialect, coltype)\
470
 
                                        or (lambda value:value)
 
488
                                        or (lambda value: value)
471
489
                def process(value):
472
490
                    return impl_processor(value) + "BIND_OUT"
473
491
                return process
510
528
 
511
529
            def bind_processor(self, dialect):
512
530
                impl_processor = super(MyUnicodeType, self).bind_processor(dialect)\
513
 
                                        or (lambda value:value)
 
531
                                        or (lambda value: value)
514
532
 
515
533
                def process(value):
516
 
                    return "BIND_IN"+ impl_processor(value)
 
534
                    return "BIND_IN" + impl_processor(value)
517
535
                return process
518
536
 
519
537
            def result_processor(self, dialect, coltype):
520
538
                impl_processor = super(MyUnicodeType, self).result_processor(dialect, coltype)\
521
 
                                        or (lambda value:value)
 
539
                                        or (lambda value: value)
522
540
                def process(value):
523
541
                    return impl_processor(value) + "BIND_OUT"
524
542
                return process
527
545
                return MyUnicodeType(self.impl.length)
528
546
 
529
547
        Table('users', metadata,
530
 
            Column('user_id', Integer, primary_key = True),
 
548
            Column('user_id', Integer, primary_key=True),
531
549
            # totall custom type
532
 
            Column('goofy', MyType, nullable = False),
 
550
            Column('goofy', MyType, nullable=False),
533
551
 
534
552
            # decorated type with an argument, so its a String
535
 
            Column('goofy2', MyDecoratedType(50), nullable = False),
 
553
            Column('goofy2', MyDecoratedType(50), nullable=False),
536
554
 
537
 
            Column('goofy4', MyUnicodeType(50), nullable = False),
538
 
            Column('goofy7', MyNewUnicodeType(50), nullable = False),
539
 
            Column('goofy8', MyNewIntType, nullable = False),
540
 
            Column('goofy9', MyNewIntSubClass, nullable = False),
 
555
            Column('goofy4', MyUnicodeType(50), nullable=False),
 
556
            Column('goofy7', MyNewUnicodeType(50), nullable=False),
 
557
            Column('goofy8', MyNewIntType, nullable=False),
 
558
            Column('goofy9', MyNewIntSubClass, nullable=False),
541
559
        )
542
560
 
543
561
class VariantTest(fixtures.TestBase, AssertsCompiledSQL):
645
663
            'fooUTWO'
646
664
        )
647
665
 
648
 
class UnicodeTest(fixtures.TestBase, AssertsExecutionResults):
649
 
    """tests the Unicode type.  also tests the TypeDecorator with instances in the types package."""
650
 
 
651
 
    @classmethod
652
 
    def setup_class(cls):
653
 
        global unicode_table, metadata
654
 
        metadata = MetaData(testing.db)
655
 
        unicode_table = Table('unicode_table', metadata,
656
 
            Column('id', Integer, Sequence('uni_id_seq', optional=True), primary_key=True),
657
 
            Column('unicode_varchar', Unicode(250)),
658
 
            Column('unicode_text', UnicodeText),
659
 
            )
660
 
        metadata.create_all()
661
 
 
662
 
    @classmethod
663
 
    def teardown_class(cls):
664
 
        metadata.drop_all()
665
 
 
666
 
    @engines.close_first
667
 
    def teardown(self):
668
 
        unicode_table.delete().execute()
 
666
class UnicodeTest(fixtures.TestBase):
 
667
    """Exercise the Unicode and related types.
 
668
 
 
669
    Note:  unicode round trip tests are now in
 
670
    sqlalchemy/testing/suite/test_types.py.
 
671
 
 
672
    """
669
673
 
670
674
    def test_native_unicode(self):
671
675
        """assert expected values for 'native unicode' mode"""
672
676
 
673
 
        if \
674
 
             (testing.against('mssql+pyodbc') and not testing.db.dialect.freetds):
675
 
            assert testing.db.dialect.returns_unicode_strings == 'conditional'
676
 
            return
677
 
 
678
 
        if testing.against('mssql+pymssql'):
679
 
            assert testing.db.dialect.returns_unicode_strings == ('charset' in testing.db.url.query)
680
 
            return
681
 
 
682
 
        assert testing.db.dialect.returns_unicode_strings == \
683
 
            ((testing.db.name, testing.db.driver) in \
684
 
            (
685
 
                ('postgresql','psycopg2'),
686
 
                ('postgresql','pypostgresql'),
687
 
                ('postgresql','pg8000'),
688
 
                ('postgresql','zxjdbc'),
689
 
                ('mysql','oursql'),
690
 
                ('mysql','zxjdbc'),
691
 
                ('mysql','mysqlconnector'),
692
 
                ('mysql','pymysql'),
693
 
                ('sqlite','pysqlite'),
694
 
                ('oracle','zxjdbc'),
695
 
                ('oracle','cx_oracle'),
696
 
            )), \
697
 
            "name: %s driver %s returns_unicode_strings=%s" % \
698
 
                                        (testing.db.name,
699
 
                                         testing.db.driver,
700
 
                                         testing.db.dialect.returns_unicode_strings)
701
 
 
702
 
    def test_round_trip(self):
703
 
        unicodedata = u"Alors vous imaginez ma surprise, au lever du jour, "\
704
 
                    u"quand une drôle de petite voix m’a réveillé. Elle "\
705
 
                    u"disait: « S’il vous plaît… dessine-moi un mouton! »"
706
 
 
707
 
        unicode_table.insert().execute(unicode_varchar=unicodedata,unicode_text=unicodedata)
708
 
 
709
 
        x = unicode_table.select().execute().first()
710
 
        assert isinstance(x['unicode_varchar'], unicode)
711
 
        assert isinstance(x['unicode_text'], unicode)
712
 
        eq_(x['unicode_varchar'], unicodedata)
713
 
        eq_(x['unicode_text'], unicodedata)
714
 
 
715
 
    def test_round_trip_executemany(self):
716
 
        # cx_oracle was producing different behavior for cursor.executemany()
717
 
        # vs. cursor.execute()
718
 
 
719
 
        unicodedata = u"Alors vous imaginez ma surprise, au lever du jour, quand "\
720
 
                        u"une drôle de petite voix m’a réveillé. "\
721
 
                        u"Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
722
 
 
723
 
        unicode_table.insert().execute(
724
 
                dict(unicode_varchar=unicodedata,unicode_text=unicodedata),
725
 
                dict(unicode_varchar=unicodedata,unicode_text=unicodedata)
726
 
        )
727
 
 
728
 
        x = unicode_table.select().execute().first()
729
 
        assert isinstance(x['unicode_varchar'], unicode)
730
 
        eq_(x['unicode_varchar'], unicodedata)
731
 
        assert isinstance(x['unicode_text'], unicode)
732
 
        eq_(x['unicode_text'], unicodedata)
733
 
 
734
 
    def test_union(self):
735
 
        """ensure compiler processing works for UNIONs"""
736
 
 
737
 
        unicodedata = u"Alors vous imaginez ma surprise, au lever du jour, quand "\
738
 
                        u"une drôle de petite voix m’a réveillé. "\
739
 
                        u"Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
740
 
 
741
 
        unicode_table.insert().execute(unicode_varchar=unicodedata,unicode_text=unicodedata)
742
 
 
743
 
        x = union(
744
 
                    select([unicode_table.c.unicode_varchar]),
745
 
                    select([unicode_table.c.unicode_varchar])
746
 
                ).execute().first()
747
 
 
748
 
        assert isinstance(x['unicode_varchar'], unicode)
749
 
        eq_(x['unicode_varchar'], unicodedata)
750
 
 
751
 
    @testing.fails_on('oracle', 'oracle converts empty strings to a blank space')
752
 
    def test_blank_strings(self):
753
 
        unicode_table.insert().execute(unicode_varchar=u'')
754
 
        assert select([unicode_table.c.unicode_varchar]).scalar() == u''
755
 
 
756
 
    def test_unicode_warnings(self):
757
 
        """test the warnings raised when SQLA must coerce unicode binds,
758
 
        *and* is using the Unicode type.
759
 
 
760
 
        """
761
 
 
762
 
        unicodedata = u"Alors vous imaginez ma surprise, au lever du jour, quand "\
763
 
                        u"une drôle de petite voix m’a réveillé. "\
764
 
                        u"Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
765
 
 
766
 
        # using Unicode explicly - warning should be emitted
767
 
        u = Unicode()
768
 
        uni = u.dialect_impl(testing.db.dialect).bind_processor(testing.db.dialect)
769
 
        if testing.db.dialect.supports_unicode_binds:
770
 
            # Py3K
771
 
            #assert_raises(exc.SAWarning, uni, b'x')
772
 
            #assert isinstance(uni(unicodedata), str)
773
 
            # Py2K
774
 
            assert_raises(exc.SAWarning, uni, 'x')
775
 
            assert isinstance(uni(unicodedata), unicode)
776
 
            # end Py2K
777
 
 
778
 
            eq_(uni(unicodedata), unicodedata)
 
677
        if (testing.against('mssql+pyodbc') and
 
678
                not testing.db.dialect.freetds) \
 
679
                or testing.against('mssql+mxodbc'):
 
680
            eq_(
 
681
                testing.db.dialect.returns_unicode_strings,
 
682
                'conditional'
 
683
            )
 
684
 
 
685
        elif testing.against('mssql+pymssql'):
 
686
            eq_(
 
687
                testing.db.dialect.returns_unicode_strings,
 
688
                ('charset' in testing.db.url.query)
 
689
            )
 
690
 
 
691
        elif testing.against('mysql+cymysql', 'mysql+pymssql'):
 
692
            eq_(
 
693
                testing.db.dialect.returns_unicode_strings,
 
694
                True if util.py3k else False
 
695
            )
 
696
 
 
697
 
779
698
        else:
780
 
            # Py3K
781
 
            #assert_raises(exc.SAWarning, uni, b'x')
782
 
            #assert isinstance(uni(unicodedata), bytes)
783
 
            # Py2K
784
 
            assert_raises(exc.SAWarning, uni, 'x')
785
 
            assert isinstance(uni(unicodedata), str)
786
 
            # end Py2K
787
 
 
788
 
            eq_(uni(unicodedata), unicodedata.encode('utf-8'))
789
 
 
790
 
        # using convert unicode at engine level -
791
 
        # this should not be raising a warning
792
 
        unicode_engine = engines.utf8_engine(options={'convert_unicode':True,})
793
 
        unicode_engine.dialect.supports_unicode_binds = False
 
699
            expected = (testing.db.name, testing.db.driver) in \
 
700
                (
 
701
                    ('postgresql', 'psycopg2'),
 
702
                    ('postgresql', 'pypostgresql'),
 
703
                    ('postgresql', 'pg8000'),
 
704
                    ('postgresql', 'zxjdbc'),
 
705
                    ('mysql', 'oursql'),
 
706
                    ('mysql', 'zxjdbc'),
 
707
                    ('mysql', 'mysqlconnector'),
 
708
                    ('sqlite', 'pysqlite'),
 
709
                    ('oracle', 'zxjdbc'),
 
710
                    ('oracle', 'cx_oracle'),
 
711
                )
 
712
 
 
713
            eq_(
 
714
                testing.db.dialect.returns_unicode_strings,
 
715
                expected
 
716
            )
 
717
 
 
718
    data = u"Alors vous imaginez ma surprise, au lever du jour, quand "\
 
719
            u"une drôle de petite voix m’a réveillé. "\
 
720
            u"Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
 
721
 
 
722
    def test_unicode_warnings_typelevel_native_unicode(self):
 
723
 
 
724
        unicodedata = self.data
 
725
        u = Unicode()
 
726
        dialect = default.DefaultDialect()
 
727
        dialect.supports_unicode_binds = True
 
728
        uni = u.dialect_impl(dialect).bind_processor(dialect)
 
729
        # Py3K
 
730
        #assert_raises(exc.SAWarning, uni, b'x')
 
731
        #assert isinstance(uni(unicodedata), str)
 
732
        # Py2K
 
733
        assert_raises(exc.SAWarning, uni, 'x')
 
734
        assert isinstance(uni(unicodedata), unicode)
 
735
        # end Py2K
 
736
 
 
737
    def test_unicode_warnings_typelevel_sqla_unicode(self):
 
738
        unicodedata = self.data
 
739
        u = Unicode()
 
740
        dialect = default.DefaultDialect()
 
741
        dialect.supports_unicode_binds = False
 
742
        uni = u.dialect_impl(dialect).bind_processor(dialect)
 
743
        # Py3K
 
744
        #assert_raises(exc.SAWarning, uni, b'x')
 
745
        #assert isinstance(uni(unicodedata), bytes)
 
746
        # Py2K
 
747
        assert_raises(exc.SAWarning, uni, 'x')
 
748
        assert isinstance(uni(unicodedata), str)
 
749
        # end Py2K
 
750
 
 
751
        eq_(uni(unicodedata), unicodedata.encode('utf-8'))
 
752
 
 
753
    def test_unicode_warnings_dialectlevel(self):
 
754
 
 
755
        unicodedata = self.data
 
756
 
 
757
        dialect = default.DefaultDialect(convert_unicode=True)
 
758
        dialect.supports_unicode_binds = False
794
759
 
795
760
        s = String()
796
 
        uni = s.dialect_impl(unicode_engine.dialect).bind_processor(unicode_engine.dialect)
 
761
        uni = s.dialect_impl(dialect).bind_processor(dialect)
797
762
        # this is not the unicode type - no warning
798
763
        # Py3K
799
764
        #uni(b'x')
805
770
 
806
771
        eq_(uni(unicodedata), unicodedata.encode('utf-8'))
807
772
 
808
 
    # Py3K
809
 
    #@testing.fails_if(
810
 
    #                    lambda: testing.db_spec("postgresql+pg8000")(testing.db),
811
 
    #                    "pg8000 appropriately does not accept 'bytes' for a VARCHAR column."
812
 
    #                    )
813
773
    def test_ignoring_unicode_error(self):
814
 
        """checks String(unicode_error='ignore') is passed to underlying codec."""
815
 
 
816
 
        unicodedata = u"Alors vous imaginez ma surprise, au lever du jour, quand "\
817
 
                        u"une drôle de petite voix m’a réveillé. "\
818
 
                        u"Elle disait: « S’il vous plaît… dessine-moi un mouton! »"
819
 
 
820
 
        asciidata = unicodedata.encode('ascii', 'ignore')
821
 
 
822
 
        m = MetaData()
823
 
        table = Table('unicode_err_table', m,
824
 
            Column('sort', Integer),
825
 
            Column('plain_varchar_no_coding_error', \
826
 
                    String(248, convert_unicode='force', unicode_error='ignore'))
827
 
            )
828
 
 
829
 
        m2 = MetaData()
830
 
        utf8_table = Table('unicode_err_table', m2,
831
 
            Column('sort', Integer),
832
 
            Column('plain_varchar_no_coding_error', \
833
 
                    String(248, convert_unicode=True))
834
 
            )
835
 
 
836
 
        engine = engines.testing_engine(options={'encoding':'ascii'})
837
 
        m.create_all(engine)
838
 
        try:
839
 
            # insert a row that should be ascii and
840
 
            # coerce from unicode with ignore on the bind side
841
 
            engine.execute(
842
 
                table.insert(),
843
 
                sort=1,
844
 
                plain_varchar_no_coding_error=unicodedata
845
 
            )
846
 
 
847
 
            # switch to utf-8
848
 
            engine.dialect.encoding = 'utf-8'
849
 
            from binascii import hexlify
850
 
 
851
 
            # the row that we put in was stored as hexlified ascii
852
 
            row = engine.execute(utf8_table.select()).first()
853
 
            x = row['plain_varchar_no_coding_error']
854
 
            connect_opts = engine.dialect.create_connect_args(testing.db.url)[1]
855
 
            if isinstance(x, unicode):
856
 
                x = x.encode('utf-8')
857
 
            a = hexlify(x)
858
 
            b = hexlify(asciidata)
859
 
            eq_(a, b)
860
 
 
861
 
            # insert another row which will be stored with
862
 
            # utf-8 only chars
863
 
            engine.execute(
864
 
                utf8_table.insert(),
865
 
                sort=2,
866
 
                plain_varchar_no_coding_error=unicodedata
867
 
            )
868
 
 
869
 
            # switch back to ascii
870
 
            engine.dialect.encoding = 'ascii'
871
 
 
872
 
            # one row will be ascii with ignores,
873
 
            # the other will be either ascii with the ignores
874
 
            # or just the straight unicode+ utf8 value if the
875
 
            # dialect just returns unicode
876
 
            result = engine.execute(table.select().order_by(table.c.sort))
877
 
            ascii_row = result.fetchone()
878
 
            utf8_row = result.fetchone()
879
 
            result.close()
880
 
 
881
 
            x = ascii_row['plain_varchar_no_coding_error']
882
 
            # on python3 "x" comes back as string (i.e. unicode),
883
 
            # hexlify requires bytes
884
 
            a = hexlify(x.encode('utf-8'))
885
 
            b = hexlify(asciidata)
886
 
            eq_(a, b)
887
 
 
888
 
            x = utf8_row['plain_varchar_no_coding_error']
889
 
            if testing.against('mssql+pyodbc') and not testing.db.dialect.freetds:
890
 
                # TODO: no clue what this is
891
 
                eq_(
892
 
                      x,
893
 
                      u'Alors vous imaginez ma surprise, au lever du jour, quand une '
894
 
                      u'drle de petite voix ma rveill. Elle disait:  Sil vous plat '
895
 
                      u'dessine-moi un mouton! '
896
 
                )
897
 
            elif engine.dialect.returns_unicode_strings:
898
 
                eq_(x, unicodedata)
899
 
            else:
900
 
                a = hexlify(x)
901
 
                eq_(a, b)
902
 
 
903
 
        finally:
904
 
            m.drop_all(engine)
 
774
        """checks String(unicode_error='ignore') is passed to
 
775
        underlying codec."""
 
776
 
 
777
        unicodedata = self.data
 
778
 
 
779
        type_ = String(248, convert_unicode='force', unicode_error='ignore')
 
780
        dialect = default.DefaultDialect(encoding='ascii')
 
781
        proc = type_.result_processor(dialect, 10)
 
782
 
 
783
        utfdata = unicodedata.encode('utf8')
 
784
        eq_(
 
785
            proc(utfdata),
 
786
            unicodedata.encode('ascii', 'ignore').decode()
 
787
        )
905
788
 
906
789
 
907
790
class EnumTest(fixtures.TestBase):
911
794
        metadata = MetaData(testing.db)
912
795
        enum_table = Table('enum_table', metadata,
913
796
            Column("id", Integer, primary_key=True),
914
 
            Column('someenum', Enum('one','two','three', name='myenum'))
 
797
            Column('someenum', Enum('one', 'two', 'three', name='myenum'))
915
798
        )
916
799
 
917
800
        non_native_enum_table = Table('non_native_enum_table', metadata,
918
801
            Column("id", Integer, primary_key=True),
919
 
            Column('someenum', Enum('one','two','three', native_enum=False)),
 
802
            Column('someenum', Enum('one', 'two', 'three', native_enum=False)),
920
803
        )
921
804
 
922
805
        metadata.create_all()
937
820
                        'but expression is of type text')
938
821
    def test_round_trip(self):
939
822
        enum_table.insert().execute([
940
 
            {'id':1, 'someenum':'two'},
941
 
            {'id':2, 'someenum':'two'},
942
 
            {'id':3, 'someenum':'one'},
 
823
            {'id': 1, 'someenum': 'two'},
 
824
            {'id': 2, 'someenum': 'two'},
 
825
            {'id': 3, 'someenum': 'one'},
943
826
        ])
944
827
 
945
828
        eq_(
953
836
 
954
837
    def test_non_native_round_trip(self):
955
838
        non_native_enum_table.insert().execute([
956
 
            {'id':1, 'someenum':'two'},
957
 
            {'id':2, 'someenum':'two'},
958
 
            {'id':3, 'someenum':'one'},
 
839
            {'id': 1, 'someenum': 'two'},
 
840
            {'id': 2, 'someenum': 'two'},
 
841
            {'id': 3, 'someenum': 'one'},
959
842
        ])
960
843
 
961
844
        eq_(
970
853
 
971
854
    def test_adapt(self):
972
855
        from sqlalchemy.dialects.postgresql import ENUM
973
 
        e1 = Enum('one','two','three', native_enum=False)
 
856
        e1 = Enum('one', 'two', 'three', native_enum=False)
974
857
        eq_(e1.adapt(ENUM).native_enum, False)
975
 
        e1 = Enum('one','two','three', native_enum=True)
 
858
        e1 = Enum('one', 'two', 'three', native_enum=True)
976
859
        eq_(e1.adapt(ENUM).native_enum, True)
977
 
        e1 = Enum('one','two','three', name='foo', schema='bar')
 
860
        e1 = Enum('one', 'two', 'three', name='foo', schema='bar')
978
861
        eq_(e1.adapt(ENUM).name, 'foo')
979
862
        eq_(e1.adapt(ENUM).schema, 'bar')
980
863
 
984
867
    def test_constraint(self):
985
868
        assert_raises(exc.DBAPIError,
986
869
            enum_table.insert().execute,
987
 
            {'id':4, 'someenum':'four'}
 
870
            {'id': 4, 'someenum': 'four'}
988
871
        )
989
872
 
990
873
    @testing.fails_on('mysql',
992
875
    def test_non_native_constraint(self):
993
876
        assert_raises(exc.DBAPIError,
994
877
            non_native_enum_table.insert().execute,
995
 
            {'id':4, 'someenum':'four'}
 
878
            {'id': 4, 'someenum': 'four'}
996
879
        )
997
880
 
998
881
    def test_mock_engine_no_prob(self):
1012
895
class BinaryTest(fixtures.TestBase, AssertsExecutionResults):
1013
896
    __excluded_on__ = (
1014
897
        ('mysql', '<', (4, 1, 1)),  # screwy varbinary types
1015
 
        )
 
898
    )
1016
899
 
1017
900
    @classmethod
1018
901
    def setup_class(cls):
1055
938
        testobj2 = pickleable.Foo('im foo 2')
1056
939
        testobj3 = pickleable.Foo('im foo 3')
1057
940
 
1058
 
        stream1 =self.load_stream('binary_data_one.dat')
1059
 
        stream2 =self.load_stream('binary_data_two.dat')
 
941
        stream1 = self.load_stream('binary_data_one.dat')
 
942
        stream2 = self.load_stream('binary_data_two.dat')
1060
943
        binary_table.insert().execute(
1061
944
                            primary_id=1,
1062
945
                            misc='binary_data_one.dat',
1081
964
            binary_table.select(order_by=binary_table.c.primary_id),
1082
965
            text(
1083
966
                "select * from binary_table order by binary_table.primary_id",
1084
 
                typemap={'pickled':PickleType,
1085
 
                        'mypickle':MyPickleType,
1086
 
                        'data':LargeBinary, 'data_slice':LargeBinary},
 
967
                typemap={'pickled': PickleType,
 
968
                        'mypickle': MyPickleType,
 
969
                        'data': LargeBinary, 'data_slice': LargeBinary},
1087
970
                bind=testing.db)
1088
971
        ):
1089
972
            l = stmt.execute().fetchall()
1095
978
            eq_(testobj3.moredata, l[0]['mypickle'].moredata)
1096
979
            eq_(l[0]['mypickle'].stuff, 'this is the right stuff')
1097
980
 
1098
 
    @testing.fails_on('oracle+cx_oracle', 'oracle fairly grumpy about binary '
1099
 
                                        'data, not really known how to make this work')
 
981
    @testing.requires.binary_comparisons
1100
982
    def test_comparison(self):
1101
983
        """test that type coercion occurs on comparison for binary"""
1102
984
 
1105
987
 
1106
988
        data = os.urandom(32)
1107
989
        binary_table.insert().execute(data=data)
1108
 
        eq_(binary_table.select().where(binary_table.c.data==data).alias().count().scalar(), 1)
 
990
        eq_(binary_table.select().
 
991
                    where(binary_table.c.data == data).alias().
 
992
                    count().scalar(), 1)
1109
993
 
1110
994
 
1111
995
    def load_stream(self, name):
1131
1015
                    return value / 10
1132
1016
                return process
1133
1017
            def adapt_operator(self, op):
1134
 
                return {operators.add:operators.sub, operators.sub:operators.add}.get(op, op)
 
1018
                return {operators.add: operators.sub,
 
1019
                    operators.sub: operators.add}.get(op, op)
1135
1020
 
1136
1021
        class MyTypeDec(types.TypeDecorator):
1137
1022
            impl = String
1154
1039
        meta.create_all()
1155
1040
 
1156
1041
        test_table.insert().execute({
1157
 
                                'id':1,
1158
 
                                'data':'somedata',
1159
 
                                'atimestamp':datetime.date(2007, 10, 15),
1160
 
                                'avalue':25, 'bvalue':'foo'})
 
1042
                                'id': 1,
 
1043
                                'data': 'somedata',
 
1044
                                'atimestamp': datetime.date(2007, 10, 15),
 
1045
                                'avalue': 25, 'bvalue': 'foo'})
1161
1046
 
1162
1047
    @classmethod
1163
1048
    def teardown_class(cls):
1181
1066
            testing.db.execute(
1182
1067
                    select([test_table.c.id, test_table.c.data, test_table.c.atimestamp])
1183
1068
                    .where(expr),
1184
 
                    {"thedate":datetime.date(2007, 10, 15)}).fetchall(),
 
1069
                    {"thedate": datetime.date(2007, 10, 15)}).fetchall(),
1185
1070
            [(1, 'somedata', datetime.date(2007, 10, 15))]
1186
1071
        )
1187
1072
 
1200
1085
 
1201
1086
        eq_(
1202
1087
            testing.db.execute(test_table.select().where(expr),
1203
 
                {"somevalue":"foo"}).fetchall(),
 
1088
                {"somevalue": "foo"}).fetchall(),
1204
1089
            [(1, 'somedata',
1205
1090
                datetime.date(2007, 10, 15), 25, 'BIND_INfooBIND_OUT')]
1206
1091
        )
1221
1106
        expr = column('foo', CHAR) == "asdf"
1222
1107
        eq_(expr.right.type.__class__, CHAR)
1223
1108
 
1224
 
 
 
1109
    @testing.uses_deprecated
1225
1110
    @testing.fails_on('firebird', 'Data type unknown on the parameter')
1226
1111
    @testing.fails_on('mssql', 'int is unsigned ?  not clear')
1227
1112
    def test_operator_adapt(self):
1263
1148
            "BIND_INfooBIND_INhiBIND_OUT"
1264
1149
        )
1265
1150
 
 
1151
    def test_typedec_is_adapt(self):
 
1152
        class CoerceNothing(TypeDecorator):
 
1153
            coerce_to_is_types = ()
 
1154
            impl = Integer
 
1155
        class CoerceBool(TypeDecorator):
 
1156
            coerce_to_is_types = (bool, )
 
1157
            impl = Boolean
 
1158
        class CoerceNone(TypeDecorator):
 
1159
            coerce_to_is_types = (type(None),)
 
1160
            impl = Integer
 
1161
 
 
1162
        c1 = column('x', CoerceNothing())
 
1163
        c2 = column('x', CoerceBool())
 
1164
        c3 = column('x', CoerceNone())
 
1165
 
 
1166
        self.assert_compile(
 
1167
            and_(c1 == None, c2 == None, c3 == None),
 
1168
            "x = :x_1 AND x = :x_2 AND x IS NULL"
 
1169
        )
 
1170
        self.assert_compile(
 
1171
            and_(c1 == True, c2 == True, c3 == True),
 
1172
            "x = :x_1 AND x = true AND x = :x_2"
 
1173
        )
 
1174
        self.assert_compile(
 
1175
            and_(c1 == 3, c2 == 3, c3 == 3),
 
1176
            "x = :x_1 AND x = :x_2 AND x = :x_3"
 
1177
        )
 
1178
        self.assert_compile(
 
1179
            and_(c1.is_(True), c2.is_(True), c3.is_(True)),
 
1180
            "x IS :x_1 AND x IS true AND x IS :x_2"
 
1181
        )
 
1182
 
 
1183
 
1266
1184
    def test_typedec_righthand_coercion(self):
1267
1185
        class MyTypeDec(types.TypeDecorator):
1268
1186
            impl = String
1288
1206
            "BIND_INfooBIND_IN6BIND_OUT"
1289
1207
        )
1290
1208
 
1291
 
 
1292
1209
    def test_bind_typing(self):
1293
1210
        from sqlalchemy.sql import column
1294
1211
 
1299
1216
            pass
1300
1217
 
1301
1218
        # unknown type + integer, right hand bind
1302
 
        # is an Integer
 
1219
        # coerces to given type
1303
1220
        expr = column("foo", MyFoobarType) + 5
1304
 
        assert expr.right.type._type_affinity is types.Integer
 
1221
        assert expr.right.type._type_affinity is MyFoobarType
1305
1222
 
1306
1223
        # untyped bind - it gets assigned MyFoobarType
1307
1224
        expr = column("foo", MyFoobarType) + bindparam("foo")
1320
1237
        assert expr.right.type._type_affinity is MyFoobarType
1321
1238
 
1322
1239
        expr = column("foo", MyFoobarType) - datetime.date(2010, 8, 25)
1323
 
        assert expr.right.type._type_affinity is types.Date
 
1240
        assert expr.right.type._type_affinity is MyFoobarType
1324
1241
 
1325
1242
    def test_date_coercion(self):
1326
1243
        from sqlalchemy.sql import column
1381
1298
        assert test_table.c.data.distinct().type == test_table.c.data.type
1382
1299
 
1383
1300
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
1384
 
    def test_default_compile(self):
1385
 
        """test that the base dialect of the type object is used
1386
 
        for default compilation.
1387
 
 
1388
 
        """
1389
 
 
1390
 
        for type_, expected in (
1391
 
            (String(), "VARCHAR"),
1392
 
            (Integer(), "INTEGER"),
1393
 
            (dialects.postgresql.INET(), "INET"),
1394
 
            (dialects.postgresql.FLOAT(), "FLOAT"),
1395
 
            (dialects.mysql.REAL(precision=8, scale=2), "REAL(8, 2)"),
1396
 
            (dialects.postgresql.REAL(), "REAL"),
1397
 
            (INTEGER(), "INTEGER"),
1398
 
            (dialects.mysql.INTEGER(display_width=5), "INTEGER(5)")
1399
 
        ):
1400
 
            self.assert_compile(type_, expected,
1401
 
                                allow_dialect_select=True)
1402
 
 
1403
 
class DateTest(fixtures.TestBase, AssertsExecutionResults):
1404
 
    @classmethod
1405
 
    def setup_class(cls):
1406
 
        global users_with_date, insert_data
1407
 
 
1408
 
        db = testing.db
1409
 
        if testing.against('oracle'):
1410
 
            insert_data =  [
1411
 
                    (7, 'jack',
1412
 
                     datetime.datetime(2005, 11, 10, 0, 0),
1413
 
                     datetime.date(2005,11,10),
1414
 
                     datetime.datetime(2005, 11, 10, 0, 0, 0, 29384)),
1415
 
                    (8, 'roy',
1416
 
                     datetime.datetime(2005, 11, 10, 11, 52, 35),
1417
 
                     datetime.date(2005,10,10),
1418
 
                     datetime.datetime(2006, 5, 10, 15, 32, 47, 6754)),
1419
 
                    (9, 'foo',
1420
 
                     datetime.datetime(2006, 11, 10, 11, 52, 35),
1421
 
                     datetime.date(1970,4,1),
1422
 
                     datetime.datetime(2004, 9, 18, 4, 0, 52, 1043)),
1423
 
                    (10, 'colber', None, None, None),
1424
 
             ]
1425
 
            fnames = ['user_id', 'user_name', 'user_datetime',
1426
 
                      'user_date', 'user_time']
1427
 
 
1428
 
            collist = [Column('user_id', INT, primary_key=True),
1429
 
                       Column('user_name', VARCHAR(20)),
1430
 
                       Column('user_datetime', DateTime),
1431
 
                       Column('user_date', Date),
1432
 
                       Column('user_time', TIMESTAMP)]
1433
 
        else:
1434
 
            datetime_micro = 54839
1435
 
            time_micro = 999
1436
 
 
1437
 
            # Missing or poor microsecond support:
1438
 
            if testing.against('mssql', 'mysql', 'firebird', '+zxjdbc'):
1439
 
                datetime_micro, time_micro = 0, 0
1440
 
            # No microseconds for TIME
1441
 
            elif testing.against('maxdb'):
1442
 
                time_micro = 0
1443
 
 
1444
 
            insert_data =  [
1445
 
                (7, 'jack',
1446
 
                 datetime.datetime(2005, 11, 10, 0, 0),
1447
 
                 datetime.date(2005, 11, 10),
1448
 
                 datetime.time(12, 20, 2)),
1449
 
                (8, 'roy',
1450
 
                 datetime.datetime(2005, 11, 10, 11, 52, 35),
1451
 
                 datetime.date(2005, 10, 10),
1452
 
                 datetime.time(0, 0, 0)),
1453
 
                (9, 'foo',
1454
 
                 datetime.datetime(2005, 11, 10, 11, 52, 35, datetime_micro),
1455
 
                 datetime.date(1970, 4, 1),
1456
 
                 datetime.time(23, 59, 59, time_micro)),
1457
 
                (10, 'colber', None, None, None),
1458
 
            ]
1459
 
 
1460
 
 
1461
 
            fnames = ['user_id', 'user_name', 'user_datetime',
1462
 
                      'user_date', 'user_time']
1463
 
 
1464
 
            collist = [Column('user_id', INT, primary_key=True),
1465
 
                       Column('user_name', VARCHAR(20)),
1466
 
                       Column('user_datetime', DateTime(timezone=False)),
1467
 
                       Column('user_date', Date),
1468
 
                       Column('user_time', Time)]
1469
 
 
1470
 
        if testing.against('sqlite', 'postgresql'):
1471
 
            insert_data.append(
1472
 
                (11, 'historic',
1473
 
                datetime.datetime(1850, 11, 10, 11, 52, 35, datetime_micro),
1474
 
                datetime.date(1727,4,1),
1475
 
                None),
1476
 
            )
1477
 
 
1478
 
        users_with_date = Table('query_users_with_date',
1479
 
                                MetaData(testing.db), *collist)
1480
 
        users_with_date.create()
1481
 
        insert_dicts = [dict(zip(fnames, d)) for d in insert_data]
1482
 
 
1483
 
        for idict in insert_dicts:
1484
 
            users_with_date.insert().execute(**idict)
1485
 
 
1486
 
    @classmethod
1487
 
    def teardown_class(cls):
1488
 
        users_with_date.drop()
1489
 
 
1490
 
    def testdate(self):
1491
 
        global insert_data
1492
 
 
1493
 
        l = map(tuple,
1494
 
                users_with_date.select().order_by(users_with_date.c.user_id).execute().fetchall())
1495
 
        self.assert_(l == insert_data,
1496
 
                     'DateTest mismatch: got:%s expected:%s' % (l, insert_data))
1497
 
 
1498
 
    def testtextdate(self):
1499
 
        x = testing.db.execute(text(
1500
 
            "select user_datetime from query_users_with_date",
1501
 
            typemap={'user_datetime':DateTime})).fetchall()
1502
 
 
1503
 
        self.assert_(isinstance(x[0][0], datetime.datetime))
1504
 
 
1505
 
        x = testing.db.execute(text(
1506
 
            "select * from query_users_with_date where user_datetime=:somedate",
1507
 
            bindparams=[bindparam('somedate', type_=types.DateTime)]),
1508
 
            somedate=datetime.datetime(2005, 11, 10, 11, 52, 35)).fetchall()
1509
 
 
1510
 
    def testdate2(self):
1511
 
        meta = MetaData(testing.db)
1512
 
        t = Table('testdate', meta,
1513
 
                  Column('id', Integer,
1514
 
                         Sequence('datetest_id_seq', optional=True),
1515
 
                         primary_key=True),
1516
 
                Column('adate', Date), Column('adatetime', DateTime))
1517
 
        t.create(checkfirst=True)
1518
 
        try:
1519
 
            d1 = datetime.date(2007, 10, 30)
1520
 
            t.insert().execute(adate=d1, adatetime=d1)
1521
 
            d2 = datetime.datetime(2007, 10, 30)
1522
 
            t.insert().execute(adate=d2, adatetime=d2)
1523
 
 
1524
 
            x = t.select().execute().fetchall()[0]
1525
 
            eq_(x.adate.__class__, datetime.date)
1526
 
            eq_(x.adatetime.__class__, datetime.datetime)
1527
 
 
1528
 
            t.delete().execute()
1529
 
 
1530
 
            # test mismatched date/datetime
1531
 
            t.insert().execute(adate=d2, adatetime=d2)
1532
 
            eq_(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
1533
 
            eq_(select([t.c.adate, t.c.adatetime], t.c.adate==d1).execute().fetchall(), [(d1, d2)])
1534
 
 
1535
 
        finally:
1536
 
            t.drop(checkfirst=True)
1537
 
 
1538
 
class StringTest(fixtures.TestBase):
 
1301
    __dialect__ = 'default'
1539
1302
 
1540
1303
    @testing.requires.unbounded_varchar
1541
 
    def test_nolength_string(self):
1542
 
        metadata = MetaData(testing.db)
1543
 
        foo = Table('foo', metadata, Column('one', String))
1544
 
 
1545
 
        foo.create()
1546
 
        foo.drop()
1547
 
 
1548
 
class NumericTest(fixtures.TestBase):
1549
 
    def setup(self):
1550
 
        global metadata
1551
 
        metadata = MetaData(testing.db)
1552
 
 
1553
 
    def teardown(self):
1554
 
        metadata.drop_all()
1555
 
 
1556
 
    @testing.emits_warning(r".*does \*not\* support Decimal objects natively")
1557
 
    def _do_test(self, type_, input_, output, filter_=None, check_scale=False):
1558
 
        t = Table('t', metadata, Column('x', type_))
1559
 
        t.create()
1560
 
        t.insert().execute([{'x':x} for x in input_])
1561
 
 
1562
 
        result = set([row[0] for row in t.select().execute()])
1563
 
        output = set(output)
1564
 
        if filter_:
1565
 
            result = set(filter_(x) for x in result)
1566
 
            output = set(filter_(x) for x in output)
1567
 
        #print result
1568
 
        #print output
1569
 
        eq_(result, output)
1570
 
        if check_scale:
1571
 
            eq_(
1572
 
                [str(x) for x in result],
1573
 
                [str(x) for x in output],
1574
 
            )
1575
 
 
1576
 
    def test_numeric_as_decimal(self):
1577
 
        self._do_test(
1578
 
            Numeric(precision=8, scale=4),
1579
 
            [15.7563, decimal.Decimal("15.7563"), None],
1580
 
            [decimal.Decimal("15.7563"), None],
1581
 
        )
1582
 
 
1583
 
    def test_numeric_as_float(self):
1584
 
        if testing.against("oracle+cx_oracle"):
1585
 
            filter_ = lambda n:n is not None and round(n, 5) or None
1586
 
        else:
1587
 
            filter_ = None
1588
 
 
1589
 
        self._do_test(
1590
 
            Numeric(precision=8, scale=4, asdecimal=False),
1591
 
            [15.7563, decimal.Decimal("15.7563"), None],
1592
 
            [15.7563, None],
1593
 
            filter_ = filter_
1594
 
        )
1595
 
 
1596
 
    def test_float_as_decimal(self):
1597
 
        self._do_test(
1598
 
            Float(precision=8, asdecimal=True),
1599
 
            [15.7563, decimal.Decimal("15.7563"), None],
1600
 
            [decimal.Decimal("15.7563"), None],
1601
 
            filter_ = lambda n:n is not None and round(n, 5) or None
1602
 
        )
1603
 
 
1604
 
    def test_float_as_float(self):
1605
 
        self._do_test(
1606
 
            Float(precision=8),
1607
 
            [15.7563, decimal.Decimal("15.7563")],
1608
 
            [15.7563],
1609
 
            filter_ = lambda n:n is not None and round(n, 5) or None
1610
 
        )
1611
 
 
1612
 
    @testing.fails_on('mssql+pymssql', 'FIXME: improve pymssql dec handling')
1613
 
    def test_precision_decimal(self):
1614
 
        numbers = set([
1615
 
            decimal.Decimal("54.234246451650"),
1616
 
            decimal.Decimal("0.004354"),
1617
 
            decimal.Decimal("900.0"),
1618
 
        ])
1619
 
 
1620
 
        self._do_test(
1621
 
            Numeric(precision=18, scale=12),
1622
 
            numbers,
1623
 
            numbers,
1624
 
        )
1625
 
 
1626
 
    @testing.fails_on('mssql+pymssql', 'FIXME: improve pymssql dec handling')
1627
 
    def test_enotation_decimal(self):
1628
 
        """test exceedingly small decimals.
1629
 
 
1630
 
        Decimal reports values with E notation when the exponent
1631
 
        is greater than 6.
1632
 
 
1633
 
        """
1634
 
 
1635
 
        numbers = set([
1636
 
            decimal.Decimal('1E-2'),
1637
 
            decimal.Decimal('1E-3'),
1638
 
            decimal.Decimal('1E-4'),
1639
 
            decimal.Decimal('1E-5'),
1640
 
            decimal.Decimal('1E-6'),
1641
 
            decimal.Decimal('1E-7'),
1642
 
            decimal.Decimal('1E-8'),
1643
 
            decimal.Decimal("0.01000005940696"),
1644
 
            decimal.Decimal("0.00000005940696"),
1645
 
            decimal.Decimal("0.00000000000696"),
1646
 
            decimal.Decimal("0.70000000000696"),
1647
 
            decimal.Decimal("696E-12"),
1648
 
        ])
1649
 
        self._do_test(
1650
 
            Numeric(precision=18, scale=14),
1651
 
            numbers,
1652
 
            numbers
1653
 
        )
1654
 
 
1655
 
    @testing.fails_on("sybase+pyodbc",
1656
 
                        "Don't know how do get these values through FreeTDS + Sybase")
1657
 
    @testing.fails_on("firebird", "Precision must be from 1 to 18")
1658
 
    def test_enotation_decimal_large(self):
1659
 
        """test exceedingly large decimals.
1660
 
 
1661
 
        """
1662
 
 
1663
 
        numbers = set([
1664
 
            decimal.Decimal('4E+8'),
1665
 
            decimal.Decimal("5748E+15"),
1666
 
            decimal.Decimal('1.521E+15'),
1667
 
            decimal.Decimal('00000000000000.1E+12'),
1668
 
        ])
1669
 
        self._do_test(
1670
 
            Numeric(precision=25, scale=2),
1671
 
            numbers,
1672
 
            numbers
1673
 
        )
1674
 
 
1675
 
    @testing.fails_on('sqlite', 'TODO')
1676
 
    @testing.fails_on("firebird", "Precision must be from 1 to 18")
1677
 
    @testing.fails_on("sybase+pysybase", "TODO")
1678
 
    @testing.fails_on('mssql+pymssql', 'FIXME: improve pymssql dec handling')
1679
 
    def test_many_significant_digits(self):
1680
 
        numbers = set([
1681
 
            decimal.Decimal("31943874831932418390.01"),
1682
 
            decimal.Decimal("319438950232418390.273596"),
1683
 
            decimal.Decimal("87673.594069654243"),
1684
 
        ])
1685
 
        self._do_test(
1686
 
            Numeric(precision=38, scale=12),
1687
 
            numbers,
1688
 
            numbers
1689
 
        )
1690
 
 
1691
 
    @testing.fails_on('oracle+cx_oracle',
1692
 
        "this may be a bug due to the difficulty in handling "
1693
 
        "oracle precision numerics"
1694
 
    )
1695
 
    @testing.fails_on('postgresql+pg8000',
1696
 
        "pg-8000 does native decimal but truncates the decimals.")
1697
 
    def test_numeric_no_decimal(self):
1698
 
        numbers = set([
1699
 
            decimal.Decimal("1.000")
1700
 
        ])
1701
 
        self._do_test(
1702
 
            Numeric(precision=5, scale=3),
1703
 
            numbers,
1704
 
            numbers,
1705
 
            check_scale=True
1706
 
        )
 
1304
    def test_string_plain(self):
 
1305
        self.assert_compile(String(), "VARCHAR")
 
1306
 
 
1307
    def test_string_length(self):
 
1308
        self.assert_compile(String(50), "VARCHAR(50)")
 
1309
 
 
1310
    def test_string_collation(self):
 
1311
        self.assert_compile(String(50, collation="FOO"),
 
1312
                'VARCHAR(50) COLLATE "FOO"')
 
1313
 
 
1314
    def test_char_plain(self):
 
1315
        self.assert_compile(CHAR(), "CHAR")
 
1316
 
 
1317
    def test_char_length(self):
 
1318
        self.assert_compile(CHAR(50), "CHAR(50)")
 
1319
 
 
1320
    def test_char_collation(self):
 
1321
        self.assert_compile(CHAR(50, collation="FOO"),
 
1322
                'CHAR(50) COLLATE "FOO"')
 
1323
 
 
1324
    def test_text_plain(self):
 
1325
        self.assert_compile(Text(), "TEXT")
 
1326
 
 
1327
    def test_text_length(self):
 
1328
        self.assert_compile(Text(50), "TEXT(50)")
 
1329
 
 
1330
    def test_text_collation(self):
 
1331
        self.assert_compile(Text(collation="FOO"),
 
1332
                'TEXT COLLATE "FOO"')
 
1333
 
 
1334
    def test_default_compile_pg_inet(self):
 
1335
        self.assert_compile(dialects.postgresql.INET(), "INET",
 
1336
                allow_dialect_select=True)
 
1337
 
 
1338
    def test_default_compile_pg_float(self):
 
1339
        self.assert_compile(dialects.postgresql.FLOAT(), "FLOAT",
 
1340
                allow_dialect_select=True)
 
1341
 
 
1342
    def test_default_compile_mysql_integer(self):
 
1343
        self.assert_compile(
 
1344
                dialects.mysql.INTEGER(display_width=5), "INTEGER(5)",
 
1345
                allow_dialect_select=True)
 
1346
 
 
1347
    def test_numeric_plain(self):
 
1348
        self.assert_compile(types.NUMERIC(), 'NUMERIC')
 
1349
 
 
1350
    def test_numeric_precision(self):
 
1351
        self.assert_compile(types.NUMERIC(2), 'NUMERIC(2)')
 
1352
 
 
1353
    def test_numeric_scale(self):
 
1354
        self.assert_compile(types.NUMERIC(2, 4), 'NUMERIC(2, 4)')
 
1355
 
 
1356
    def test_decimal_plain(self):
 
1357
        self.assert_compile(types.DECIMAL(), 'DECIMAL')
 
1358
 
 
1359
    def test_decimal_precision(self):
 
1360
        self.assert_compile(types.DECIMAL(2), 'DECIMAL(2)')
 
1361
 
 
1362
    def test_decimal_scale(self):
 
1363
        self.assert_compile(types.DECIMAL(2, 4), 'DECIMAL(2, 4)')
 
1364
 
 
1365
 
 
1366
 
1707
1367
 
1708
1368
class NumericRawSQLTest(fixtures.TestBase):
1709
1369
    """Test what DBAPIs and dialects return without any typing
1721
1381
    @testing.provide_metadata
1722
1382
    def test_decimal_fp(self):
1723
1383
        metadata = self.metadata
1724
 
        t = self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45.5"))
 
1384
        self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45.5"))
1725
1385
        val = testing.db.execute("select val from t").scalar()
1726
1386
        assert isinstance(val, decimal.Decimal)
1727
1387
        eq_(val, decimal.Decimal("45.5"))
1730
1390
    @testing.provide_metadata
1731
1391
    def test_decimal_int(self):
1732
1392
        metadata = self.metadata
1733
 
        t = self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45"))
 
1393
        self._fixture(metadata, Numeric(10, 5), decimal.Decimal("45"))
1734
1394
        val = testing.db.execute("select val from t").scalar()
1735
1395
        assert isinstance(val, decimal.Decimal)
1736
1396
        eq_(val, decimal.Decimal("45"))
1738
1398
    @testing.provide_metadata
1739
1399
    def test_ints(self):
1740
1400
        metadata = self.metadata
1741
 
        t = self._fixture(metadata, Integer, 45)
 
1401
        self._fixture(metadata, Integer, 45)
1742
1402
        val = testing.db.execute("select val from t").scalar()
1743
1403
        assert isinstance(val, (int, long))
1744
1404
        eq_(val, 45)
1746
1406
    @testing.provide_metadata
1747
1407
    def test_float(self):
1748
1408
        metadata = self.metadata
1749
 
        t = self._fixture(metadata, Float, 46.583)
 
1409
        self._fixture(metadata, Float, 46.583)
1750
1410
        val = testing.db.execute("select val from t").scalar()
1751
1411
        assert isinstance(val, float)
1752
1412
 
1753
1413
        # some DBAPIs have unusual float handling
1754
 
        if testing.against('oracle+cx_oracle', 'mysql+oursql'):
 
1414
        if testing.against('oracle+cx_oracle', 'mysql+oursql', 'firebird'):
1755
1415
            eq_(round_decimal(val, 3), 46.583)
1756
1416
        else:
1757
1417
            eq_(val, 46.583)
1879
1539
        p1 = PickleType()
1880
1540
 
1881
1541
        for obj in (
1882
 
            {'1':'2'},
 
1542
            {'1': '2'},
1883
1543
            pickleable.Bar(5, 6),
1884
1544
            pickleable.OldSchool(10, 11)
1885
1545
        ):
1894
1554
        p1 = PickleType()
1895
1555
 
1896
1556
        for obj in (
1897
 
            {'1':'2'},
 
1557
            {'1': '2'},
1898
1558
            pickleable.Bar(5, 6),
1899
1559
            pickleable.OldSchool(10, 11)
1900
1560
        ):