summaryrefslogtreecommitdiff
path: root/test/dialect/test_sqlite.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/dialect/test_sqlite.py')
-rw-r--r--test/dialect/test_sqlite.py117
1 files changed, 58 insertions, 59 deletions
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index eb4581e20..448ee947c 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -4,7 +4,7 @@ from sqlalchemy.test.testing import eq_, assert_raises, assert_raises_message
import datetime
from sqlalchemy import *
from sqlalchemy import exc, sql
-from sqlalchemy.databases import sqlite
+from sqlalchemy.dialects.sqlite import base as sqlite, pysqlite as pysqlite_dialect
from sqlalchemy.test import *
@@ -19,7 +19,7 @@ class TestTypes(TestBase, AssertsExecutionResults):
meta = MetaData(testing.db)
t = Table('bool_table', meta,
Column('id', Integer, primary_key=True),
- Column('boo', sqlite.SLBoolean))
+ Column('boo', Boolean))
try:
meta.create_all()
@@ -39,7 +39,7 @@ class TestTypes(TestBase, AssertsExecutionResults):
def test_time_microseconds(self):
dt = datetime.datetime(2008, 6, 27, 12, 0, 0, 125) # 125 usec
eq_(str(dt), '2008-06-27 12:00:00.000125')
- sldt = sqlite.SLDateTime()
+ sldt = sqlite._SLDateTime()
bp = sldt.bind_processor(None)
eq_(bp(dt), '2008-06-27 12:00:00.000125')
@@ -69,59 +69,44 @@ class TestTypes(TestBase, AssertsExecutionResults):
bindproc = t.dialect_impl(dialect).bind_processor(dialect)
assert not bindproc or isinstance(bindproc(u"some string"), unicode)
- @testing.uses_deprecated('Using String type with no length')
def test_type_reflection(self):
# (ask_for, roundtripped_as_if_different)
- specs = [( String(), sqlite.SLString(), ),
- ( String(1), sqlite.SLString(1), ),
- ( String(3), sqlite.SLString(3), ),
- ( Text(), sqlite.SLText(), ),
- ( Unicode(), sqlite.SLString(), ),
- ( Unicode(1), sqlite.SLString(1), ),
- ( Unicode(3), sqlite.SLString(3), ),
- ( UnicodeText(), sqlite.SLText(), ),
- ( CLOB, sqlite.SLText(), ),
- ( sqlite.SLChar(1), ),
- ( CHAR(3), sqlite.SLChar(3), ),
- ( NCHAR(2), sqlite.SLChar(2), ),
- ( SmallInteger(), sqlite.SLSmallInteger(), ),
- ( sqlite.SLSmallInteger(), ),
- ( Binary(3), sqlite.SLBinary(), ),
- ( Binary(), sqlite.SLBinary() ),
- ( sqlite.SLBinary(3), sqlite.SLBinary(), ),
- ( NUMERIC, sqlite.SLNumeric(), ),
- ( NUMERIC(10,2), sqlite.SLNumeric(10,2), ),
- ( Numeric, sqlite.SLNumeric(), ),
- ( Numeric(10, 2), sqlite.SLNumeric(10, 2), ),
- ( DECIMAL, sqlite.SLNumeric(), ),
- ( DECIMAL(10, 2), sqlite.SLNumeric(10, 2), ),
- ( Float, sqlite.SLFloat(), ),
- ( sqlite.SLNumeric(), ),
- ( INT, sqlite.SLInteger(), ),
- ( Integer, sqlite.SLInteger(), ),
- ( sqlite.SLInteger(), ),
- ( TIMESTAMP, sqlite.SLDateTime(), ),
- ( DATETIME, sqlite.SLDateTime(), ),
- ( DateTime, sqlite.SLDateTime(), ),
- ( sqlite.SLDateTime(), ),
- ( DATE, sqlite.SLDate(), ),
- ( Date, sqlite.SLDate(), ),
- ( sqlite.SLDate(), ),
- ( TIME, sqlite.SLTime(), ),
- ( Time, sqlite.SLTime(), ),
- ( sqlite.SLTime(), ),
- ( BOOLEAN, sqlite.SLBoolean(), ),
- ( Boolean, sqlite.SLBoolean(), ),
- ( sqlite.SLBoolean(), ),
+ specs = [( String(), String(), ),
+ ( String(1), String(1), ),
+ ( String(3), String(3), ),
+ ( Text(), Text(), ),
+ ( Unicode(), String(), ),
+ ( Unicode(1), String(1), ),
+ ( Unicode(3), String(3), ),
+ ( UnicodeText(), Text(), ),
+ ( CHAR(1), ),
+ ( CHAR(3), CHAR(3), ),
+ ( NUMERIC, NUMERIC(), ),
+ ( NUMERIC(10,2), NUMERIC(10,2), ),
+ ( Numeric, NUMERIC(), ),
+ ( Numeric(10, 2), NUMERIC(10, 2), ),
+ ( DECIMAL, DECIMAL(), ),
+ ( DECIMAL(10, 2), DECIMAL(10, 2), ),
+ ( Float, Float(), ),
+ ( NUMERIC(), ),
+ ( TIMESTAMP, TIMESTAMP(), ),
+ ( DATETIME, DATETIME(), ),
+ ( DateTime, DateTime(), ),
+ ( DateTime(), ),
+ ( DATE, DATE(), ),
+ ( Date, Date(), ),
+ ( TIME, TIME(), ),
+ ( Time, Time(), ),
+ ( BOOLEAN, BOOLEAN(), ),
+ ( Boolean, Boolean(), ),
]
columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
db = testing.db
m = MetaData(db)
t_table = Table('types', m, *columns)
+ m.create_all()
try:
- m.create_all()
-
m2 = MetaData(db)
rt = Table('types', m2, autoload=True)
try:
@@ -131,7 +116,7 @@ class TestTypes(TestBase, AssertsExecutionResults):
expected = [len(c) > 1 and c[1] or c[0] for c in specs]
for table in rt, rv:
for i, reflected in enumerate(table.c):
- assert isinstance(reflected.type, type(expected[i])), type(expected[i])
+ assert isinstance(reflected.type, type(expected[i])), "%d: %r" % (i, type(expected[i]))
finally:
db.execute('DROP VIEW types_v')
finally:
@@ -163,7 +148,7 @@ class TestDefaults(TestBase, AssertsExecutionResults):
rt = Table('t_defaults', m2, autoload=True)
expected = [c[1] for c in specs]
for i, reflected in enumerate(rt.c):
- eq_(reflected.server_default.arg.text, expected[i])
+ eq_(str(reflected.server_default.arg), expected[i])
finally:
m.drop_all()
@@ -173,7 +158,7 @@ class TestDefaults(TestBase, AssertsExecutionResults):
db = testing.db
m = MetaData(db)
- expected = ["'my_default'", '0']
+ expected = ["my_default", '0']
table = """CREATE TABLE r_defaults (
data VARCHAR(40) DEFAULT 'my_default',
val INTEGER NOT NULL DEFAULT 0
@@ -184,7 +169,7 @@ class TestDefaults(TestBase, AssertsExecutionResults):
rt = Table('r_defaults', m, autoload=True)
for i, reflected in enumerate(rt.c):
- eq_(reflected.server_default.arg.text, expected[i])
+ eq_(str(reflected.server_default.arg), expected[i])
finally:
db.execute("DROP TABLE r_defaults")
@@ -247,24 +232,24 @@ class DialectTest(TestBase, AssertsExecutionResults):
def test_attached_as_schema(self):
cx = testing.db.connect()
try:
- cx.execute('ATTACH DATABASE ":memory:" AS alt_schema')
+ cx.execute('ATTACH DATABASE ":memory:" AS test_schema')
dialect = cx.dialect
- assert dialect.table_names(cx, 'alt_schema') == []
+ assert dialect.table_names(cx, 'test_schema') == []
meta = MetaData(cx)
Table('created', meta, Column('id', Integer),
- schema='alt_schema')
+ schema='test_schema')
alt_master = Table('sqlite_master', meta, autoload=True,
- schema='alt_schema')
+ schema='test_schema')
meta.create_all(cx)
- eq_(dialect.table_names(cx, 'alt_schema'),
+ eq_(dialect.table_names(cx, 'test_schema'),
['created'])
assert len(alt_master.c) > 0
meta.clear()
reflected = Table('created', meta, autoload=True,
- schema='alt_schema')
+ schema='test_schema')
assert len(reflected.c) == 1
cx.execute(reflected.insert(), dict(id=1))
@@ -282,9 +267,9 @@ class DialectTest(TestBase, AssertsExecutionResults):
# note that sqlite_master is cleared, above
meta.drop_all()
- assert dialect.table_names(cx, 'alt_schema') == []
+ assert dialect.table_names(cx, 'test_schema') == []
finally:
- cx.execute('DETACH DATABASE alt_schema')
+ cx.execute('DETACH DATABASE test_schema')
@testing.exclude('sqlite', '<', (2, 6), 'no database support')
def test_temp_table_reflection(self):
@@ -305,6 +290,20 @@ class DialectTest(TestBase, AssertsExecutionResults):
pass
raise
+ def test_set_isolation_level(self):
+ """Test setting the read uncommitted/serializable levels"""
+ eng = create_engine(testing.db.url)
+ eq_(eng.execute("PRAGMA read_uncommitted").scalar(), 0)
+
+ eng = create_engine(testing.db.url, isolation_level="READ UNCOMMITTED")
+ eq_(eng.execute("PRAGMA read_uncommitted").scalar(), 1)
+
+ eng = create_engine(testing.db.url, isolation_level="SERIALIZABLE")
+ eq_(eng.execute("PRAGMA read_uncommitted").scalar(), 0)
+
+ assert_raises(exc.ArgumentError, create_engine, testing.db.url,
+ isolation_level="FOO")
+
class SQLTest(TestBase, AssertsCompiledSQL):
"""Tests SQLite-dialect specific compilation."""