Skip to content

Instantly share code, notes, and snippets.

@dcramer
Created March 26, 2013 02:04
Show Gist options
  • Select an option

  • Save dcramer/5242540 to your computer and use it in GitHub Desktop.

Select an option

Save dcramer/5242540 to your computer and use it in GitHub Desktop.
$ py.test -x
======================================================== test session starts ========================================================
platform darwin -- Python 2.7.2 -- pytest-2.3.4
plugins: django
collected 3 items
tests/djnydus/db/base/tests.py E
============================================================== ERRORS ===============================================================
________________________________________ ERROR at setup of DjangoConnectionsTest.test_simple ________________________________________
request = <FixtureRequest for <Session 'nydus-django'>>
_django_cursor_wrapper = <pytest_django.plugin.CursorManager object at 0x10ab9a710>
@pytest.fixture(autouse=True)
def _django_setup_unittest(request, _django_cursor_wrapper):
"""Setup a django unittest, internal to pytest-django"""
if django_settings_is_configured() and is_django_unittest(request.node):
request.getfuncargvalue('_django_runner')
> request.getfuncargvalue('_django_db_setup')
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/pytest_django/plugin.py:159:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <FixtureRequest for <Session 'nydus-django'>>, argname = '_django_db_setup'
def getfuncargvalue(self, argname):
""" Dynamically retrieve a named fixture function argument.
As of pytest-2.3, it is easier and usually better to access other
fixture values by stating it as an input argument in the fixture
function. If you only can decide about using another fixture at test
setup time, you may use this function to retrieve it inside a fixture
function body.
"""
try:
return self._funcargs[argname]
except KeyError:
pass
try:
fixturedef = self._getnextfixturedef(argname)
except FixtureLookupError:
if argname == "request":
return self
raise
self._fixturestack.append(fixturedef)
try:
> result = self._getfuncargvalue(fixturedef)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/_pytest/python.py:1195:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <FixtureRequest for <Session 'nydus-django'>>, fixturedef = <FixtureDef name='_django_db_setup' scope='session'>
def _getfuncargvalue(self, fixturedef):
try:
return fixturedef.cached_result # set by fixturedef.execute()
except AttributeError:
pass
# prepare request fixturename and param attributes before
# calling into fixture function
argname = fixturedef.argname
node = self._pyfuncitem
mp = monkeypatch()
mp.setattr(self, 'fixturename', argname)
try:
param = node.callspec.getparam(argname)
except (AttributeError, ValueError):
pass
else:
mp.setattr(self, 'param', param, raising=False)
# if a parametrize invocation set a scope it will override
# the static scope defined with the fixture function
scope = fixturedef.scope
try:
paramscopenum = node.callspec._arg2scopenum[argname]
except (KeyError, AttributeError):
pass
else:
if paramscopenum != scopenum_subfunction:
scope = scopes[paramscopenum]
# check if a higher-level scoped fixture accesses a lower level one
if scope is not None:
__tracebackhide__ = True
if scopemismatch(self.scope, scope):
# try to report something helpful
lines = self._factorytraceback()
raise ScopeMismatchError("You tried to access the %r scoped "
"funcarg %r with a %r scoped request object, "
"involved factories\n%s" %(
(scope, argname, self.scope, "\n".join(lines))))
__tracebackhide__ = False
mp.setattr(self, "scope", scope)
# route request.addfinalizer to fixturedef
mp.setattr(self, "addfinalizer", fixturedef.addfinalizer)
# perform the fixture call
> val = fixturedef.execute(request=self)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/_pytest/python.py:1248:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <FixtureDef name='_django_db_setup' scope='session'>, request = <FixtureRequest for <Session 'nydus-django'>>
def execute(self, request):
kwargs = {}
for newname in self.argnames:
kwargs[newname] = request.getfuncargvalue(newname)
if self.unittest:
result = self.func(request.instance, **kwargs)
else:
fixturefunc = self.func
# the fixture function needs to be bound to the actual
# request.instance so that code working with "self" behaves
# as expected. XXX request.instance should maybe return None
# instead of raising AttributeError
try:
if request.instance is not None:
fixturefunc = getimfunc(self.func)
if fixturefunc != self.func:
fixturefunc = fixturefunc.__get__(request.instance)
except AttributeError:
pass
> result = fixturefunc(**kwargs)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/_pytest/python.py:1662:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
request = <FixtureRequest for <Session 'nydus-django'>>
_django_runner = <django.test.simple.DjangoTestSuiteRunner object at 0x10b18d0d0>
_django_cursor_wrapper = <pytest_django.plugin.CursorManager object at 0x10ab9a710>
@pytest.fixture(scope='session')
def _django_db_setup(request, _django_runner, _django_cursor_wrapper):
"""Session-wide database setup, internal to pytest-django"""
skip_if_no_django()
from django.core import management
# Disable south's syncdb command
commands = management.get_commands()
if commands['syncdb'] == 'south':
management._commands['syncdb'] = 'django.core'
with _django_cursor_wrapper:
# Monkey patch Django's setup code to support database re-use
if request.config.getvalue('reuse_db'):
if not request.config.getvalue('create_db'):
monkey_patch_creation_for_db_reuse()
_django_runner.teardown_databases = lambda db_cfg: None
# Create the database
> db_cfg = _django_runner.setup_databases()
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/pytest_django/fixtures.py:44:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.test.simple.DjangoTestSuiteRunner object at 0x10b18d0d0>, kwargs = {}
connections = <django.db.utils.ConnectionHandler object at 0x10a820790>, DEFAULT_DB_ALIAS = 'default', mirrored_aliases = {}
test_databases = {('', '', 'djnydus.db.backend', ''): ('', ['default'])}, dependencies = {}, alias = 'default'
connection = <djnydus.db.backend.base.DatabaseWrapper object at 0x10a8208d0>
def setup_databases(self, **kwargs):
from django.db import connections, DEFAULT_DB_ALIAS
# First pass -- work out which databases actually need to be created,
# and which ones are test mirrors or duplicate entries in DATABASES
mirrored_aliases = {}
test_databases = {}
dependencies = {}
for alias in connections:
connection = connections[alias]
if connection.settings_dict['TEST_MIRROR']:
# If the database is marked as a test mirror, save
# the alias.
mirrored_aliases[alias] = (
connection.settings_dict['TEST_MIRROR'])
else:
# Store a tuple with DB parameters that uniquely identify it.
# If we have two aliases with the same values for that tuple,
# we only need to create the test database once.
item = test_databases.setdefault(
connection.creation.test_db_signature(),
(connection.settings_dict['NAME'], [])
)
item[1].append(alias)
if 'TEST_DEPENDENCIES' in connection.settings_dict:
dependencies[alias] = (
connection.settings_dict['TEST_DEPENDENCIES'])
else:
if alias != DEFAULT_DB_ALIAS:
dependencies[alias] = connection.settings_dict.get(
'TEST_DEPENDENCIES', [DEFAULT_DB_ALIAS])
# Second pass -- actually create the databases.
old_names = []
mirrors = []
for signature, (db_name, aliases) in dependency_ordered(
test_databases.items(), dependencies):
# Actually create the database for the first connection
connection = connections[aliases[0]]
old_names.append((connection, db_name, True))
test_db_name = connection.creation.create_test_db(
> self.verbosity, autoclobber=not self.interactive)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/test/simple.py:317:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <djnydus.db.backend.base.DatabaseCreation object at 0x10aa1c190>, args = (1,), kwargs = {'autoclobber': True}
supports_trans = True, connection = <djnydus.db.base.DjangoDatabase object at 0x10a1e8f10>
def create_test_db(self, *args, **kwargs):
supports_trans = True
for connection in self.cluster.hosts.itervalues():
> connection.creation.create_test_db(*args, **kwargs)
src/djnydus/db/backend/base.py:36:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.db.backends.sqlite3.creation.DatabaseCreation object at 0x10aa09e50>, verbosity = 1, autoclobber = True
def create_test_db(self, verbosity=1, autoclobber=False):
"""
Creates a test database, prompting the user for confirmation if the
database already exists. Returns the name of the test database created.
"""
# Don't import django.core.management if it isn't needed.
from django.core.management import call_command
test_database_name = self._get_test_db_name()
if verbosity >= 1:
test_db_repr = ''
if verbosity >= 2:
test_db_repr = " ('%s')" % test_database_name
print "Creating test database for alias '%s'%s..." % (
self.connection.alias, test_db_repr)
self._create_test_db(verbosity, autoclobber)
self.connection.close()
self.connection.settings_dict["NAME"] = test_database_name
# Confirm the feature set of the test database
self.connection.features.confirm()
# Report syncdb messages at one level lower than that requested.
# This ensures we don't get flooded with messages during testing
# (unless you really ask to be flooded)
call_command('syncdb',
verbosity=max(verbosity - 1, 0),
interactive=False,
database=self.connection.alias,
load_initial_data=False)
# We need to then do a flush to ensure that any data installed by
# custom SQL has been removed. The only test data should come from
# test fixtures, or autogenerated from post_syncdb triggers.
# This has the side effect of loading initial data (which was
# intentionally skipped in the syncdb).
call_command('flush',
verbosity=max(verbosity - 1, 0),
interactive=False,
> database=self.connection.alias)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/db/backends/creation.py:281:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
name = 'flush', args = (), options = {'database': 'default', 'interactive': False, 'verbosity': 0}, app_name = 'django.core'
klass = <django.core.management.commands.flush.Command object at 0x10b18d990>
defaults = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}
opt = <Option at 0x10b197998: --database>
def call_command(name, *args, **options):
"""
Calls the given command, with the given options and args/kwargs.
This is the primary API you should use for calling specific commands.
Some examples:
call_command('syncdb')
call_command('shell', plain=True)
call_command('sqlall', 'myapp')
"""
# Load the command object.
try:
app_name = get_commands()[name]
if isinstance(app_name, BaseCommand):
# If the command is already loaded, use it directly.
klass = app_name
else:
klass = load_command_class(app_name, name)
except KeyError:
raise CommandError("Unknown command: %r" % name)
# Grab out a list of defaults from the options. optparse does this for us
# when the script runs from the command line, but since call_command can
# be called programatically, we need to simulate the loading and handling
# of defaults (see #10080 for details).
defaults = {}
for opt in klass.option_list:
if opt.default is NO_DEFAULT:
defaults[opt.dest] = None
else:
defaults[opt.dest] = opt.default
defaults.update(options)
> return klass.execute(*args, **defaults)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/__init__.py:150:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.core.management.commands.flush.Command object at 0x10b18d990>, args = ()
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}, show_traceback = None
saved_lang = 'en-us'
translation = <module 'django.utils.translation' from '/Users/dcramer/.virtualenvs/nydus-django/lib/python2.7/site-packages/django/utils/translation/__init__.py'>
def execute(self, *args, **options):
"""
Try to execute this command, performing model validation if
needed (as controlled by the attribute
``self.requires_model_validation``). If the command raises a
``CommandError``, intercept it and print it sensibly to
stderr.
"""
show_traceback = options.get('traceback', False)
# Switch to English, because django-admin.py creates database content
# like permissions, and those shouldn't contain any translations.
# But only do this if we can assume we have a working settings file,
# because django.utils.translation requires settings.
saved_lang = None
if self.can_import_settings:
try:
from django.utils import translation
saved_lang = translation.get_language()
translation.activate('en-us')
except ImportError, e:
# If settings should be available, but aren't,
# raise the error and quit.
if show_traceback:
traceback.print_exc()
else:
sys.stderr.write(smart_str(self.style.ERROR('Error: %s\n' % e)))
sys.exit(1)
try:
self.stdout = options.get('stdout', sys.stdout)
self.stderr = options.get('stderr', sys.stderr)
if self.requires_model_validation:
self.validate()
> output = self.handle(*args, **options)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/base.py:232:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.core.management.commands.flush.Command object at 0x10b18d990>, args = ()
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}
def handle(self, *args, **options):
if args:
raise CommandError("Command doesn't accept any arguments")
> return self.handle_noargs(**options)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/base.py:371:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.core.management.commands.flush.Command object at 0x10b18d990>
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}, db = 'default'
connection = <djnydus.db.backend.base.DatabaseWrapper object at 0x10a8208d0>, verbosity = 0, interactive = False
app_name = 'djnydus', sql_list = [], confirm = 'yes'
def handle_noargs(self, **options):
db = options.get('database')
connection = connections[db]
verbosity = int(options.get('verbosity'))
interactive = options.get('interactive')
self.style = no_style()
# Import the 'management' module within each installed app, to register
# dispatcher events.
for app_name in settings.INSTALLED_APPS:
try:
import_module('.management', app_name)
except ImportError:
pass
sql_list = sql_flush(self.style, connection, only_django=True)
if interactive:
confirm = raw_input("""You have requested a flush of the database.
This will IRREVERSIBLY DESTROY all data currently in the %r database,
and return each table to the state it was in after syncdb.
Are you sure you want to do this?
Type 'yes' to continue, or 'no' to cancel: """ % connection.settings_dict['NAME'])
else:
confirm = 'yes'
if confirm == 'yes':
try:
cursor = connection.cursor()
for sql in sql_list:
cursor.execute(sql)
except Exception, e:
transaction.rollback_unless_managed(using=db)
raise CommandError("""Database %s couldn't be flushed. Possible reasons:
* The database isn't running or isn't configured correctly.
* At least one of the expected database tables doesn't exist.
* The SQL was invalid.
Hint: Look at the output of 'django-admin.py sqlflush'. That's the SQL this command wasn't able to run.
The full error: %s""" % (connection.settings_dict['NAME'], e))
transaction.commit_unless_managed(using=db)
# Emit the post sync signal. This allows individual
# applications to respond as if the database had been
# sync'd from scratch.
all_models = []
for app in models.get_apps():
all_models.extend([
m for m in models.get_models(app, include_auto_created=True)
if router.allow_syncdb(db, m)
])
emit_post_sync_signal(set(all_models), verbosity, interactive, db)
# Reinstall the initial_data fixture.
kwargs = options.copy()
kwargs['database'] = db
> call_command('loaddata', 'initial_data', **kwargs)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/commands/flush.py:82:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
name = 'loaddata', args = ('initial_data',)
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}, app_name = 'django.core'
klass = <django.core.management.commands.loaddata.Command object at 0x10ae0c110>
defaults = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}
opt = <Option at 0x10ae096c8: --database>
def call_command(name, *args, **options):
"""
Calls the given command, with the given options and args/kwargs.
This is the primary API you should use for calling specific commands.
Some examples:
call_command('syncdb')
call_command('shell', plain=True)
call_command('sqlall', 'myapp')
"""
# Load the command object.
try:
app_name = get_commands()[name]
if isinstance(app_name, BaseCommand):
# If the command is already loaded, use it directly.
klass = app_name
else:
klass = load_command_class(app_name, name)
except KeyError:
raise CommandError("Unknown command: %r" % name)
# Grab out a list of defaults from the options. optparse does this for us
# when the script runs from the command line, but since call_command can
# be called programatically, we need to simulate the loading and handling
# of defaults (see #10080 for details).
defaults = {}
for opt in klass.option_list:
if opt.default is NO_DEFAULT:
defaults[opt.dest] = None
else:
defaults[opt.dest] = opt.default
defaults.update(options)
> return klass.execute(*args, **defaults)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/__init__.py:150:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.core.management.commands.loaddata.Command object at 0x10ae0c110>, args = ('initial_data',)
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}, show_traceback = None
saved_lang = 'en-us'
translation = <module 'django.utils.translation' from '/Users/dcramer/.virtualenvs/nydus-django/lib/python2.7/site-packages/django/utils/translation/__init__.py'>
def execute(self, *args, **options):
"""
Try to execute this command, performing model validation if
needed (as controlled by the attribute
``self.requires_model_validation``). If the command raises a
``CommandError``, intercept it and print it sensibly to
stderr.
"""
show_traceback = options.get('traceback', False)
# Switch to English, because django-admin.py creates database content
# like permissions, and those shouldn't contain any translations.
# But only do this if we can assume we have a working settings file,
# because django.utils.translation requires settings.
saved_lang = None
if self.can_import_settings:
try:
from django.utils import translation
saved_lang = translation.get_language()
translation.activate('en-us')
except ImportError, e:
# If settings should be available, but aren't,
# raise the error and quit.
if show_traceback:
traceback.print_exc()
else:
sys.stderr.write(smart_str(self.style.ERROR('Error: %s\n' % e)))
sys.exit(1)
try:
self.stdout = options.get('stdout', sys.stdout)
self.stderr = options.get('stderr', sys.stderr)
if self.requires_model_validation:
self.validate()
> output = self.handle(*args, **options)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/base.py:232:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.core.management.commands.loaddata.Command object at 0x10ae0c110>, fixture_labels = ('initial_data',)
options = {'database': 'default', 'interactive': False, 'pythonpath': None, 'settings': None, ...}, using = 'default'
connection = <djnydus.db.backend.base.DatabaseWrapper object at 0x10a8208d0>, verbosity = 0, show_traceback = None, commit = True
fixture_count = 0, loaded_object_count = 0, fixture_object_count = 0, models = set([]), humanize = <function <lambda> at 0x10ae0a410>
def handle(self, *fixture_labels, **options):
using = options.get('database')
connection = connections[using]
self.style = no_style()
if not len(fixture_labels):
self.stderr.write(
self.style.ERROR("No database fixture specified. Please provide the path of at least one fixture in the command line.\n")
)
return
verbosity = int(options.get('verbosity'))
show_traceback = options.get('traceback')
# commit is a stealth option - it isn't really useful as
# a command line option, but it can be useful when invoking
# loaddata from within another script.
# If commit=True, loaddata will use its own transaction;
# if commit=False, the data load SQL will become part of
# the transaction in place when loaddata was invoked.
commit = options.get('commit', True)
# Keep a count of the installed objects and fixtures
fixture_count = 0
loaded_object_count = 0
fixture_object_count = 0
models = set()
humanize = lambda dirname: "'%s'" % dirname if dirname else 'absolute path'
# Get a cursor (even though we don't need one yet). This has
# the side effect of initializing the test database (if
# it isn't already initialized).
cursor = connection.cursor()
# Start transaction management. All fixtures are installed in a
# single transaction to ensure that all references are resolved.
if commit:
transaction.commit_unless_managed(using=using)
transaction.enter_transaction_management(using=using)
> transaction.managed(True, using=using)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/core/management/commands/loaddata.py:78:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
flag = True, using = 'default'
def managed(flag=True, using=None):
"""
Puts the transaction manager into a manual state: managed transactions have
to be committed explicitly by the user. If you switch off transaction
management and there is a pending commit/rollback, the data will be
commited.
"""
if using is None:
using = DEFAULT_DB_ALIAS
connection = connections[using]
> connection.managed(flag)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/db/transaction.py:126:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <nydus.db.base.CallProxy object at 0x10ae0c0d0>, args = (True,), kwargs = {}
def __call__(self, *args, **kwargs):
> return self.__cluster.execute(self.__path, args, kwargs)
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/nydus/db/base.py:133:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <nydus.db.base.BaseCluster object at 0x10a9f8d90>, path = 'managed', args = (True,), kwargs = {}
def execute(self, path, args, kwargs):
connections = self.__connections_for(path, args=args, kwargs=kwargs)
results = []
for conn in connections:
func = conn
for piece in path.split('.'):
func = getattr(func, piece)
for retry in xrange(self.max_connection_retries):
try:
> results.append(func(*args, **kwargs))
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/nydus/db/base.py:79:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <django.db.backends.sqlite3.base.DatabaseWrapper object at 0x10aa09d50>, flag = True
def managed(self, flag=True):
"""
Puts the transaction manager into a manual state: managed transactions have
to be committed explicitly by the user. If you switch off transaction
management and there is a pending commit/rollback, the data will be
commited.
"""
top = self.transaction_state
if top:
top[-1] = flag
if not flag and self.is_dirty():
self._commit()
self.set_clean()
else:
> raise TransactionManagementError("This code isn't under transaction "
"management")
E TransactionManagementError: This code isn't under transaction management
../../.virtualenvs/nydus-django/lib/python2.7/site-packages/django/db/backends/__init__.py:205: TransactionManagementError
---------------------------------------------------------- Captured stdout ----------------------------------------------------------
Creating test database for alias 'default'...
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Interrupted: stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
====================================================== 1 error in 1.32 seconds ======================================================
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment