diff options
Diffstat (limited to 'test/dialect/test_mssql.py')
| -rw-r--r-- | test/dialect/test_mssql.py | 55 |
1 files changed, 46 insertions, 9 deletions
diff --git a/test/dialect/test_mssql.py b/test/dialect/test_mssql.py index 52ba77310..f1cd3fe85 100644 --- a/test/dialect/test_mssql.py +++ b/test/dialect/test_mssql.py @@ -13,10 +13,10 @@ from sqlalchemy.engine import url from sqlalchemy.testing import fixtures, AssertsCompiledSQL, \ AssertsExecutionResults, ComparesTables from sqlalchemy import testing -from sqlalchemy.testing import eq_, emits_warning_on, \ - assert_raises_message -from sqlalchemy.util.compat import decimal +from sqlalchemy.testing import emits_warning_on, assert_raises_message +import decimal from sqlalchemy.engine.reflection import Inspector +from sqlalchemy.util.compat import b class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = mssql.dialect() @@ -1210,28 +1210,28 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL): eq_([1, 3, 5], [r.id for r in results]) -class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): - @classmethod - def setup_class(cls): - global dialect - dialect = pyodbc.dialect() +class ParseConnectTest(fixtures.TestBase): def test_pyodbc_connect_dsn_trusted(self): + dialect = pyodbc.dialect() u = url.make_url('mssql://mydsn') connection = dialect.create_connect_args(u) eq_([['dsn=mydsn;Trusted_Connection=Yes'], {}], connection) def test_pyodbc_connect_old_style_dsn_trusted(self): + dialect = pyodbc.dialect() u = url.make_url('mssql:///?dsn=mydsn') connection = dialect.create_connect_args(u) eq_([['dsn=mydsn;Trusted_Connection=Yes'], {}], connection) def test_pyodbc_connect_dsn_non_trusted(self): + dialect = pyodbc.dialect() u = url.make_url('mssql://username:password@mydsn') connection = dialect.create_connect_args(u) eq_([['dsn=mydsn;UID=username;PWD=password'], {}], connection) def test_pyodbc_connect_dsn_extra(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql://username:password@mydsn/?LANGUAGE=us_' 'english&foo=bar') @@ -1241,12 +1241,14 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): assert ";foo=bar" in dsn_string def test_pyodbc_connect(self): + dialect = pyodbc.dialect() u = url.make_url('mssql://username:password@hostspec/database') connection = dialect.create_connect_args(u) eq_([['DRIVER={SQL Server};Server=hostspec;Database=database;UI' 'D=username;PWD=password'], {}], connection) def test_pyodbc_connect_comma_port(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql://username:password@hostspec:12345/data' 'base') @@ -1255,6 +1257,7 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): 'ase;UID=username;PWD=password'], {}], connection) def test_pyodbc_connect_config_port(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql://username:password@hostspec/database?p' 'ort=12345') @@ -1263,6 +1266,7 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): 'D=username;PWD=password;port=12345'], {}], connection) def test_pyodbc_extra_connect(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql://username:password@hostspec/database?L' 'ANGUAGE=us_english&foo=bar') @@ -1275,6 +1279,7 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): 'username;PWD=password;LANGUAGE=us_english;foo=bar'), True) def test_pyodbc_odbc_connect(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql:///?odbc_connect=DRIVER%3D%7BSQL+Server' '%7D%3BServer%3Dhostspec%3BDatabase%3Ddatabase' @@ -1284,6 +1289,7 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): 'D=username;PWD=password'], {}], connection) def test_pyodbc_odbc_connect_with_dsn(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql:///?odbc_connect=dsn%3Dmydsn%3BDatabase' '%3Ddatabase%3BUID%3Dusername%3BPWD%3Dpassword' @@ -1293,6 +1299,7 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): {}], connection) def test_pyodbc_odbc_connect_ignores_other_values(self): + dialect = pyodbc.dialect() u = \ url.make_url('mssql://userdiff:passdiff@localhost/dbdiff?od' 'bc_connect=DRIVER%3D%7BSQL+Server%7D%3BServer' @@ -1321,7 +1328,22 @@ class ParseConnectTest(fixtures.TestBase, AssertsCompiledSQL): 'user': 'scott', 'database': 'test'}], connection ) - @testing.only_on(['mssql+pyodbc', 'mssql+pymssql'], "FreeTDS specific test") + def test_pymssql_disconnect(self): + dialect = pymssql.dialect() + + for error in [ + 'Adaptive Server connection timed out', + 'message 20003', + "Error 10054", + "Not connected to any MS SQL server", + "Connection is closed" + ]: + eq_(dialect.is_disconnect(error, None, None), True) + + eq_(dialect.is_disconnect("not an error", None, None), False) + + @testing.only_on(['mssql+pyodbc', 'mssql+pymssql'], + "FreeTDS specific test") def test_bad_freetds_warning(self): engine = engines.testing_engine() @@ -1926,6 +1948,21 @@ class TypeRoundTripTest(fixtures.TestBase, AssertsExecutionResults, ComparesTabl not in list(engine.execute(tbl.select()).first()) engine.execute(tbl.delete()) +class MonkeyPatchedBinaryTest(fixtures.TestBase): + __only_on__ = 'mssql+pymssql' + + def test_unicode(self): + module = __import__('pymssql') + result = module.Binary(u'foo') + eq_(result, u'foo') + + def test_bytes(self): + module = __import__('pymssql') + input = b('\x80\x03]q\x00X\x03\x00\x00\x00oneq\x01a.') + expected_result = input + result = module.Binary(input) + eq_(result, expected_result) + class BinaryTest(fixtures.TestBase, AssertsExecutionResults): """Test the Binary and VarBinary types""" |