Add basic unit tests using py.test.

Signed-off-by: Laura <[email protected]>

.coveragerc

@@ -0,0 +1,29 @@
+# .coveragerc to control coverage.py
+[run]
+branch = True
+
+[report]
+# Regexes for lines to exclude from consideration
+exclude_lines =
+    # Have to re-enable the standard pragma
+    pragma: no cover
+
+    # Don't complain about missing debug-only code:
+    def __repr__
+    if self\.debug
+
+    # Don't complain if tests don't hit defensive assertion code:
+    raise AssertionError
+    raise NotImplementedError
+
+    # Don't complain if non-runnable code isn't run:
+    if 0:
+    if __name__ == .__main__.:
+
+ignore_errors = True
+omit =
+    # migration tool is uncovered
+    asyncqlio/orm/ddl/migration_tool.py
+
+[html]
+directory = coverage_html_report

setup.cfg

@@ -0,0 +1,6 @@
+[aliases]
+test=pytest
+
+[tool:pytest]
+addopts = --verbose --cov=asyncqlio
+python_files = tests/*.py

setup.py

@@ -33,7 +33,8 @@
     description="An asyncio ORM for Python 3.5+",
     long_description=Path(__file__).with_name("README.rst").read_text(encoding="utf-8"),
     setup_requires=[
-        "setuptools_scm"
+        "setuptools_scm",
+        "pytest-runner"
     ],
     install_requires=[
         "cached_property==1.3.0",
@@ -52,6 +53,11 @@
             "aiomysql>=0.0.9",
         ]
     },
+    test_requires=[
+        "pytest",
+        "pytest-asyncio",
+        "pytest-cov"
+    ],
     python_requires=">=3.5.2",
     entry_points={
         "console_scripts": ["asql-migrate=asyncqlio.orm.ddl.migration_tool:cli"]

tests/conftest.py

@@ -0,0 +1,25 @@
+"""
+py.test configuration
+"""
+import asyncio
+import os
+
+import pytest
+
+from asyncqlio import DatabaseInterface
+
+# set the iface as a global so it can be closed later
+iface = DatabaseInterface()
+
+
+@pytest.fixture(scope='module')
+async def db() -> DatabaseInterface:
+    await iface.connect(dsn=os.environ["ASQL_DSN"])
+    return iface
+
+
+# override for a module scope
+@pytest.fixture(scope="module")
+def event_loop():
+    return asyncio.get_event_loop()
+

tests/test_0lowlevel.py

@@ -0,0 +1,73 @@
+"""
+Tests the low-level API.
+"""
+
+import pytest
+
+from asyncqlio import BaseTransaction, DatabaseInterface
+
+# mark all test_ functions as coroutines
+pytestmark = pytest.mark.asyncio
+
+
+async def test_db_connected(db: DatabaseInterface):
+    assert db.connected
+    assert db.connector is not None
+
+
+async def test_acquire_transaction(db: DatabaseInterface):
+    tr = db.get_transaction()
+
+    assert isinstance(tr, BaseTransaction)
+
+
+async def test_transaction_use(db: DatabaseInterface):
+    tr = db.get_transaction()
+    await tr.begin()
+
+    # this just ensures the connection doesn't error
+    await tr.execute("SELECT 1 + 1;")
+    await tr.rollback()
+    await tr.close()
+
+async def test_transaction_fetch_one(db: DatabaseInterface):
+    tr = db.get_transaction()
+    await tr.begin()
+
+    cursor = await tr.cursor("SELECT 1 + 1;")
+    async with cursor:
+        row = await cursor.fetch_row()
+    # rowdict
+    assert row[0] == 2
+    await tr.rollback()
+    await tr.close()
+
+
+async def test_transaction_fetch_multiple(db: DatabaseInterface):
+    tr = db.get_transaction()
+    await tr.begin()
+
+    cursor = await tr.cursor('SELECT 1 AS result UNION ALL SELECT 2;')
+    previous = 0
+    async with cursor:
+        async for row in cursor:
+            assert row["result"] > previous
+            previous = row["result"]
+
+    await tr.rollback()
+    await tr.close()
+
+
+async def test_transaction_fetch_many(db: DatabaseInterface):
+    tr = db.get_transaction()
+    await tr.begin()
+
+    cursor = await tr.cursor('SELECT 1 AS result UNION ALL SELECT 2;')
+    async with cursor:
+        rows = await cursor.fetch_many(n=2)
+
+    assert rows[0]["result"] == 1
+    assert rows[1]["result"] == 2
+
+    await tr.rollback()
+    await tr.close()