← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/engine/test_parseconnect.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/engine/test_parseconnect.py
1
 
from test.lib.testing import assert_raises, assert_raises_message, eq_
 
1
from sqlalchemy.testing import assert_raises, assert_raises_message, eq_
2
2
import ConfigParser
3
3
import StringIO
4
4
import sqlalchemy.engine.url as url
5
5
from sqlalchemy import create_engine, engine_from_config, exc, pool
6
 
from sqlalchemy.engine import _coerce_config
 
6
from sqlalchemy.engine.util import _coerce_config
 
7
from sqlalchemy.engine.default import DefaultDialect
7
8
import sqlalchemy as tsa
8
 
from test.lib import fixtures, testing
 
9
from sqlalchemy.testing import fixtures
 
10
from sqlalchemy import testing
 
11
from sqlalchemy.testing.mock import Mock
 
12
 
9
13
 
10
14
class ParseConnectTest(fixtures.TestBase):
11
15
    def test_rfc1738(self):
117
121
            }
118
122
 
119
123
        prefixed = dict(ini.items('prefixed'))
120
 
        self.assert_(tsa.engine._coerce_config(prefixed, 'sqlalchemy.')
 
124
        self.assert_(_coerce_config(prefixed, 'sqlalchemy.')
121
125
                     == expected)
122
126
 
123
127
        plain = dict(ini.items('plain'))
124
 
        self.assert_(tsa.engine._coerce_config(plain, '') == expected)
 
128
        self.assert_(_coerce_config(plain, '') == expected)
125
129
 
126
130
    def test_engine_from_config(self):
127
131
        dbapi = mock_dbapi
249
253
        every backend.
250
254
 
251
255
        """
252
 
        # pretend pysqlite throws the
253
 
        # "Cannot operate on a closed database." error
254
 
        # on connect.   IRL we'd be getting Oracle's "shutdown in progress"
255
256
 
256
257
        e = create_engine('sqlite://')
257
258
        sqlite3 = e.dialect.dbapi
258
 
        class ThrowOnConnect(MockDBAPI):
259
 
            dbapi = sqlite3
260
 
            Error = sqlite3.Error
261
 
            ProgrammingError = sqlite3.ProgrammingError
262
 
            def connect(self, *args, **kw):
263
 
                raise sqlite3.ProgrammingError("Cannot operate on a closed database.")
 
259
 
 
260
        dbapi = MockDBAPI()
 
261
        dbapi.Error = sqlite3.Error,
 
262
        dbapi.ProgrammingError = sqlite3.ProgrammingError
 
263
        dbapi.connect = Mock(side_effect=sqlite3.ProgrammingError(
 
264
                                    "Cannot operate on a closed database."))
264
265
        try:
265
 
            create_engine('sqlite://', module=ThrowOnConnect()).connect()
 
266
            create_engine('sqlite://', module=dbapi).connect()
266
267
            assert False
267
268
        except tsa.exc.DBAPIError, de:
268
269
            assert de.connection_invalidated
315
316
            _initialize=False,
316
317
            )
317
318
 
318
 
class MockDBAPI(object):
319
 
    version_info = sqlite_version_info = 99, 9, 9
320
 
    sqlite_version = '99.9.9'
321
 
 
322
 
    def __init__(self, **kwargs):
323
 
        self.kwargs = kwargs
324
 
        self.paramstyle = 'named'
325
 
 
326
 
    def connect(self, *args, **kwargs):
327
 
        for k in self.kwargs:
 
319
class TestRegNewDBAPI(fixtures.TestBase):
 
320
    def test_register_base(self):
 
321
        from sqlalchemy.dialects import registry
 
322
        registry.register("mockdialect", __name__, "MockDialect")
 
323
 
 
324
        e = create_engine("mockdialect://")
 
325
        assert isinstance(e.dialect, MockDialect)
 
326
 
 
327
    def test_register_dotted(self):
 
328
        from sqlalchemy.dialects import registry
 
329
        registry.register("mockdialect.foob", __name__, "MockDialect")
 
330
 
 
331
        e = create_engine("mockdialect+foob://")
 
332
        assert isinstance(e.dialect, MockDialect)
 
333
 
 
334
    def test_register_legacy(self):
 
335
        from sqlalchemy.dialects import registry
 
336
        tokens = __name__.split(".")
 
337
 
 
338
        global dialect
 
339
        dialect = MockDialect
 
340
        registry.register("mockdialect.foob", ".".join(tokens[0:-1]), tokens[-1])
 
341
 
 
342
        e = create_engine("mockdialect+foob://")
 
343
        assert isinstance(e.dialect, MockDialect)
 
344
 
 
345
    def test_register_per_dbapi(self):
 
346
        from sqlalchemy.dialects import registry
 
347
        registry.register("mysql.my_mock_dialect", __name__, "MockDialect")
 
348
 
 
349
        e = create_engine("mysql+my_mock_dialect://")
 
350
        assert isinstance(e.dialect, MockDialect)
 
351
 
 
352
class MockDialect(DefaultDialect):
 
353
    @classmethod
 
354
    def dbapi(cls, **kw):
 
355
        return MockDBAPI()
 
356
 
 
357
def MockDBAPI(**assert_kwargs):
 
358
    connection = Mock(get_server_version_info=Mock(return_value='5.0'))
 
359
    def connect(*args, **kwargs):
 
360
        for k in assert_kwargs:
328
361
            assert k in kwargs, 'key %s not present in dictionary' % k
329
 
            assert kwargs[k] == self.kwargs[k], \
330
 
                'value %s does not match %s' % (kwargs[k],
331
 
                    self.kwargs[k])
332
 
        return MockConnection()
333
 
 
334
 
 
335
 
class MockConnection(object):
336
 
    def get_server_info(self):
337
 
        return '5.0'
338
 
 
339
 
    def close(self):
340
 
        pass
341
 
 
342
 
    def cursor(self):
343
 
        return MockCursor()
344
 
 
345
 
class MockCursor(object):
346
 
    def close(self):
347
 
        pass
 
362
            eq_(
 
363
                kwargs[k], assert_kwargs[k]
 
364
            )
 
365
        return connection
 
366
 
 
367
    return Mock(
 
368
                sqlite_version_info=(99, 9, 9,),
 
369
                version_info=(99, 9, 9,),
 
370
                sqlite_version='99.9.9',
 
371
                paramstyle='named',
 
372
                connect=Mock(side_effect=connect)
 
373
            )
348
374
 
349
375
mock_dbapi = MockDBAPI()
350
376
mock_sqlite_dbapi = msd = MockDBAPI()