mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-16 21:57:22 -04:00
dc91c7db7f
"Implicit autocommit", which is the COMMIT that occurs when a DML or DDL statement is emitted on a connection, is deprecated and won't be part of SQLAlchemy 2.0. A 2.0-style warning is emitted when autocommit takes effect, so that the calling code may be adjusted to use an explicit transaction. As part of this change, DDL methods such as :meth:`_schema.MetaData.create_all` when used against a :class:`_engine.Engine` or :class:`_engine.Connection` will run the operation in a BEGIN block if one is not started already. The MySQL and MariaDB dialects now query from the information_schema.tables system view in order to determine if a particular table exists or not. Previously, the "DESCRIBE" command was used with an exception catch to detect non-existent, which would have the undesirable effect of emitting a ROLLBACK on the connection. There appeared to be legacy encoding issues which prevented the use of "SHOW TABLES", for this, but as MySQL support is now at 5.0.2 or above due to 🎫`4189`, the information_schema tables are now available in all cases. Fixes: #4846 Change-Id: I733a7e0e17477a63607fb9931c87c393bbd7ac57
160 lines
4.7 KiB
Python
160 lines
4.7 KiB
Python
from __future__ import print_function
|
|
|
|
import doctest
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
from sqlalchemy import testing
|
|
from sqlalchemy.testing import config
|
|
from sqlalchemy.testing import fixtures
|
|
|
|
|
|
class DocTest(fixtures.TestBase):
|
|
__requires__ = ("python3",)
|
|
|
|
def _setup_logger(self):
|
|
rootlogger = logging.getLogger("sqlalchemy.engine.Engine")
|
|
|
|
class MyStream(object):
|
|
def write(self, string):
|
|
sys.stdout.write(string)
|
|
sys.stdout.flush()
|
|
|
|
def flush(self):
|
|
pass
|
|
|
|
self._handler = handler = logging.StreamHandler(MyStream())
|
|
handler.setFormatter(logging.Formatter("%(message)s"))
|
|
rootlogger.addHandler(handler)
|
|
|
|
def _teardown_logger(self):
|
|
rootlogger = logging.getLogger("sqlalchemy.engine.Engine")
|
|
rootlogger.removeHandler(self._handler)
|
|
|
|
def _setup_create_table_patcher(self):
|
|
from sqlalchemy.sql import ddl
|
|
|
|
self.orig_sort = ddl.sort_tables_and_constraints
|
|
|
|
def our_sort(tables, **kw):
|
|
return self.orig_sort(sorted(tables, key=lambda t: t.key), **kw)
|
|
|
|
ddl.sort_tables_and_constraints = our_sort
|
|
|
|
def _teardown_create_table_patcher(self):
|
|
from sqlalchemy.sql import ddl
|
|
|
|
ddl.sort_tables_and_constraints = self.orig_sort
|
|
|
|
def setup(self):
|
|
self._setup_logger()
|
|
self._setup_create_table_patcher()
|
|
|
|
def teardown(self):
|
|
self._teardown_create_table_patcher()
|
|
self._teardown_logger()
|
|
|
|
def _run_doctest_for_content(self, name, content):
|
|
optionflags = (
|
|
doctest.ELLIPSIS
|
|
| doctest.NORMALIZE_WHITESPACE
|
|
| doctest.IGNORE_EXCEPTION_DETAIL
|
|
| _get_allow_unicode_flag()
|
|
)
|
|
runner = doctest.DocTestRunner(
|
|
verbose=None,
|
|
optionflags=optionflags,
|
|
checker=_get_unicode_checker(),
|
|
)
|
|
globs = {"print_function": print_function}
|
|
parser = doctest.DocTestParser()
|
|
test = parser.get_doctest(content, globs, name, name, 0)
|
|
runner.run(test)
|
|
runner.summarize()
|
|
assert not runner.failures
|
|
|
|
def _run_doctest(self, fname):
|
|
here = os.path.dirname(__file__)
|
|
sqla_base = os.path.normpath(os.path.join(here, "..", ".."))
|
|
path = os.path.join(sqla_base, "doc/build", fname)
|
|
if not os.path.exists(path):
|
|
config.skip_test("Can't find documentation file %r" % path)
|
|
with open(path) as file_:
|
|
content = file_.read()
|
|
content = re.sub(r"{(?:stop|sql|opensql)}", "", content)
|
|
self._run_doctest_for_content(fname, content)
|
|
|
|
def test_orm(self):
|
|
self._run_doctest("orm/tutorial.rst")
|
|
|
|
@testing.emits_warning()
|
|
def test_core(self):
|
|
self._run_doctest("core/tutorial.rst")
|
|
|
|
|
|
# unicode checker courtesy pytest
|
|
|
|
|
|
def _get_unicode_checker():
|
|
"""
|
|
Returns a doctest.OutputChecker subclass that takes in account the
|
|
ALLOW_UNICODE option to ignore u'' prefixes in strings. Useful
|
|
when the same doctest should run in Python 2 and Python 3.
|
|
|
|
An inner class is used to avoid importing "doctest" at the module
|
|
level.
|
|
"""
|
|
if hasattr(_get_unicode_checker, "UnicodeOutputChecker"):
|
|
return _get_unicode_checker.UnicodeOutputChecker()
|
|
|
|
import doctest
|
|
import re
|
|
|
|
class UnicodeOutputChecker(doctest.OutputChecker):
|
|
"""
|
|
Copied from doctest_nose_plugin.py from the nltk project:
|
|
https://github.com/nltk/nltk
|
|
"""
|
|
|
|
_literal_re = re.compile(r"(\W|^)[uU]([rR]?[\'\"])", re.UNICODE)
|
|
|
|
def check_output(self, want, got, optionflags):
|
|
res = doctest.OutputChecker.check_output(
|
|
self, want, got, optionflags
|
|
)
|
|
if res:
|
|
return True
|
|
|
|
if not (optionflags & _get_allow_unicode_flag()):
|
|
return False
|
|
|
|
else: # pragma: no cover
|
|
# the code below will end up executed only in Python 2 in
|
|
# our tests, and our coverage check runs in Python 3 only
|
|
def remove_u_prefixes(txt):
|
|
return re.sub(self._literal_re, r"\1\2", txt)
|
|
|
|
want = remove_u_prefixes(want)
|
|
got = remove_u_prefixes(got)
|
|
res = doctest.OutputChecker.check_output(
|
|
self, want, got, optionflags
|
|
)
|
|
return res
|
|
|
|
_get_unicode_checker.UnicodeOutputChecker = UnicodeOutputChecker
|
|
return _get_unicode_checker.UnicodeOutputChecker()
|
|
|
|
|
|
def _get_allow_unicode_flag():
|
|
"""
|
|
Registers and returns the ALLOW_UNICODE flag.
|
|
"""
|
|
import doctest
|
|
|
|
return doctest.register_optionflag("ALLOW_UNICODE")
|
|
|
|
|
|
# increase number to force pipeline run. 1
|