Database integration tests

.github/workflows/test.yaml

@@ -27,6 +27,11 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
 
+      - name: Setup Test Infrastructure
+        run: |
+          docker-compose -f tests/docker-compose.yaml up -d
+        shell: bash
+
       - name: Install poetry
         run: pip install poetry
         shell: bash
@@ -38,3 +43,9 @@ jobs:
       - name: Test with pytest
         run: poetry run pytest
         shell: bash
+
+      - name: Teardown Test Infrastructure
+        if: always()
+        run: |
+          docker-compose -f tests/docker-compose.yaml down -v
+        shell: bash

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+    "python.testing.pytestArgs": [
+        "."
+    ],
+    "python.testing.unittestEnabled": false,
+    "python.testing.pytestEnabled": true
+}

README.md

@@ -9,12 +9,14 @@ pip install database-setup-tools
 ```
 
 ## Features
+
 - **Database creation on app startup**
 - Thread-safe database **session manager**
 - Opinionated towards `FastAPI` and `SQLModel` but feasible with any other framework or pure `sqlalchemy`
 - Easily use a local database in your tests
 
 ## Planned features
+
 - Database migrations with `Alembic`
 
 ## Example
@@ -67,27 +69,40 @@ if __name__ == '__main__':
 ## Example for pytest
 
 **conftest.py**
+
 ```python
 database_setup = DatabaseSetup(model_metadata=model_metadata, database_uri=DATABASE_URI)
 
+
 def pytest_sessionstart(session):
     database_setup.drop_database()
     database_setup.create_database()
 ```
 
 **test_users.py**
+
 ```python
 session_manager = SessionManager(database_uri=DATABASE_URI)
 
+
 @pytest.fixture
 def session():
-	with session_manager.get_session() as session:
-		yield session
+    with session_manager.get_session() as session:
+        yield session
+
 
 def test_create_user(session: Session):
-	user = User(name='Test User')
-	session.add(user)
-	session.commit()
-	assert session.query(User).count() == 1
-	assert session.query(User).first().name == 'Test User'
-```
+    user = User(name='Test User')
+    session.add(user)
+    session.commit()
+    assert session.query(User).count() == 1
+    assert session.query(User).first().name == 'Test User'
+```
+
+## Development
+
+### Testing
+
+1. Spin up databases for local integration tests: `docker-compose -f tests/docker-compose.yaml up -d`
+1. Create virtual environment & install dependencies: `poetry install`
+1. Run tests: `poetry run pytest`

database_setup_tools/__init__.py

@@ -1 +1,4 @@
 __version__='1.0.1'
+
+from .session_manager import SessionManager
+from .setup import DatabaseSetup

database_setup_tools/session_manager.py

@@ -1,11 +1,11 @@
 import threading
 from functools import cached_property
-from typing import Iterator, Optional
+from typing import Generator, Optional
 
-import sqlalchemy as sqla
+from sqlalchemy import create_engine
 from sqlalchemy.engine import Engine
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.orm.scoping import ScopedSession, scoped_session
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.orm.scoping import scoped_session
 
 
 class SessionManager:
@@ -47,19 +47,19 @@ def database_uri(self) -> str:
         """ Getter for the database URI """
         return self._database_uri
 
-    @cached_property
+    @property
     def engine(self) -> Engine:
         """ Getter for the engine """
         return self._engine
 
-    def get_session(self) -> Iterator[ScopedSession]:
+    def get_session(self) -> Generator[Session, None, None]:
         """ Provides a (thread safe) scoped session that is wrapped in a context manager """
         with self._Session() as session:
             yield session
 
     def _get_engine(self, **kwargs) -> Engine:
         """ Provides a database engine """
-        return sqla.create_engine(self.database_uri, **kwargs)
+        return create_engine(self.database_uri, **kwargs)
 
     @classmethod
     def _get_cached_instance(cls, args: tuple, kwargs: dict) -> Optional[object]:

database_setup_tools/setup.py

@@ -66,11 +66,14 @@ def drop_database(self) -> bool:
             return True
         return False
 
-    def create_database(self):
+    def create_database(self) -> bool:
         """ Create the database and the tables if not done yet """
-        sqlalchemy_utils.create_database(self.database_uri)
-        session_manager = SessionManager(self.database_uri)
-        self.model_metadata.create_all(session_manager.engine)
+        if not sqlalchemy_utils.database_exists(self.database_uri):
+            sqlalchemy_utils.create_database(self.database_uri)
+            session_manager = SessionManager(self.database_uri)
+            self.model_metadata.create_all(session_manager.engine)
+            return True
+        return False
 
     @classmethod
     def _get_cached_instance(cls, args: tuple, kwargs: dict) -> Optional[object]:

pyproject.toml

@@ -13,8 +13,6 @@ sqlalchemy = "^1.4.41"
 sqlalchemy-utils = "^0.38.3"
 
 [tool.poetry.dev-dependencies]
-fastapi = "0.87.0"
-uvicorn = "0.20.0"
 sqlmodel = "0.0.8"
 psycopg2-binary = "^2.9.5"
 pytest-cov = "^4.0.0"

tests/docker-compose.yaml

@@ -0,0 +1,12 @@
+version: "3.9"
+
+services:
+  postgresql-database:
+    image: postgres:14
+    container_name: "database-setup-tools-test-postgres-database"
+    ports:
+      - "5432:5432"
+    environment:
+      POSTGRES_USER: "postgres"
+      POSTGRES_PASSWORD: "postgres"
+      POSTGRES_DB: "postgres"

tests/integration/database_config.py

@@ -0,0 +1,7 @@
+SQLITE_DATABASE_URI = "sqlite:///test.db"
+POSTGRESQL_DATABASE_URI = "postgresql+psycopg2://postgres:postgres@localhost:5432/test"
+
+DATABASE_URIS = [
+    SQLITE_DATABASE_URI,
+    POSTGRESQL_DATABASE_URI,
+]

tests/integration/test_database_integration.py

@@ -1,13 +1,16 @@
+from typing import Iterator
+
 import pytest
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm.scoping import ScopedSession
 
 from database_setup_tools.session_manager import SessionManager
 from database_setup_tools.setup import DatabaseSetup
-from tests.sample_model import model_metadata, User
+from tests.integration.database_config import DATABASE_URIS
+from tests.sample_model import User, model_metadata
 
 
-@pytest.mark.parametrize('database_uri', ["sqlite:///test.db"])
+@pytest.mark.parametrize('database_uri', DATABASE_URIS)
 class TestDatabaseIntegration:
 
     @pytest.fixture
@@ -17,20 +20,21 @@ def database_setup(self, database_uri: str) -> DatabaseSetup:
         setup.drop_database()
 
     @pytest.fixture
-    def database_session(self, database_uri: str) -> ScopedSession:
+    def database_session(self, database_uri: str) -> Iterator[ScopedSession]:
         """ Get a database session """
         session_manager = SessionManager(database_uri)
         return next(session_manager.get_session())
 
     def test_create_database_and_tables(self, database_setup: DatabaseSetup, database_session: ScopedSession):
         """ Test that the tables are created correctly """
-        database_setup.create_database()
-
         # noinspection SqlInjection,SqlDialectInspection
-        test_query = database_session.execute(f'SELECT * FROM {User.__tablename__}')
+        database_session.execute(f'SELECT * FROM {User.__tablename__}')
 
-        assert test_query.cursor.description[0][0] == 'id'
-        assert test_query.cursor.description[1][0] == 'name'
+    def test_create_database_multiple_times(self, database_setup: DatabaseSetup, database_session: ScopedSession):
+        """ Test that creating the database multiple times does not cause problems """
+        database_setup.create_database()
+        # noinspection SqlInjection,SqlDialectInspection
+        database_session.execute(f'SELECT * FROM {User.__tablename__}')
 
     def test_drop_database(self, database_setup: DatabaseSetup, database_session: ScopedSession):
         """ Test that the database is dropped correctly """