from sqlalchemy.testing import eq_, engines, pickleable
import datetime
import os
-from sqlalchemy import *
+from sqlalchemy import Table, Column, MetaData, Float, \
+ Integer, String, Boolean, TIMESTAMP, Sequence, Numeric, select, \
+ Date, Time, DateTime, DefaultClause, PickleType, text
from sqlalchemy import types, schema
from sqlalchemy.databases import mssql
from sqlalchemy.dialects.mssql.base import TIME
from sqlalchemy.testing import fixtures, \
- AssertsExecutionResults, ComparesTables
+ AssertsExecutionResults, ComparesTables
from sqlalchemy import testing
from sqlalchemy.testing import emits_warning_on
import decimal
class TypeDDLTest(fixtures.TestBase):
+
def test_boolean(self):
"Exercise type specification for boolean type."
# column type, args, kwargs, expected ddl
(Boolean, [], {},
'BIT'),
- ]
+ ]
metadata = MetaData()
table_args = ['test_mssql_boolean', metadata]
for col in boolean_table.c:
index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
+ testing.eq_(
+ gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
self.assert_(repr(col))
-
def test_numeric(self):
"Exercise type specification and options for numeric types."
'TINYINT'),
(types.SmallInteger, [], {},
'SMALLINT'),
- ]
+ ]
metadata = MetaData()
table_args = ['test_mssql_numeric', metadata]
for col in numeric_table.c:
index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
+ testing.eq_(
+ gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
self.assert_(repr(col))
-
def test_char(self):
"""Exercise COLLATE-ish options on string types."""
'NTEXT'),
(mssql.MSNText, [], {'collation': 'Latin1_General_CI_AS'},
'NTEXT COLLATE Latin1_General_CI_AS'),
- ]
+ ]
metadata = MetaData()
table_args = ['test_mssql_charset', metadata]
for col in charset_table.c:
index = int(col.name[1:])
- testing.eq_(gen.get_column_specification(col),
- "%s %s" % (col.name, columns[index][3]))
+ testing.eq_(
+ gen.get_column_specification(col),
+ "%s %s" % (col.name, columns[index][3]))
self.assert_(repr(col))
-
def test_timestamp(self):
"""Exercise TIMESTAMP column."""
metadata = MetaData()
spec, expected = (TIMESTAMP, 'TIMESTAMP')
- t = Table('mssql_ts', metadata,
- Column('id', Integer, primary_key=True),
- Column('t', spec, nullable=None))
+ t = Table(
+ 'mssql_ts', metadata,
+ Column('id', Integer, primary_key=True),
+ Column('t', spec, nullable=None))
gen = dialect.ddl_compiler(dialect, schema.CreateTable(t))
testing.eq_(gen.get_column_specification(t.c.t), "t %s" % expected)
self.assert_(repr(t.c.t))
% (col.name, columns[index][3]))
self.assert_(repr(col))
-class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults, ComparesTables):
+metadata = None
+
+
+class TypeRoundTripTest(
+ fixtures.TestBase, AssertsExecutionResults, ComparesTables):
__only_on__ = 'mssql'
@classmethod
def teardown(self):
metadata.drop_all()
- @testing.fails_on_everything_except('mssql+pyodbc',
- 'this is some pyodbc-specific feature')
+ @testing.fails_on_everything_except(
+ 'mssql+pyodbc',
+ 'this is some pyodbc-specific feature')
def test_decimal_notation(self):
- numeric_table = Table('numeric_table', metadata, Column('id',
- Integer, Sequence('numeric_id_seq',
- optional=True), primary_key=True),
- Column('numericcol',
- Numeric(precision=38, scale=20,
- asdecimal=True)))
+ numeric_table = Table(
+ 'numeric_table', metadata,
+ Column(
+ 'id', Integer,
+ Sequence('numeric_id_seq', optional=True), primary_key=True),
+ Column(
+ 'numericcol',
+ Numeric(precision=38, scale=20, asdecimal=True)))
metadata.create_all()
test_items = [decimal.Decimal(d) for d in (
'1500000.00000000000000000000',
'000000000000.32E12',
'00000000000000.1E+12',
'000000000000.2E-32',
- )]
+ )]
for value in test_items:
numeric_table.insert().execute(numericcol=value)
assert value[0] in test_items, "%r not in test_items" % value[0]
def test_float(self):
- float_table = Table('float_table', metadata, Column('id',
- Integer, Sequence('numeric_id_seq',
- optional=True), primary_key=True),
- Column('floatcol', Float()))
+ float_table = Table(
+ 'float_table', metadata,
+ Column(
+ 'id', Integer,
+ Sequence('numeric_id_seq', optional=True), primary_key=True),
+ Column('floatcol', Float()))
+
metadata.create_all()
try:
test_items = [float(d) for d in (
'1E-6',
'1E-7',
'1E-8',
- )]
+ )]
for value in test_items:
float_table.insert().execute(floatcol=value)
except Exception as e:
raise e
-
# todo this should suppress warnings, but it does not
@emits_warning_on('mssql+mxodbc', r'.*does not have any indexes.*')
def test_dates(self):
(mssql.MSDateTime2, [1], {},
'DATETIME2(1)', ['>=', (10,)]),
- ]
+ ]
table_args = ['test_mssql_dates', metadata]
for index, spec in enumerate(columns):
type_, args, kw, res, requires = spec[0:5]
- if requires and testing._is_excluded('mssql', *requires) \
- or not requires:
- c = Column('c%s' % index, type_(*args,
- **kw), nullable=None)
+ if requires and \
+ testing._is_excluded('mssql', *requires) or not requires:
+ c = Column('c%s' % index, type_(*args, **kw), nullable=None)
testing.db.dialect.type_descriptor(c.type)
table_args.append(c)
dates_table = Table(*table_args)
- gen = testing.db.dialect.ddl_compiler(testing.db.dialect,
- schema.CreateTable(dates_table))
+ gen = testing.db.dialect.ddl_compiler(
+ testing.db.dialect,
+ schema.CreateTable(dates_table))
for col in dates_table.c:
index = int(col.name[1:])
testing.eq_(gen.get_column_specification(col), '%s %s'
self.assert_types_base(col, dates_table.c[col.key])
def test_date_roundtrip(self):
- t = Table('test_dates', metadata,
- Column('id', Integer,
- Sequence('datetest_id_seq', optional=True),
- primary_key=True),
- Column('adate', Date),
- Column('atime', Time),
- Column('adatetime', DateTime))
+ t = Table(
+ 'test_dates', metadata,
+ Column('id', Integer,
+ Sequence('datetest_id_seq', optional=True),
+ primary_key=True),
+ Column('adate', Date),
+ Column('atime', Time),
+ Column('adatetime', DateTime))
metadata.create_all()
d1 = datetime.date(2007, 10, 30)
t1 = datetime.time(11, 2, 32)
testing.eq_(col.type.length,
binary_table.c[col.name].type.length)
-
def test_autoincrement(self):
- Table('ai_1', metadata,
- Column('int_y', Integer, primary_key=True),
- Column('int_n', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False))
- Table('ai_2', metadata,
- Column('int_y', Integer, primary_key=True),
- Column('int_n', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False))
- Table('ai_3', metadata,
- Column('int_n', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False),
- Column('int_y', Integer, primary_key=True))
- Table('ai_4', metadata,
- Column('int_n', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False),
- Column('int_n2', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False))
- Table('ai_5', metadata,
- Column('int_y', Integer, primary_key=True),
- Column('int_n', Integer, DefaultClause('0'),
- primary_key=True, autoincrement=False))
- Table('ai_6', metadata,
- Column('o1', String(1), DefaultClause('x'),
- primary_key=True),
- Column('int_y', Integer, primary_key=True))
- Table('ai_7', metadata,
- Column('o1', String(1), DefaultClause('x'),
- primary_key=True),
- Column('o2', String(1), DefaultClause('x'),
- primary_key=True),
- Column('int_y', Integer, primary_key=True))
- Table('ai_8', metadata,
- Column('o1', String(1), DefaultClause('x'),
- primary_key=True),
- Column('o2', String(1), DefaultClause('x'),
- primary_key=True))
+ Table(
+ 'ai_1', metadata,
+ Column('int_y', Integer, primary_key=True),
+ Column(
+ 'int_n', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False))
+ Table(
+ 'ai_2', metadata,
+ Column('int_y', Integer, primary_key=True),
+ Column('int_n', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False))
+ Table(
+ 'ai_3', metadata,
+ Column('int_n', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False),
+ Column('int_y', Integer, primary_key=True))
+
+ Table(
+ 'ai_4', metadata,
+ Column('int_n', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False),
+ Column('int_n2', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False))
+ Table(
+ 'ai_5', metadata,
+ Column('int_y', Integer, primary_key=True),
+ Column('int_n', Integer, DefaultClause('0'),
+ primary_key=True, autoincrement=False))
+ Table(
+ 'ai_6', metadata,
+ Column('o1', String(1), DefaultClause('x'),
+ primary_key=True),
+ Column('int_y', Integer, primary_key=True))
+ Table(
+ 'ai_7', metadata,
+ Column('o1', String(1), DefaultClause('x'),
+ primary_key=True),
+ Column('o2', String(1), DefaultClause('x'),
+ primary_key=True),
+ Column('int_y', Integer, primary_key=True))
+ Table(
+ 'ai_8', metadata,
+ Column('o1', String(1), DefaultClause('x'),
+ primary_key=True),
+ Column('o2', String(1), DefaultClause('x'),
+ primary_key=True))
metadata.create_all()
table_names = ['ai_1', 'ai_2', 'ai_3', 'ai_4',
- 'ai_5', 'ai_6', 'ai_7', 'ai_8']
+ 'ai_5', 'ai_6', 'ai_7', 'ai_8']
mr = MetaData(testing.db)
for name in table_names:
if testing.db.driver == 'mxodbc':
eng = \
- [engines.testing_engine(options={'implicit_returning'
- : True})]
+ [engines.testing_engine(options={
+ 'implicit_returning': True})]
else:
eng = \
- [engines.testing_engine(options={'implicit_returning'
- : False}),
- engines.testing_engine(options={'implicit_returning'
- : True})]
+ [engines.testing_engine(options={
+ 'implicit_returning': False}),
+ engines.testing_engine(options={
+ 'implicit_returning': True})]
for counter, engine in enumerate(eng):
engine.execute(tbl.insert())
if 'int_y' in tbl.c:
assert engine.scalar(select([tbl.c.int_y])) \
== counter + 1
- assert list(engine.execute(tbl.select()).first()).\
- count(counter + 1) == 1
+ assert list(
+ engine.execute(tbl.select()).first()).\
+ count(counter + 1) == 1
else:
assert 1 \
not in list(engine.execute(tbl.select()).first())
engine.execute(tbl.delete())
+
class MonkeyPatchedBinaryTest(fixtures.TestBase):
__only_on__ = 'mssql+pymssql'
result = module.Binary(input)
eq_(result, expected_result)
+binary_table = None
+MyPickleType = None
+
+
class BinaryTest(fixtures.TestBase, AssertsExecutionResults):
+
"""Test the Binary and VarBinary types"""
__only_on__ = 'mssql'
Column('misc', String(30)),
Column('pickled', PickleType),
Column('mypickle', MyPickleType),
- )
+ )
binary_table.create()
def teardown(self):
data_slice=stream1[0:100],
pickled=testobj1,
mypickle=testobj3,
- )
+ )
binary_table.insert().execute(
primary_id=2,
misc='binary_data_two.dat',
data_image=stream2,
data_slice=stream2[0:99],
pickled=testobj2,
- )
+ )
# TODO: pyodbc does not seem to accept "None" for a VARBINARY
# column (data=None). error: [Microsoft][ODBC SQL Server
# misc='binary_data_two.dat', data=None, data_image=None,
# data_slice=stream2[0:99], pickled=None)
- binary_table.insert().execute(primary_id=3,
- misc='binary_data_two.dat', data_image=None,
- data_slice=stream2[0:99], pickled=None)
+ binary_table.insert().execute(
+ primary_id=3,
+ misc='binary_data_two.dat', data_image=None,
+ data_slice=stream2[0:99], pickled=None)
for stmt in \
binary_table.select(order_by=binary_table.c.primary_id), \
- text('select * from binary_table order by '
- 'binary_table.primary_id',
- typemap=dict(data=mssql.MSVarBinary(8000),
- data_image=mssql.MSImage,
- data_slice=types.BINARY(100), pickled=PickleType,
- mypickle=MyPickleType), bind=testing.db):
+ text(
+ 'select * from binary_table order by '
+ 'binary_table.primary_id',
+ typemap=dict(
+ data=mssql.MSVarBinary(8000),
+ data_image=mssql.MSImage,
+ data_slice=types.BINARY(100), pickled=PickleType,
+ mypickle=MyPickleType),
+ bind=testing.db):
l = stmt.execute().fetchall()
eq_(list(stream1), list(l[0]['data']))
paddedstream = list(stream1[0:100])
eq_(l[0]['mypickle'].stuff, 'this is the right stuff')
def load_stream(self, name, len=3000):
- fp = open(os.path.join(os.path.dirname(__file__), "..", "..", name), 'rb')
+ fp = open(
+ os.path.join(os.path.dirname(__file__), "..", "..", name), 'rb')
stream = fp.read(len)
fp.close()
return stream
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
