databases
databases copied to clipboard
Published
20 hours ago •
encode
encode
0.5.5: pytest is failing and some warnings
I'm trying to package your module as an rpm package. So I'm using the typical PEP517 based build, install and test cycle used on building packages from non-root account.
-
python3 -sBm build -w --no-isolation - because I'm calling
buildwith--no-isolationI'm using during all processes only locally installed modules - install .whl file in </install/prefix>
- run pytest with PYTHONPATH pointing to sitearch and sitelib inside </install/prefix>
Generally this ticket is about databases tsting methodology 😄
Here is raw pytest output:
+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.5.5-2.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.5.5-2.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra
=========================================================================== test session starts ============================================================================
platform linux -- Python 3.8.12, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.5.5
plugins: forked-1.4.0, xdist-2.5.0, hypothesis-6.36.0, anyio-3.3.4, cov-3.0.0
collected 49 items / 2 errors / 47 selected
================================================================================== ERRORS ==================================================================================
____________________________________________________________ ERROR collecting tests/test_connection_options.py _____________________________________________________________
ImportError while importing test module '/home/tkloczko/rpmbuild/BUILD/databases-0.5.5/tests/test_connection_options.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/lib64/python3.8/importlib/__init__.py:127: in import_module
return _bootstrap._gcd_import(name[level:], package, level)
tests/test_connection_options.py:11: in <module>
from tests.test_databases import DATABASE_URLS, async_adapter
E ModuleNotFoundError: No module named 'tests'
________________________________________________________________ ERROR collecting tests/test_integration.py ________________________________________________________________
ImportError while importing test module '/home/tkloczko/rpmbuild/BUILD/databases-0.5.5/tests/test_integration.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/lib64/python3.8/importlib/__init__.py:127: in import_module
return _bootstrap._gcd_import(name[level:], package, level)
tests/test_integration.py:8: in <module>
from tests.test_databases import DATABASE_URLS, mysql_versions
E ModuleNotFoundError: No module named 'tests'
========================================================================= short test summary info ==========================================================================
ERROR tests/test_connection_options.py
ERROR tests/test_integration.py
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Interrupted: 2 errors during collection !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
============================================================================ 2 errors in 0.70s =============================================================================
In this case it cannot be tests module from source tree because there is no tests/init.py file (or something is issing).
After ignore those two files pytest show one warning
+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.5.5-2.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.5.5-2.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra --ignore tests/test_connection_options.py --ignore tests/test_integration.py
=========================================================================== test session starts ============================================================================
platform linux -- Python 3.8.12, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.5.5
plugins: forked-1.4.0, xdist-2.5.0, hypothesis-6.36.0, anyio-3.3.4, cov-3.0.0
collected 49 items
tests/test_database_url.py ...... [ 12%]
tests/test_databases.py ............s...............ss.s...s.. [ 89%]
tests/test_importer.py ..... [100%]
============================================================================= warnings summary =============================================================================
tests/test_databases.py::test_decimal_field[sqlite:///testsuite]
/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.5.5-2.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:95: SAWarning: Dialect sqlite+pysqlite does *not* support Decimal objects natively, and SQLAlchemy must convert from floating point - rounding errors and other issues may occur. Please consider storing Decimal numbers as strings or integers on this platform for lossless storage.
metadata = CursorResultMetaData(context, cursor.description)
-- Docs: https://docs.pytest.org/en/stable/warnings.html
========================================================================= short test summary info ==========================================================================
SKIPPED [1] tests/test_databases.py:523: Test (currently) only supports asyncpg
SKIPPED [1] tests/test_databases.py:1014: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1036: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1087: SQLite interface does not work with temporary tables.
SKIPPED [1] tests/test_databases.py:1158: Test is only for asyncpg
================================================================= 44 passed, 5 skipped, 1 warning in 1.55s =================================================================
Hi,
I think for the first one that sounds reasonable, we could turn the tests into a package.
But for the second point, I'm not sure what needs to be done, the warnings are clear and telling that you probably are running tests with SQLite. You might want to ignore the warnings?
Hi,
I think for the first one that sounds reasonable, we could turn the
testsinto a package. But for the second point, I'm not sure what needs to be done, the warnings are clear and telling that you probably are running tests with SQLite. You might want to ignore the warnings?
Please dond't get me wrong .. I'm not asking to fix that but more I'm flagging some issue 😄
tests module .. if that module would have _init.py file it would be possible to use it by run pytest --pargs tests
gentle ping .. just tested new 0.6.0 and looks like issue still is around 😋
Updated pyetst output with two files in --ignore list
platform linux -- Python 3.8.13, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.6.0
plugins: anyio-3.5.0, cov-3.0.0
collected 50 items
tests/test_database_url.py ...... [ 12%]
tests/test_databases.py .............s...............ss.s..s... [ 90%]
tests/test_importer.py ..... [100%]
============================================================================= warnings summary =============================================================================
tests/test_databases.py::test_decimal_field[sqlite:///testsuite]
/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.6.0-2.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:100: SAWarning: Dialect sqlite+pysqlite does *not* support Decimal objects natively, and SQLAlchemy must convert from floating point - rounding errors and other issues may occur. Please consider storing Decimal numbers as strings or integers on this platform for lossless storage.
metadata = CursorResultMetaData(context, cursor.description)
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
========================================================================= short test summary info ==========================================================================
SKIPPED [1] tests/test_databases.py:521: Test (currently) only supports asyncpg
SKIPPED [1] tests/test_databases.py:1012: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1034: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1085: SQLite interface does not work with temporary tables.
SKIPPED [1] tests/test_databases.py:1137: Test is only for asyncpg
================================================================= 45 passed, 5 skipped, 1 warning in 3.32s =================================================================
I forgot about this completely, what would be the fix here? this is the expected behaviour, I'm not sure about the packaging.
Packaging has nothing to do with that issue. That part about packaging is only about context that exactly the same approach/methodology to build and test is working already in case +950 of other packaged python modules.
Just made quick hack
--- a/tests/test_connection_options.py
+++ b/tests/test_connection_options.py
@@ -8,7 +8,7 @@
from databases.backends.aiopg import AiopgBackend
from databases.backends.postgres import PostgresBackend
from databases.core import DatabaseURL
-from tests.test_databases import DATABASE_URLS, async_adapter
+from test_databases import DATABASE_URLS, async_adapter
if sys.version_info >= (3, 7): # pragma: no cover
from databases.backends.asyncmy import AsyncMyBackend
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -5,7 +5,7 @@
from starlette.testclient import TestClient
from databases import Database, DatabaseURL
-from tests.test_databases import DATABASE_URLS
+from test_databases import DATABASE_URLS
metadata = sqlalchemy.MetaData()
than with above:
usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra -m 'not network'
==================================================================================== test session starts ====================================================================================
platform linux -- Python 3.8.16, pytest-7.3.0, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.7.0
plugins: anyio-3.6.2, cov-4.0.0
collected 51 items / 1 error
========================================================================================== ERRORS ===========================================================================================
_____________________________________________________________________ ERROR collecting tests/test_connection_options.py _____________________________________________________________________
ImportError while importing test module '/home/tkloczko/rpmbuild/BUILD/databases-0.7.0/tests/test_connection_options.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/lib/python3.8/site-packages/aiopg/sa/engine.py:11: in <module>
from sqlalchemy.dialects.postgresql.psycopg2 import (
E ImportError: cannot import name 'PGCompiler_psycopg2' from 'sqlalchemy.dialects.postgresql.psycopg2' (/usr/lib64/python3.8/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py)
During handling of the above exception, another exception occurred:
/usr/lib64/python3.8/importlib/__init__.py:127: in import_module
return _bootstrap._gcd_import(name[level:], package, level)
tests/test_connection_options.py:8: in <module>
from databases.backends.aiopg import AiopgBackend
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/aiopg.py:8: in <module>
from aiopg.sa.engine import APGCompiler_psycopg2
/usr/lib/python3.8/site-packages/aiopg/sa/__init__.py:3: in <module>
from .engine import Engine, create_engine
/usr/lib/python3.8/site-packages/aiopg/sa/engine.py:16: in <module>
raise ImportError("aiopg.sa requires sqlalchemy")
E ImportError: aiopg.sa requires sqlalchemy
================================================================================== short test summary info ==================================================================================
ERROR tests/test_connection_options.py
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
===================================================================================== 1 error in 0.52s ======================================================================================
and that with --ignore that files it exposes more issues
+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra -m 'not network' --ignore tests/test_connection_options.py
==================================================================================== test session starts ====================================================================================
platform linux -- Python 3.8.16, pytest-7.3.0, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.7.0
plugins: anyio-3.6.2, cov-4.0.0
collected 51 items
tests/test_database_url.py ...... [ 11%]
tests/test_databases.py FF...FF..F...s....FFFF.......ss.sFFs... [ 88%]
tests/test_importer.py ..... [ 98%]
tests/test_integration.py F [100%]
========================================================================================= FAILURES ==========================================================================================
_____________________________________________________________________________ test_queries[sqlite:///testsuite] _____________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (using SQLAlchemy core).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = notes.insert()
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = notes.select()
results = await database.fetch_all(query=query)
assert len(results) == 3
> assert results[0]["text"] == "example1"
tests/test_databases.py:156:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________________________________________________________________ test_queries_raw[sqlite:///testsuite] ___________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries_raw(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (raw queries).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = "SELECT * FROM notes WHERE completed = :completed"
results = await database.fetch_all(query=query, values={"completed": True})
assert len(results) == 2
> assert results[0]["text"] == "example1"
tests/test_databases.py:232:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
________________________________________________________________ test_results_support_mapping_interface[sqlite:///testsuite] ________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_mapping_interface(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch_all()
query = notes.select()
results = await database.fetch_all(query=query)
> results_as_dicts = [dict(item) for item in results]
tests/test_databases.py:325:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f0d7b92a580>
> results_as_dicts = [dict(item) for item in results]
E TypeError: cannot convert dictionary update sequence element #0 to a sequence
tests/test_databases.py:325: TypeError
________________________________________________________________ test_results_support_column_reference[sqlite:///testsuite] _________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_column_reference(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
today = datetime.date.today()
# execute()
query = articles.insert()
values = {"title": "Hello, world Article", "published": now}
await database.execute(query, values)
query = custom_date.insert()
values = {"title": "Hello, world Custom", "published": today}
await database.execute(query, values)
# fetch_all()
> query = sqlalchemy.select([articles, custom_date])
tests/test_databases.py:357:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/_selectable_constructors.py:493: in select
return Select(*entities)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5161: in __init__
self._raw_columns = [
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5162: in <listcomp>
coercions.expect(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:413: in expect
resolved = impl._literal_coercion(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:652: in _literal_coercion
self._raise_for_expected(element, argname)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:1143: in _raise_for_expected
return super()._raise_for_expected(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:711: in _raise_for_expected
super()._raise_for_expected(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.sql.coercions.ColumnsClauseImpl object at 0x7f0d7ca54240>
element = [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('ti...itle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]
argname = None, resolved = None
advice = "Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, ...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
code = None, err = None, kw = {}
got = "[Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('t...tle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]"
msg = "Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
def _raise_for_expected(
self,
element: Any,
argname: Optional[str] = None,
resolved: Optional[Any] = None,
advice: Optional[str] = None,
code: Optional[str] = None,
err: Optional[Exception] = None,
**kw: Any,
) -> NoReturn:
if resolved is not None and resolved is not element:
got = "%r object resolved from %r object" % (resolved, element)
else:
got = repr(element)
if argname:
msg = "%s expected for argument %r; got %s." % (
self.name,
argname,
got,
)
else:
msg = "%s expected, got %s." % (self.name, got)
if advice:
msg += " " + advice
> raise exc.ArgumentError(msg, code=code) from err
E sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]. Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:536: ArgumentError
_______________________________________________________________________ test_execute_return_val[sqlite:///testsuite] ________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_execute_return_val(database_url):
"""
Test using return value from `execute()` to get an inserted primary key.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
query = notes.insert()
values = {"text": "example1", "completed": True}
pk = await database.execute(query, values)
assert isinstance(pk, int)
# Apparently for `aiopg` it's OID that will always 0 in this case
# As it's only one action within this cursor life cycle
# It's recommended to use the `RETURNING` clause
# For obtaining the record id
if database.url.scheme == "postgresql+aiopg":
assert pk == 0
else:
query = notes.select().where(notes.c.id == pk)
result = await database.fetch_one(query)
> assert result["text"] == "example1"
tests/test_databases.py:418:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
_________________________________________________________________________ test_datetime_field[sqlite:///testsuite] __________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_datetime_field(database_url):
"""
Test DataTime columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
# execute()
query = articles.insert()
values = {"title": "Hello, world", "published": now}
await database.execute(query, values)
# fetch_all()
query = articles.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["title"] == "Hello, world"
tests/test_databases.py:643:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
__________________________________________________________________________ test_decimal_field[sqlite:///testsuite] __________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_decimal_field(database_url):
"""
Test Decimal (NUMERIC) columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
price = decimal.Decimal("0.700000000000001")
# execute()
query = prices.insert()
values = {"price": price}
await database.execute(query, values)
# fetch_all()
query = prices.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
if database_url.startswith("sqlite"):
# aiosqlite does not support native decimals --> a roud-off error is expected
> assert results[0]["price"] == pytest.approx(price)
tests/test_databases.py:669:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________________________________________________________________ test_json_field[sqlite:///testsuite] ____________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_json_field(database_url):
"""
Test JSON columns, to ensure correct cross-database support.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
data = {"text": "hello", "boolean": True, "int": 1}
values = {"data": data}
query = session.insert()
await database.execute(query, values)
# fetch_all()
query = session.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["data"] == {"text": "hello", "boolean": True, "int": 1}
tests/test_databases.py:693:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
__________________________________________________________________________ test_custom_field[sqlite:///testsuite] ___________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_custom_field(database_url):
"""
Test custom column types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
today = datetime.date.today()
# execute()
query = custom_date.insert()
values = {"title": "Hello, world", "published": today}
await database.execute(query, values)
# fetch_all()
query = custom_date.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["title"] == "Hello, world"
tests/test_databases.py:717:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________________________________________________________ test_column_names[select_query0-sqlite:///testsuite] ____________________________________________________________________
database_url = 'sqlite:///testsuite', select_query = <sqlalchemy.sql.selectable.Select object at 0x7f0d7c4aa190>
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
results = await database.fetch_all(query=select_query)
assert len(results) == 1
assert sorted(results[0]._mapping.keys()) == ["completed", "id", "text"]
> assert results[0]["text"] == "example1"
tests/test_databases.py:1074:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
________________________________________________________________ test_column_names[SELECT * FROM notes-sqlite:///testsuite] _________________________________________________________________
database_url = 'sqlite:///testsuite', select_query = 'SELECT * FROM notes'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
results = await database.fetch_all(query=select_query)
assert len(results) == 1
assert sorted(results[0]._mapping.keys()) == ["completed", "id", "text"]
> assert results[0]["text"] == "example1"
tests/test_databases.py:1074:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________________________________________________________________ test_integration[sqlite:///testsuite] ___________________________________________________________________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
def test_integration(database_url):
app = get_app(database_url)
with TestClient(app) as client:
response = client.post("/notes", json={"text": "example", "completed": True})
assert response.status_code == 200
assert response.json() == {"text": "example", "completed": True}
> response = client.get("/notes")
tests/test_integration.py:95:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib/python3.8/site-packages/requests/sessions.py:600: in get
return self.request("GET", url, **kwargs)
/usr/lib/python3.8/site-packages/starlette/testclient.py:473: in request
return super().request(
/usr/lib/python3.8/site-packages/requests/sessions.py:587: in request
resp = self.send(prep, **send_kwargs)
/usr/lib/python3.8/site-packages/requests/sessions.py:701: in send
r = adapter.send(request, **kwargs)
/usr/lib/python3.8/site-packages/starlette/testclient.py:267: in send
raise exc
/usr/lib/python3.8/site-packages/starlette/testclient.py:264: in send
portal.call(self.app, scope, receive, send)
/usr/lib/python3.8/site-packages/anyio/from_thread.py:283: in call
return cast(T_Retval, self.start_task_soon(func, *args).result())
/usr/lib64/python3.8/concurrent/futures/_base.py:444: in result
return self.__get_result()
/usr/lib64/python3.8/concurrent/futures/_base.py:389: in __get_result
raise self._exception
/usr/lib/python3.8/site-packages/anyio/from_thread.py:219: in _call_func
retval = await retval
/usr/lib/python3.8/site-packages/starlette/applications.py:124: in __call__
await self.middleware_stack(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/middleware/errors.py:184: in __call__
raise exc
/usr/lib/python3.8/site-packages/starlette/middleware/errors.py:162: in __call__
await self.app(scope, receive, _send)
/usr/lib/python3.8/site-packages/starlette/middleware/exceptions.py:75: in __call__
raise exc
/usr/lib/python3.8/site-packages/starlette/middleware/exceptions.py:64: in __call__
await self.app(scope, receive, sender)
/usr/lib/python3.8/site-packages/starlette/routing.py:680: in __call__
await route.handle(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/routing.py:275: in handle
await self.app(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/routing.py:65: in app
response = await func(request)
tests/test_integration.py:70: in list_notes
content = [
tests/test_integration.py:71: in <listcomp>
{"text": result["text"], "completed": result["completed"]}
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
================================================================================== short test summary info ==================================================================================
SKIPPED [1] tests/test_databases.py:490: Test (currently) only supports asyncpg
SKIPPED [1] tests/test_databases.py:966: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:988: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1038: SQLite interface does not work with temporary tables.
SKIPPED [1] tests/test_databases.py:1089: Test is only for asyncpg
FAILED tests/test_databases.py::test_queries[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_queries_raw[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_results_support_mapping_interface[sqlite:///testsuite] - TypeError: cannot convert dictionary update sequence element #0 to a sequence
FAILED tests/test_databases.py::test_results_support_column_reference[sqlite:///testsuite] - sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, pri...
FAILED tests/test_databases.py::test_execute_return_val[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_datetime_field[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_decimal_field[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_json_field[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_custom_field[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_column_names[select_query0-sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_databases.py::test_column_names[SELECT * FROM notes-sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
FAILED tests/test_integration.py::test_integration[sqlite:///testsuite] - TypeError: tuple indices must be integers or slices, not str
========================================================================= 12 failed, 34 passed, 5 skipped in 3.25s ==========================================================================
The same result without above patch and only with --ignore list
+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra -m 'not network' --ignore tests/test_connection_options.py --ignore tests/test_integration.py
============================= test session starts ==============================
platform linux -- Python 3.8.16, pytest-7.3.0, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.7.0
plugins: anyio-3.6.2, cov-4.0.0
collected 50 items
tests/test_database_url.py ...... [ 12%]
tests/test_databases.py FF...FF..F...s....FFFF.......ss.sFFs... [ 90%]
tests/test_importer.py ..... [100%]
=================================== FAILURES ===================================
______________________ test_queries[sqlite:///testsuite] _______________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (using SQLAlchemy core).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = notes.insert()
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = notes.select()
results = await database.fetch_all(query=query)
assert len(results) == 3
> assert results[0]["text"] == "example1"
tests/test_databases.py:156:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
____________________ test_queries_raw[sqlite:///testsuite] _____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries_raw(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (raw queries).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = "SELECT * FROM notes WHERE completed = :completed"
results = await database.fetch_all(query=query, values={"completed": True})
assert len(results) == 2
> assert results[0]["text"] == "example1"
tests/test_databases.py:232:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
_________ test_results_support_mapping_interface[sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_mapping_interface(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch_all()
query = notes.select()
results = await database.fetch_all(query=query)
> results_as_dicts = [dict(item) for item in results]
tests/test_databases.py:325:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f19df534310>
> results_as_dicts = [dict(item) for item in results]
E TypeError: cannot convert dictionary update sequence element #0 to a sequence
tests/test_databases.py:325: TypeError
__________ test_results_support_column_reference[sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_column_reference(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
today = datetime.date.today()
# execute()
query = articles.insert()
values = {"title": "Hello, world Article", "published": now}
await database.execute(query, values)
query = custom_date.insert()
values = {"title": "Hello, world Custom", "published": today}
await database.execute(query, values)
# fetch_all()
> query = sqlalchemy.select([articles, custom_date])
tests/test_databases.py:357:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/_selectable_constructors.py:493: in select
return Select(*entities)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5161: in __init__
self._raw_columns = [
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5162: in <listcomp>
coercions.expect(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:413: in expect
resolved = impl._literal_coercion(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:652: in _literal_coercion
self._raise_for_expected(element, argname)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:1143: in _raise_for_expected
return super()._raise_for_expected(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:711: in _raise_for_expected
super()._raise_for_expected(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.sql.coercions.ColumnsClauseImpl object at 0x7f19dfb5d600>
element = [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('ti...itle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]
argname = None, resolved = None
advice = "Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, ...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
code = None, err = None, kw = {}
got = "[Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('t...tle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]"
msg = "Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
def _raise_for_expected(
self,
element: Any,
argname: Optional[str] = None,
resolved: Optional[Any] = None,
advice: Optional[str] = None,
code: Optional[str] = None,
err: Optional[Exception] = None,
**kw: Any,
) -> NoReturn:
if resolved is not None and resolved is not element:
got = "%r object resolved from %r object" % (resolved, element)
else:
got = repr(element)
if argname:
msg = "%s expected for argument %r; got %s." % (
self.name,
argname,
got,
)
else:
msg = "%s expected, got %s." % (self.name, got)
if advice:
msg += " " + advice
> raise exc.ArgumentError(msg, code=code) from err
E sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]. Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:536: ArgumentError
_________________ test_execute_return_val[sqlite:///testsuite] _________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_execute_return_val(database_url):
"""
Test using return value from `execute()` to get an inserted primary key.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
query = notes.insert()
values = {"text": "example1", "completed": True}
pk = await database.execute(query, values)
assert isinstance(pk, int)
# Apparently for `aiopg` it's OID that will always 0 in this case
# As it's only one action within this cursor life cycle
# It's recommended to use the `RETURNING` clause
# For obtaining the record id
if database.url.scheme == "postgresql+aiopg":
assert pk == 0
else:
query = notes.select().where(notes.c.id == pk)
result = await database.fetch_one(query)
> assert result["text"] == "example1"
tests/test_databases.py:418:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________ test_datetime_field[sqlite:///testsuite] ___________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_datetime_field(database_url):
"""
Test DataTime columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
# execute()
query = articles.insert()
values = {"title": "Hello, world", "published": now}
await database.execute(query, values)
# fetch_all()
query = articles.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["title"] == "Hello, world"
tests/test_databases.py:643:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
___________________ test_decimal_field[sqlite:///testsuite] ____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_decimal_field(database_url):
"""
Test Decimal (NUMERIC) columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
price = decimal.Decimal("0.700000000000001")
# execute()
query = prices.insert()
values = {"price": price}
await database.execute(query, values)
# fetch_all()
query = prices.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
if database_url.startswith("sqlite"):
# aiosqlite does not support native decimals --> a roud-off error is expected
> assert results[0]["price"] == pytest.approx(price)
tests/test_databases.py:669:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
_____________________ test_json_field[sqlite:///testsuite] _____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_json_field(database_url):
"""
Test JSON columns, to ensure correct cross-database support.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
data = {"text": "hello", "boolean": True, "int": 1}
values = {"data": data}
query = session.insert()
await database.execute(query, values)
# fetch_all()
query = session.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["data"] == {"text": "hello", "boolean": True, "int": 1}
tests/test_databases.py:693:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
____________________ test_custom_field[sqlite:///testsuite] ____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_custom_field(database_url):
"""
Test custom column types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
today = datetime.date.today()
# execute()
query = custom_date.insert()
values = {"title": "Hello, world", "published": today}
await database.execute(query, values)
# fetch_all()
query = custom_date.select()
results = await database.fetch_all(query=query)
assert len(results) == 1
> assert results[0]["title"] == "Hello, world"
tests/test_databases.py:717:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
_____________ test_column_names[select_query0-sqlite:///testsuite] _____________
database_url = 'sqlite:///testsuite'
select_query = <sqlalchemy.sql.selectable.Select object at 0x7f19df5b50a0>
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
results = await database.fetch_all(query=select_query)
assert len(results) == 1
assert sorted(results[0]._mapping.keys()) == ["completed", "id", "text"]
> assert results[0]["text"] == "example1"
tests/test_databases.py:1074:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
__________ test_column_names[SELECT * FROM notes-sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite', select_query = 'SELECT * FROM notes'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
results = await database.fetch_all(query=select_query)
assert len(results) == 1
assert sorted(results[0]._mapping.keys()) == ["completed", "id", "text"]
> assert results[0]["text"] == "example1"
tests/test_databases.py:1074:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> ???
E TypeError: tuple indices must be integers or slices, not str
lib/sqlalchemy/cyextension/resultproxy.pyx:68: TypeError
=========================== short test summary info ============================
SKIPPED [1] tests/test_databases.py:490: Test (currently) only supports asyncpg
SKIPPED [1] tests/test_databases.py:966: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:988: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1038: SQLite interface does not work with temporary tables.
SKIPPED [1] tests/test_databases.py:1089: Test is only for asyncpg
FAILED tests/test_databases.py::test_queries[sqlite:///testsuite] - TypeError...
FAILED tests/test_databases.py::test_queries_raw[sqlite:///testsuite] - TypeE...
FAILED tests/test_databases.py::test_results_support_mapping_interface[sqlite:///testsuite]
FAILED tests/test_databases.py::test_results_support_column_reference[sqlite:///testsuite]
FAILED tests/test_databases.py::test_execute_return_val[sqlite:///testsuite]
FAILED tests/test_databases.py::test_datetime_field[sqlite:///testsuite] - Ty...
FAILED tests/test_databases.py::test_decimal_field[sqlite:///testsuite] - Typ...
FAILED tests/test_databases.py::test_json_field[sqlite:///testsuite] - TypeEr...
FAILED tests/test_databases.py::test_custom_field[sqlite:///testsuite] - Type...
FAILED tests/test_databases.py::test_column_names[select_query0-sqlite:///testsuite]
FAILED tests/test_databases.py::test_column_names[SELECT * FROM notes-sqlite:///testsuite]
=================== 11 failed, 34 passed, 5 skipped in 2.77s ===================
Package Version
------------------ -------
aiopg 1.4.0
anyio 3.6.2
async-timeout 4.0.2
asyncpg 0.27.0
build 0.10.0
charset-normalizer 3.1.0
coverage 7.2.3
distro 1.8.0
exceptiongroup 1.0.0
gpg 1.19.0
greenlet 1.1.3
idna 3.4
iniconfig 2.0.0
installer 0.7.0
libcomps 0.1.19
packaging 23.0
pluggy 1.0.0
psycopg2 2.9.5
PyMySQL 1.0.3
pyproject_hooks 1.0.0
pytest 7.3.0
pytest-cov 4.0.0
python-dateutil 2.8.2
requests 2.28.2
setuptools 65.6.3
six 1.16.0
sniffio 1.2.0
SQLAlchemy 2.0.9
starlette 0.20.4
tomli 2.0.1
typing_extensions 4.5.0
urllib3 1.26.15
wheel 0.40.0
Just FTR: updated pytest output for 0.7.0
+ PYTHONPATH=/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib64/python3.8/site-packages:/home/tkloczko/rpmbuild/BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages
+ /usr/bin/pytest -ra -m 'not network' --ignore tests/test_connection_options.py
============================= test session starts ==============================
platform linux -- Python 3.8.16, pytest-7.3.1, pluggy-1.0.0
rootdir: /home/tkloczko/rpmbuild/BUILD/databases-0.7.0
plugins: anyio-3.6.2
collected 51 items
tests/test_database_url.py ...... [ 11%]
tests/test_databases.py FF...FFF.F..Fs.F.FFFFFFF.....ssFsFFs.FF [ 88%]
tests/test_importer.py ..... [ 98%]
tests/test_integration.py F [100%]
=================================== FAILURES ===================================
______________________ test_queries[sqlite:///testsuite] _______________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (using SQLAlchemy core).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = notes.insert()
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:153:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f4517273610>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
____________________ test_queries_raw[sqlite:///testsuite] _____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_queries_raw(database_url):
"""
Test that the basic `execute()`, `execute_many()`, `fetch_all()``, and
`fetch_one()` interfaces are all supported (raw queries).
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# execute_many()
query = "INSERT INTO notes(text, completed) VALUES (:text, :completed)"
values = [
{"text": "example2", "completed": False},
{"text": "example3", "completed": True},
]
await database.execute_many(query, values)
# fetch_all()
query = "SELECT * FROM notes WHERE completed = :completed"
> results = await database.fetch_all(query=query, values={"completed": True})
tests/test_databases.py:230:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f4516931f10>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
_________ test_results_support_mapping_interface[sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_mapping_interface(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch_all()
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:324:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45168df3a0>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
__________ test_results_support_column_reference[sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_results_support_column_reference(database_url):
"""
Casting results to a dict should work, since the interface defines them
as supporting the mapping interface.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
today = datetime.date.today()
# execute()
query = articles.insert()
values = {"title": "Hello, world Article", "published": now}
await database.execute(query, values)
query = custom_date.insert()
values = {"title": "Hello, world Custom", "published": today}
await database.execute(query, values)
# fetch_all()
> query = sqlalchemy.select([articles, custom_date])
tests/test_databases.py:357:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/_selectable_constructors.py:489: in select
return Select(*entities)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5139: in __init__
self._raw_columns = [
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/selectable.py:5140: in <listcomp>
coercions.expect(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:413: in expect
resolved = impl._literal_coercion(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:651: in _literal_coercion
self._raise_for_expected(element, argname)
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:1142: in _raise_for_expected
return super()._raise_for_expected(
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:710: in _raise_for_expected
super()._raise_for_expected(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <sqlalchemy.sql.coercions.ColumnsClauseImpl object at 0x7f4517cfee80>
element = [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('ti...itle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]
argname = None, resolved = None
advice = "Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, ...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
code = None, err = None, kw = {}
got = "[Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('t...tle', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]"
msg = "Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('...le', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?"
def _raise_for_expected(
self,
element: Any,
argname: Optional[str] = None,
resolved: Optional[Any] = None,
advice: Optional[str] = None,
code: Optional[str] = None,
err: Optional[Exception] = None,
**kw: Any,
) -> NoReturn:
if resolved is not None and resolved is not element:
got = "%r object resolved from %r object" % (resolved, element)
else:
got = repr(element)
if argname:
msg = "%s expected for argument %r; got %s." % (
self.name,
argname,
got,
)
else:
msg = "%s expected, got %s." % (self.name, got)
if advice:
msg += " " + advice
> raise exc.ArgumentError(msg, code=code) from err
E sqlalchemy.exc.ArgumentError: Column expression, FROM clause, or other columns clause element expected, got [Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None)]. Did you mean to say select(Table('articles', MetaData(), Column('id', Integer(), table=<articles>, primary_key=True, nullable=False), Column('title', String(length=100), table=<articles>), Column('published', DateTime(), table=<articles>), schema=None), Table('custom_date', MetaData(), Column('id', Integer(), table=<custom_date>, primary_key=True, nullable=False), Column('title', String(length=100), table=<custom_date>), Column('published', MyEpochType(), table=<custom_date>), schema=None))?
/usr/lib64/python3.8/site-packages/sqlalchemy/sql/coercions.py:535: ArgumentError
________ test_result_values_allow_duplicate_names[sqlite:///testsuite] _________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_result_values_allow_duplicate_names(database_url):
"""
The values of a result should respect when two columns are selected
with the same name.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
query = "SELECT 1 AS id, 2 AS id"
> row = await database.fetch_one(query=query)
tests/test_databases.py:376:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:152: in fetch_one
return await connection.fetch_one(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:269: in fetch_one
return await self._connection.fetch_one(built_query)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <databases.backends.sqlite.SQLiteConnection object at 0x7f4516850700>
query = <sqlalchemy.sql.elements.TextClause object at 0x7f4516850b50>
async def fetch_one(self, query: ClauseElement) -> typing.Optional[Record]:
assert self._connection is not None, "Connection is not acquired"
query_str, args, context = self._compile(query)
async with self._connection.execute(query_str, args) as cursor:
row = await cursor.fetchone()
if row is None:
return None
metadata = CursorResultMetaData(context, cursor.description)
return Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:125: AttributeError
_________________ test_execute_return_val[sqlite:///testsuite] _________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_execute_return_val(database_url):
"""
Test using return value from `execute()` to get an inserted primary key.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
query = notes.insert()
values = {"text": "example1", "completed": True}
pk = await database.execute(query, values)
assert isinstance(pk, int)
# Apparently for `aiopg` it's OID that will always 0 in this case
# As it's only one action within this cursor life cycle
# It's recommended to use the `RETURNING` clause
# For obtaining the record id
if database.url.scheme == "postgresql+aiopg":
assert pk == 0
else:
query = notes.select().where(notes.c.id == pk)
> result = await database.fetch_one(query)
tests/test_databases.py:417:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:152: in fetch_one
return await connection.fetch_one(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:269: in fetch_one
return await self._connection.fetch_one(built_query)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <databases.backends.sqlite.SQLiteConnection object at 0x7f45167fc700>
query = <sqlalchemy.sql.selectable.Select object at 0x7f45167fcd00>
async def fetch_one(self, query: ClauseElement) -> typing.Optional[Record]:
assert self._connection is not None, "Connection is not acquired"
query_str, args, context = self._compile(query)
async with self._connection.execute(query_str, args) as cursor:
row = await cursor.fetchone()
if row is None:
return None
metadata = CursorResultMetaData(context, cursor.description)
return Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:125: AttributeError
_________________ test_transaction_commit[sqlite:///testsuite] _________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_transaction_commit(database_url):
"""
Ensure that transaction commit is supported.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
async with database.transaction():
query = notes.insert().values(text="example1", completed=True)
await database.execute(query)
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:476:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f451574afd0>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
____________ test_transaction_commit_low_level[sqlite:///testsuite] ____________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_transaction_commit_low_level(database_url):
"""
Ensure that an explicit `await transaction.commit()` is supported.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
transaction = await database.transaction()
try:
query = notes.insert().values(text="example1", completed=True)
await database.execute(query)
except: # pragma: no cover
await transaction.rollback()
else:
await transaction.commit()
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:565:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f451678ac70>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
_______________ test_transaction_decorator[sqlite:///testsuite] ________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_transaction_decorator(database_url):
"""
Ensure that @database.transaction() is supported.
"""
database = Database(database_url, force_rollback=True)
@database.transaction()
async def insert_data(raise_exception):
query = notes.insert().values(text="example", completed=True)
await database.execute(query)
if raise_exception:
raise RuntimeError()
async with database:
with pytest.raises(RuntimeError):
await insert_data(raise_exception=True)
query = notes.select()
results = await database.fetch_all(query=query)
assert len(results) == 0
await insert_data(raise_exception=False)
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:619:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f4515690340>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
___________________ test_datetime_field[sqlite:///testsuite] ___________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_datetime_field(database_url):
"""
Test DataTime columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
now = datetime.datetime.now().replace(microsecond=0)
# execute()
query = articles.insert()
values = {"title": "Hello, world", "published": now}
await database.execute(query, values)
# fetch_all()
query = articles.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:641:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f451542a610>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
___________________ test_decimal_field[sqlite:///testsuite] ____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_decimal_field(database_url):
"""
Test Decimal (NUMERIC) columns, to ensure records are coerced to/from proper Python types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
price = decimal.Decimal("0.700000000000001")
# execute()
query = prices.insert()
values = {"price": price}
await database.execute(query, values)
# fetch_all()
query = prices.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:665:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f451574ac10>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
_____________________ test_json_field[sqlite:///testsuite] _____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_json_field(database_url):
"""
Test JSON columns, to ensure correct cross-database support.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# execute()
data = {"text": "hello", "boolean": True, "int": 1}
values = {"data": data}
query = session.insert()
await database.execute(query, values)
# fetch_all()
query = session.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:691:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45167e6e50>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
____________________ test_custom_field[sqlite:///testsuite] ____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_custom_field(database_url):
"""
Test custom column types.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
today = datetime.date.today()
# execute()
query = custom_date.insert()
values = {"title": "Hello, world", "published": today}
await database.execute(query, values)
# fetch_all()
query = custom_date.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:715:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45167d3c10>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
_______________ test_connections_isolation[sqlite:///testsuite] ________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_connections_isolation(database_url):
"""
Ensure that changes are visible between different connections.
To check this we have to not create a transaction, so that
each query ends up on a different connection from the pool.
"""
async with Database(database_url) as database:
try:
query = notes.insert().values(text="example1", completed=True)
await database.execute(query)
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:736:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f451696e430>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
_____________ test_commit_on_root_transaction[sqlite:///testsuite] _____________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_commit_on_root_transaction(database_url):
"""
Because our tests are generally wrapped in rollback-islation, they
don't have coverage for commiting the root transaction.
Deal with this here, and delete the records rather than rolling back.
"""
async with Database(database_url) as database:
try:
async with database.transaction():
query = notes.insert().values(text="example1", completed=True)
await database.execute(query)
query = notes.select()
> results = await database.fetch_all(query=query)
tests/test_databases.py:760:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45169225b0>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
______ test_iterate_outside_transaction_with_values[sqlite:///testsuite] _______
database_url = DatabaseURL('sqlite:///testsuite')
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_iterate_outside_transaction_with_values(database_url):
"""
Ensure `iterate()` works even without a transaction on all drivers.
The asyncpg driver relies on server-side cursors without hold
for iteration, which requires a transaction to be created.
This is mentionned in both their documentation and their test suite.
"""
database_url = DatabaseURL(database_url)
if database_url.dialect == "mysql":
pytest.skip("MySQL does not support `FROM (VALUES ...)` (F641)")
async with Database(database_url) as database:
query = "SELECT * FROM (VALUES (1), (2), (3), (4), (5)) as t"
iterate_results = []
> async for result in database.iterate(query=query):
tests/test_databases.py:1022:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:183: in iterate
async for record in connection.iterate(query, values):
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:305: in iterate
async for record in self._connection.iterate(built_query):
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <databases.backends.sqlite.SQLiteConnection object at 0x7f45156b3b20>
query = <sqlalchemy.sql.elements.TextClause object at 0x7f451534d820>
async def iterate(
self, query: ClauseElement
) -> typing.AsyncGenerator[typing.Any, None]:
assert self._connection is not None, "Connection is not acquired"
query_str, args, context = self._compile(query)
async with self._connection.execute(query_str, args) as cursor:
metadata = CursorResultMetaData(context, cursor.description)
async for row in cursor:
yield Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:155: AttributeError
_____________ test_column_names[select_query0-sqlite:///testsuite] _____________
database_url = 'sqlite:///testsuite'
select_query = <sqlalchemy.sql.selectable.Select object at 0x7f45175ee790>
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
> results = await database.fetch_all(query=select_query)
tests/test_databases.py:1070:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45175eefa0>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
__________ test_column_names[SELECT * FROM notes-sqlite:///testsuite] __________
database_url = 'sqlite:///testsuite', select_query = 'SELECT * FROM notes'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@pytest.mark.parametrize("select_query", [notes.select(), "SELECT * FROM notes"])
@async_adapter
async def test_column_names(database_url, select_query):
"""
Test that column names are exposed correctly through `._mapping.keys()` on each row.
"""
async with Database(database_url) as database:
async with database.transaction(force_rollback=True):
# insert values
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
# fetch results
> results = await database.fetch_all(query=select_query)
tests/test_databases.py:1070:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f4517273490>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
________________ test_result_named_access[sqlite:///testsuite] _________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_result_named_access(database_url):
async with Database(database_url) as database:
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
query = notes.select().where(notes.c.text == "example1")
> result = await database.fetch_one(query=query)
tests/test_databases.py:1150:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:152: in fetch_one
return await connection.fetch_one(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:269: in fetch_one
return await self._connection.fetch_one(built_query)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <databases.backends.sqlite.SQLiteConnection object at 0x7f451681a760>
query = <sqlalchemy.sql.selectable.Select object at 0x7f451681a8e0>
async def fetch_one(self, query: ClauseElement) -> typing.Optional[Record]:
assert self._connection is not None, "Connection is not acquired"
query_str, args, context = self._compile(query)
async with self._connection.execute(query_str, args) as cursor:
row = await cursor.fetchone()
if row is None:
return None
metadata = CursorResultMetaData(context, cursor.description)
return Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:125: AttributeError
_____________ test_mapping_property_interface[sqlite:///testsuite] _____________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
@async_adapter
async def test_mapping_property_interface(database_url):
"""
Test that all connections implement interface with `_mapping` property
"""
async with Database(database_url) as database:
query = notes.insert()
values = {"text": "example1", "completed": True}
await database.execute(query, values)
query = notes.select()
> single_result = await database.fetch_one(query=query)
tests/test_databases.py:1168:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:152: in fetch_one
return await connection.fetch_one(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:269: in fetch_one
return await self._connection.fetch_one(built_query)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <databases.backends.sqlite.SQLiteConnection object at 0x7f4516931640>
query = <sqlalchemy.sql.selectable.Select object at 0x7f45169316a0>
async def fetch_one(self, query: ClauseElement) -> typing.Optional[Record]:
assert self._connection is not None, "Connection is not acquired"
query_str, args, context = self._compile(query)
async with self._connection.execute(query_str, args) as cursor:
row = await cursor.fetchone()
if row is None:
return None
metadata = CursorResultMetaData(context, cursor.description)
return Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:125: AttributeError
____________________ test_integration[sqlite:///testsuite] _____________________
database_url = 'sqlite:///testsuite'
@pytest.mark.parametrize("database_url", DATABASE_URLS)
def test_integration(database_url):
app = get_app(database_url)
with TestClient(app) as client:
response = client.post("/notes", json={"text": "example", "completed": True})
assert response.status_code == 200
assert response.json() == {"text": "example", "completed": True}
> response = client.get("/notes")
tests/test_integration.py:95:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib/python3.8/site-packages/starlette/testclient.py:499: in get
return super().get(
/usr/lib/python3.8/site-packages/httpx/_client.py:1041: in get
return self.request(
/usr/lib/python3.8/site-packages/starlette/testclient.py:465: in request
return super().request(
/usr/lib/python3.8/site-packages/httpx/_client.py:814: in request
return self.send(request, auth=auth, follow_redirects=follow_redirects)
/usr/lib/python3.8/site-packages/httpx/_client.py:901: in send
response = self._send_handling_auth(
/usr/lib/python3.8/site-packages/httpx/_client.py:929: in _send_handling_auth
response = self._send_handling_redirects(
/usr/lib/python3.8/site-packages/httpx/_client.py:966: in _send_handling_redirects
response = self._send_single_request(request)
/usr/lib/python3.8/site-packages/httpx/_client.py:1002: in _send_single_request
response = transport.handle_request(request)
/usr/lib/python3.8/site-packages/starlette/testclient.py:342: in handle_request
raise exc
/usr/lib/python3.8/site-packages/starlette/testclient.py:339: in handle_request
portal.call(self.app, scope, receive, send)
/usr/lib/python3.8/site-packages/anyio/from_thread.py:283: in call
return cast(T_Retval, self.start_task_soon(func, *args).result())
/usr/lib64/python3.8/concurrent/futures/_base.py:444: in result
return self.__get_result()
/usr/lib64/python3.8/concurrent/futures/_base.py:389: in __get_result
raise self._exception
/usr/lib/python3.8/site-packages/anyio/from_thread.py:219: in _call_func
retval = await retval
/usr/lib/python3.8/site-packages/starlette/applications.py:122: in __call__
await self.middleware_stack(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/middleware/errors.py:184: in __call__
raise exc
/usr/lib/python3.8/site-packages/starlette/middleware/errors.py:162: in __call__
await self.app(scope, receive, _send)
/usr/lib/python3.8/site-packages/starlette/middleware/exceptions.py:79: in __call__
raise exc
/usr/lib/python3.8/site-packages/starlette/middleware/exceptions.py:68: in __call__
await self.app(scope, receive, sender)
/usr/lib/python3.8/site-packages/starlette/routing.py:718: in __call__
await route.handle(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/routing.py:276: in handle
await self.app(scope, receive, send)
/usr/lib/python3.8/site-packages/starlette/routing.py:66: in app
response = await func(request)
tests/test_integration.py:69: in list_notes
results = await database.fetch_all(query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:144: in fetch_all
return await connection.fetch_all(query, values)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/core.py:260: in fetch_all
return await self._connection.fetch_all(built_query)
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:101: in fetch_all
return [
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.0 = <list_iterator object at 0x7f45167fa490>
return [
Row(
metadata,
metadata._processors,
metadata._keymap,
> Row._default_key_style,
row,
)
for row in rows
]
E AttributeError: type object 'Row' has no attribute '_default_key_style'
../../BUILDROOT/python-databases-0.7.0-4.fc35.x86_64/usr/lib/python3.8/site-packages/databases/backends/sqlite.py:106: AttributeError
=============================== warnings summary ===============================
tests/test_integration.py::test_integration[sqlite:///testsuite]
tests/test_integration.py::test_integration[sqlite:///testsuite]
/usr/lib/python3.8/site-packages/starlette/routing.py:852: DeprecationWarning: The `on_event` decorator is deprecated, and will be removed in version 1.0.0. Refer to https://www.starlette.io/lifespan/ for recommended approach.
warnings.warn(
tests/test_integration.py::test_integration[sqlite:///testsuite]
tests/test_integration.py::test_integration[sqlite:///testsuite]
/usr/lib/python3.8/site-packages/starlette/applications.py:200: DeprecationWarning: The `route` decorator is deprecated, and will be removed in version 1.0.0. Refer to https://www.starlette.io/routing/ for the recommended approach.
warnings.warn(
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
SKIPPED [1] tests/test_databases.py:490: Test (currently) only supports asyncpg
SKIPPED [1] tests/test_databases.py:966: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:988: Test requires `pg_sleep()`
SKIPPED [1] tests/test_databases.py:1038: SQLite interface does not work with temporary tables.
SKIPPED [1] tests/test_databases.py:1089: Test is only for asyncpg
FAILED tests/test_databases.py::test_queries[sqlite:///testsuite] - Attribute...
FAILED tests/test_databases.py::test_queries_raw[sqlite:///testsuite] - Attri...
FAILED tests/test_databases.py::test_results_support_mapping_interface[sqlite:///testsuite]
FAILED tests/test_databases.py::test_results_support_column_reference[sqlite:///testsuite]
FAILED tests/test_databases.py::test_result_values_allow_duplicate_names[sqlite:///testsuite]
FAILED tests/test_databases.py::test_execute_return_val[sqlite:///testsuite]
FAILED tests/test_databases.py::test_transaction_commit[sqlite:///testsuite]
FAILED tests/test_databases.py::test_transaction_commit_low_level[sqlite:///testsuite]
FAILED tests/test_databases.py::test_transaction_decorator[sqlite:///testsuite]
FAILED tests/test_databases.py::test_datetime_field[sqlite:///testsuite] - At...
FAILED tests/test_databases.py::test_decimal_field[sqlite:///testsuite] - Att...
FAILED tests/test_databases.py::test_json_field[sqlite:///testsuite] - Attrib...
FAILED tests/test_databases.py::test_custom_field[sqlite:///testsuite] - Attr...
FAILED tests/test_databases.py::test_connections_isolation[sqlite:///testsuite]
FAILED tests/test_databases.py::test_commit_on_root_transaction[sqlite:///testsuite]
FAILED tests/test_databases.py::test_iterate_outside_transaction_with_values[sqlite:///testsuite]
FAILED tests/test_databases.py::test_column_names[select_query0-sqlite:///testsuite]
FAILED tests/test_databases.py::test_column_names[SELECT * FROM notes-sqlite:///testsuite]
FAILED tests/test_databases.py::test_result_named_access[sqlite:///testsuite]
FAILED tests/test_databases.py::test_mapping_property_interface[sqlite:///testsuite]
FAILED tests/test_integration.py::test_integration[sqlite:///testsuite] - Att...
============= 21 failed, 25 passed, 5 skipped, 4 warnings in 4.22s =============