feat: pytest plugin (#11)

Author: hadrien

README.md

@@ -45,3 +45,77 @@ router = APIRouter()
 def example(session: Session = Depends(with_session)):
     return session.execute("SELECT now()").scalar()
 ```
+
+## Pytest fixtures
+This library provides a set of utility fixtures, through its PyTest plugin, which is
+automatically installed with the library.
+
+By default, no records are actually written to the database when running tests.
+There currently is no way to change this behaviour.
+
+### `sqla_modules`
+
+You must define this fixture, in order for the plugin to reflect table metadata in your
+SQLAlchemy entities. It should just import all of the application's modules which contain
+SQLAlchemy models.
+
+Example:
+
+```python
+# tests/conftest.py
+from pytest import fixture
+
+
+@fixture
+def sqla_modules():
+    from er import sqla  # noqa
+```
+
+### `db_url`
+
+The DB url to use.
+
+When `CI` key is set in environment variables, it defaults to using `postgres` as the host name:
+
+```
+postgresql://postgres@posgres/postgres
+```
+
+In other cases, the host is set to `localhost`:
+
+```
+postgresql://postgres@localhost/postgres
+```
+
+Of course, you can override it by overloading the fixture:
+
+```python
+from pytest import fixture
+
+
+@fixture(scope="session")
+def db_url():
+    return "postgresql://postgres@localhost/test_database"
+```
+
+
+### `session`
+
+Sqla session to create db fixture:
+* All changes done at test setiup or during the test are rollbacked at test tear down;
+* No record will actually be written in the database;
+* Changes in one session need to be committed to be available from other sessions;
+
+Example:
+```python
+from pytest import fixture
+
+
+@fixture
+def patient(session):
+    from er.sqla import Patient
+    patient = Patient(first_name="Bob", last_name="David")
+    session.add(patient)
+    session.commit()
+    return patient
+```

fastapi_sqla.py fastapi_sqla/__init__.pyfastapi_sqla.py renamed to fastapi_sqla/__init__.py

@@ -0,0 +1,71 @@
+import os
+from unittest.mock import patch
+
+from pytest import fixture
+from sqlalchemy import create_engine
+
+
+@fixture(scope="session")
+def db_url():
+    """Default db url used by depending fixtures.
+
+    When CI key is set in environment variables, it uses `postgres` as host name:
+    postgresql://postgres@posgres/postgres
+
+    Else, host used is `localhost`: postgresql://postgres@localhost/postgres
+    """
+    host = "postgres" if "CI" in os.environ else "localhost"
+    return f"postgresql://postgres@{host}/postgres"
+
+
+@fixture(scope="session")
+def sqla_connection(db_url):
+    engine = create_engine(db_url)
+    connection = engine.connect()
+    yield connection
+    connection.close()
+
+
+@fixture
+def sqla_modules():
+    raise Exception(
+        "sqla_modules fixture is not defined. Define a sqla_modules fixture which "
+        "imports all modules with sqla entities deriving from fastapi_sqla.Base ."
+    )
+
+
+@fixture(autouse=True)
+def sqla_reflection(sqla_modules, sqla_connection, db_url):
+    import fastapi_sqla
+
+    fastapi_sqla.Base.metadata.bind = sqla_connection
+    fastapi_sqla.Base.prepare(sqla_connection)
+
+
+@fixture(autouse=True)
+def patch_sessionmaker(db_url, sqla_connection, sqla_transaction):
+    """So that all DB operations are never written to db for real."""
+    with patch("fastapi_sqla.engine_from_config") as engine_from_config:
+        engine_from_config.return_value = sqla_connection
+        yield engine_from_config
+
+
+@fixture
+def sqla_transaction(sqla_connection):
+    transaction = sqla_connection.begin()
+    yield transaction
+    transaction.rollback()
+
+
+@fixture
+def session(sqla_transaction, sqla_connection):
+    """Sqla session to use when creating db fixtures.
+
+    While it does not write any record in DB, the application will still be able to access any record
+    committed with that session.
+    """
+    import fastapi_sqla
+
+    session = fastapi_sqla._Session(bind=sqla_connection)
+    yield session
+    session.close()

fastapi_sqla/_pytest_plugin.py

@@ -31,3 +31,6 @@ build-backend = "poetry.masonry.api"
 version_variable = "pyproject.toml:version"
 upload_to_pypi = false
 commit_message = "Version generated by python-semantic-release [ci skip]"
+
+[tool.poetry.plugins."pytest11"]
+fastapi_sqla = "fastapi_sqla._pytest_plugin"

pyproject.toml

@@ -14,7 +14,7 @@ ext = .py,.yaml,.cfg,.yml
 
 [coverage:run]
 branch = True
-omit = tests/*,.venv/*
+omit = tests/*,.venv/*,fastapi_sqla/_pytest_plugin.py
 
 [coverage:report]
 skip_covered = true

setup.cfg

@@ -1,24 +1,24 @@
 import importlib
-import os
 from unittest.mock import patch
 
 from pytest import fixture
+from sqlalchemy import engine_from_config
 from sqlalchemy.orm.session import close_all_sessions
 
 
-@fixture(scope="session")
-def db_uri():
-    host = "postgres" if "CIRCLECI" in os.environ else "localhost"
-    return f"postgresql://postgres@{host}/postgres"
-
-
-@fixture(autouse=True)
-def environ(db_uri):
-    values = {"sqlalchemy_url": db_uri}
+@fixture(scope="session", autouse=True)
+def environ(db_url):
+    values = {"sqlalchemy_url": db_url}
     with patch.dict("os.environ", values=values, clear=True):
         yield values
 
 
+@fixture(scope="session")
+def engine(environ):
+    engine = engine_from_config(environ, prefix="sqlalchemy_")
+    return engine
+
+
 @fixture(autouse=True)
 def tear_down():
     import fastapi_sqla
@@ -28,3 +28,8 @@ def tear_down():
     close_all_sessions()
     # reload fastapi_sqla to clear sqla deferred reflection mapping stored in Base
     importlib.reload(fastapi_sqla)
+
+
+@fixture
+def sqla_modules():
+    pass

tests/conftest.py

@@ -1,16 +1,14 @@
 from pytest import fixture, raises
 from sqlalchemy import engine_from_config
 from sqlalchemy.exc import NoSuchTableError
-from sqlalchemy.orm.session import close_all_sessions
 
 
-@fixture(autouse=True)
+@fixture(autouse=True, scope="module")
 def setup_tear_down(environ):
     engine = engine_from_config(environ, prefix="sqlalchemy_")
 
     engine.execute("CREATE TABLE IF NOT EXISTS test_table (id integer primary key)")
     yield
-    close_all_sessions()
     engine.execute("DROP TABLE test_table")
 
 

tests/test_base.py

@@ -5,20 +5,12 @@
 from fastapi import Depends, FastAPI
 from pydantic import BaseModel
 from pytest import fixture, mark
-from sqlalchemy import engine_from_config
-from sqlalchemy.orm.session import close_all_sessions
 from structlog.testing import capture_logs
 
 pytestmark = mark.asyncio
 
 
-@fixture
-def engine(environ):
-    engine = engine_from_config(environ, prefix="sqlalchemy_")
-    return engine
-
-
-@fixture(autouse=True)
+@fixture(scope="module", autouse=True)
 def setup_tear_down(engine):
     engine.execute(
         """
@@ -30,7 +22,6 @@ def setup_tear_down(engine):
     """
     )
     yield
-    close_all_sessions()
     engine.execute("DROP TABLE public.user")
 
 

tests/test_middleware.py

@@ -0,0 +1,47 @@
+from pytest import fixture
+
+
+@fixture(scope="module", autouse=True)
+def setup_tear_down(engine):
+    engine.execute(
+        """
+        CREATE TABLE IF NOT EXISTS singer (
+           id integer primary key,
+           name varchar,
+           country varchar
+        )
+    """
+    )
+    yield
+    engine.execute("DROP TABLE singer")
+
+
+@fixture
+def singer_cls():
+    from fastapi_sqla import Base
+
+    class Singer(Base):
+        __tablename__ = "singer"
+
+    return Singer
+
+
+@fixture
+def sqla_modules(singer_cls):
+    pass
+
+
+def test_session_fixture_does_not_write_in_db(session, singer_cls, engine):
+    session.add(singer_cls(id=1, name="Bob Marley", country="Jamaica"))
+    session.commit()
+    assert engine.execute("select count(*) from singer").scalar() == 0
+
+
+def test_all_opened_sessions_are_within_the_same_transaction(session, singer_cls):
+    from fastapi_sqla import _Session
+
+    session.add(singer_cls(id=1, name="Bob Marley", country="Jamaica"))
+    session.commit()
+
+    other_session = _Session()
+    assert other_session.query(singer_cls).get(1)