Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_invalid_format():
with pytest.raises(ImportFromStringError) as exc_info:
import_from_string("example:")
expected = 'Import string "example:" must be in format ":".'
assert expected in str(exc_info.value)
def test_invalid_module():
with pytest.raises(ImportFromStringError) as exc_info:
import_from_string("module_does_not_exist:myattr")
expected = 'Could not import module "module_does_not_exist".'
assert expected in str(exc_info.value)
def test_invalid_attr():
with pytest.raises(ImportFromStringError) as exc_info:
import_from_string("tempfile:attr_does_not_exist")
expected = 'Attribute "attr_does_not_exist" not found in module "tempfile".'
assert expected in str(exc_info.value)
def test_internal_import_error():
with pytest.raises(ImportError):
import_from_string("tests.importer.raise_import_error:myattr")
def test_valid_import():
instance = import_from_string("tempfile:TemporaryFile")
from tempfile import TemporaryFile
assert instance == TemporaryFile
def __init__(
self, url: typing.Union[str, DatabaseURL], force_rollback: bool = False
):
self.url = DatabaseURL(url)
self.force_rollback = force_rollback
backend_str = self.backends[self.url.dialect]
backend_cls = import_from_string(backend_str)
assert issubclass(backend_cls, DatabaseBackend)
self.backend = backend_cls(self.url)
self.is_connected = False
self.rollback_transaction = None
self.session_context = ContextVar("session_context") # type: ContextVar
def __init__(
self,
url: typing.Union[str, "DatabaseURL"],
*,
force_rollback: bool = False,
**options: typing.Any,
):
self.url = DatabaseURL(url)
self.options = options
self.is_connected = False
self._force_rollback = force_rollback
backend_str = self.SUPPORTED_BACKENDS[self.url.dialect]
backend_cls = import_from_string(backend_str)
assert issubclass(backend_cls, DatabaseBackend)
self._backend = backend_cls(self.url, **self.options)
# Connections are stored as task-local state.
self._connection_context = ContextVar("connection_context") # type: ContextVar
# When `force_rollback=True` is used, we use a single global
# connection, within a transaction that always rolls back.
self._global_connection: typing.Optional[Connection] = None
self._global_transaction: typing.Optional[Transaction] = None