Merge pull request #10 from getmarkus/cm-branch-9

refactor(tests): enhance test structure with fixtures for DB sessions

Author: getmarkus

main.py

@@ -21,27 +21,28 @@
 # https://betterstack.com/community/guides/logging/loguru/
 
 env = os.getenv("PYTHON_TEMPLATE_ENV", ".env")
-running = False
-
 
 def isRunning() -> bool:
-    logger.info(f"running: {running}")
+    #logger.info(f"App state: {dict(app.state.__dict__)}")
+    running = getattr(app.state, "running", False)
+    logger.info(f"Running state: {running}")
     return running
 
 
 # https://fastapi.tiangolo.com/advanced/events/
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    global running
-    running = True
+    app.state.running = True
     logger.info("Lifespan started")
+    logger.info(f"App state: {dict(app.state.__dict__)}")
+    logger.info(app.state.running)
 
     # Initialize database if using SQLModel
     if Settings.get_settings().execution_mode == "sqlmodel" and not Settings.get_settings().migrate_database:
         init_db()
 
     yield
-    running = False
+    app.state.running = False
     logger.info("Lifespan stopped")
 
 
@@ -164,7 +165,6 @@ async def add_process_time_header(request, call_next):
 async def root(
     settings: Annotated[Settings, Depends(Settings.get_settings)],
 ) -> Dict[str, Any]:
-    logger.info(f"running: {running}")
     return {
         "app_name": settings.project_name,
         "system_time": datetime.datetime.now(),

src/app/usecases/analyze_issue.py

@@ -23,6 +23,7 @@ def __init__(
     def analyze(self) -> Issue:
         logger.info(f"analyzing issue: {self.issue_number}")
         issue = self.repo.get_by_id(self.issue_number)
+        logger.info(f"issue: {issue}")
         if issue.issue_number == 0:
             raise NotFoundException(
                 message="Issue not found",

src/interface_adapters/endpoints/issues.py

@@ -19,7 +19,6 @@
 # https://fastapi.tiangolo.com/tutorial/dependencies/
 def configure_repository(
     session: Annotated[Session, Depends(get_db)]
-    #session: SessionDep,
 ) -> IssueRepository:
     execution_mode = Settings.get_settings().execution_mode
     if execution_mode == "in-memory":

tests/test_issues.py

@@ -1,16 +1,51 @@
+import pytest
 from fastapi.testclient import TestClient
+from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel.pool import StaticPool
 
-from src.interface_adapters.endpoints import issues
+from main import app
+from src.domain.issue import Issue, IssueState
+from src.resource_adapters.persistence.sqlmodel.database import get_db
+from src.resource_adapters.persistence.sqlmodel.issues import SQLModelIssueRepository
 
-client = TestClient(issues.router)
+
+# https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/
+@pytest.fixture(name="session")
+def session_fixture():
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
+    )
+    SQLModel.metadata.create_all(engine)
+    with Session(engine) as session:
+        yield session
+
+
+@pytest.fixture(name="client")
+def client_fixture(session: Session):
+    def get_session_override():
+        return session
+
+    app.dependency_overrides[get_db] = get_session_override
+    app.state.running = True
+    client = TestClient(app)
+    yield client
+    app.dependency_overrides.clear()
+    app.state.running = False
 
 
 class TestAnalyzeIssue:
-    def test_analyze_issue(self):
+    def test_analyze_issue(self, client: TestClient, session: Session):
         # Test case 1: Successful analysis
-        response = client.post("/issues/123/analyze")
-        assert response.status_code == 200
-        assert response.json() == {"version": 1, "issue_number": 123}
+        issue_number = 1
+        test_issue = Issue(issue_number=issue_number, issue_state=IssueState.OPEN)
+
+        repository = SQLModelIssueRepository(session)
+
+        repository.add(test_issue)
+
+        response = client.post("/issues/1/analyze")
+        #assert response.status_code == 200
+        assert response.json() == {"version": 1, "issue_number": 1}
 
         # Test case 2: Invalid issue number
         try:

tests/test_main.py

@@ -1,36 +1,59 @@
+import pytest
 from fastapi.testclient import TestClient
+from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel.pool import StaticPool
 
 from main import app
+from src.resource_adapters.persistence.sqlmodel.database import get_db
+
+
+# https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/
+@pytest.fixture(name="session")
+def session_fixture():
+    engine = create_engine(
+        "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
+    )
+    SQLModel.metadata.create_all(engine)
+    with Session(engine) as session:
+        yield session
+
+
+@pytest.fixture(name="client")
+def client_fixture(session: Session):
+    def get_session_override():
+        return session
+
+    app.dependency_overrides[get_db] = get_session_override
+    app.state.running = True
+    client = TestClient(app)
+    yield client
+    app.dependency_overrides.clear()
+    app.state.running = False
 
 
 class TestHealth:
-    def test_root(self):
-        with TestClient(app) as client:
-            response = client.get("/")
-            assert response.status_code == 200
-            assert response.json()["app_name"] == "python-template"
-
-    def test_health(self):
-        with TestClient(app) as client:
-            response = client.get("/health")
-            assert response.status_code == 200
-
-    def test_startup(self):
-        with TestClient(app) as client:
-            response = client.get("/startup")
-            assert response.status_code == 200
-
-    def test_readiness(self):
-        with TestClient(app) as client:
-            response = client.get("/readiness")
-            assert response.status_code == 200
-
-    def test_liveness(self):
-        with TestClient(app) as client:
-            response = client.get("/liveness")
-            assert response.status_code == 200
-
-    def test_smoke(self):
-        with TestClient(app) as client:
-            response = client.get("/smoke")
-            assert response.status_code == 200
+
+    def test_root(self, client: TestClient):
+        response = client.get("/")
+        assert response.status_code == 200
+        assert response.json()["app_name"] == "python-template"
+
+    def test_health(self, client: TestClient):
+        response = client.get("/health")
+        assert response.status_code == 200
+
+    def test_startup(self, client: TestClient):
+        response = client.get("/startup")
+        assert response.status_code == 200
+
+    def test_readiness(self, client: TestClient):
+        response = client.get("/readiness")
+        assert response.status_code == 200
+
+    def test_liveness(self, client: TestClient):
+        response = client.get("/liveness")
+        assert response.status_code == 200
+
+    def test_smoke(self, client: TestClient):
+        response = client.get("/smoke")
+        assert response.status_code == 200

tests/test_sqlmodel_repository.py

@@ -1,111 +1,107 @@
 import pytest
-from sqlmodel import SQLModel
+from sqlmodel import Session, SQLModel, create_engine
+from sqlmodel.pool import StaticPool
 
 from src.domain.issue import Issue, IssueState
-from src.resource_adapters.persistence.sqlmodel.database import get_engine
-from src.resource_adapters.persistence.sqlmodel.unit_of_work import SQLModelUnitOfWork
+from src.resource_adapters.persistence.sqlmodel.issues import SQLModelIssueRepository
 
 
-@pytest.fixture
-def uow():
-    """Create a SQLModelUnitOfWork with in-memory database."""
-    # Reset the global engine for each test
-    import src.resource_adapters.persistence.sqlmodel.database as db
-
-    db._engine = None
-
-    # Create a fresh database
-    database_url = "sqlite://"
-    uow = SQLModelUnitOfWork(database_url=database_url)
-    engine = get_engine(database_url)
-    SQLModel.metadata.create_all(engine)
-    return uow
-
-
-class TestSQLModelRepository:
+# https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/
+class TestSQLModelIssueRepository:
     """Test suite for SQLModel-based issue repository."""
 
-    @pytest.mark.anyio
-    async def test_add_and_get_issue(self, uow):
+    @pytest.fixture(name="session")
+    def session_fixture(self):
+        engine = create_engine(
+            "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
+        )
+        SQLModel.metadata.create_all(engine)
+        with Session(engine) as session:
+            yield session
+
+    def test_add_and_get_issue(self, session: Session):
         """Test adding and retrieving an issue."""
         issue_number = 1
         test_issue = Issue(issue_number=issue_number, issue_state=IssueState.OPEN)
 
+        uow = SQLModelIssueRepository(session)
+
         with uow:
-            await uow.issues.add(test_issue)
+            uow.add(test_issue)
             uow.commit()
 
         with uow:
-            retrieved_issue = await uow.issues.get_by_id(issue_number)
+            retrieved_issue = uow.get_by_id(issue_number)
             assert retrieved_issue.issue_number == issue_number
             assert retrieved_issue.issue_state == IssueState.OPEN
 
-    @pytest.mark.anyio
-    async def test_list_issues(self, uow):
+    def test_list_issues(self, session: Session):
         """Test listing all issues."""
         issue_number = 1
         test_issue = Issue(issue_number=issue_number, issue_state=IssueState.OPEN)
 
+        uow = SQLModelIssueRepository(session)
+
         with uow:
-            await uow.issues.add(test_issue)
+            uow.add(test_issue)
             uow.commit()
 
         with uow:
-            issues = await uow.issues.list()
+            issues = uow.list()
             assert len(issues) == 1
             assert issues[0].issue_number == issue_number
 
-    @pytest.mark.anyio
-    async def test_update_issue(self, uow):
+    def test_update_issue(self, session: Session):
         """Test updating an issue's state."""
         issue_number = 1
         test_issue = Issue(issue_number=issue_number, issue_state=IssueState.OPEN)
 
+        uow = SQLModelIssueRepository(session)
         with uow:
-            await uow.issues.add(test_issue)
+            uow.add(test_issue)
             uow.commit()
 
         with uow:
-            issue_to_update = await uow.issues.get_by_id(issue_number)
+            issue_to_update = uow.get_by_id(issue_number)
             issue_to_update.issue_state = IssueState.CLOSED
-            await uow.issues.update(issue_to_update)
+            uow.update(issue_to_update)
             uow.commit()
 
         with uow:
-            updated_issue = await uow.issues.get_by_id(issue_number)
+            updated_issue = uow.get_by_id(issue_number)
             assert updated_issue.issue_state == IssueState.CLOSED
 
-    @pytest.mark.anyio
-    async def test_filter_issues(self, uow):
+    def test_filter_issues(self, session: Session):
         """Test filtering issues with a predicate."""
         issue_number = 1
         test_issue = Issue(issue_number=issue_number, issue_state=IssueState.CLOSED)
 
+        uow = SQLModelIssueRepository(session)
         with uow:
-            await uow.issues.add(test_issue)
+            uow.add(test_issue)
             uow.commit()
 
         with uow:
-            closed_issues = await uow.issues.list_with_predicate(
+            closed_issues = uow.list_with_predicate(
                 lambda i: i.issue_state == IssueState.CLOSED
             )
             assert len(closed_issues) == 1
 
-    @pytest.mark.anyio
-    async def test_remove_issue(self, uow):
+    def test_remove_issue(self, session: Session):
         """Test removing an issue."""
         issue_number = 1
         test_issue = Issue(issue_number=issue_number, issue_state=IssueState.OPEN)
 
+        uow = SQLModelIssueRepository(session)
         with uow:
-            await uow.issues.add(test_issue)
+            uow.add(test_issue)
             uow.commit()
 
         with uow:
-            issue_to_remove = await uow.issues.get_by_id(issue_number)
-            await uow.issues.remove(issue_to_remove)
+            issue_to_remove = uow.get_by_id(issue_number)
+            uow.remove(issue_to_remove)
             uow.commit()
 
         with uow:
-            remaining_issues = await uow.issues.list()
+            remaining_issues = uow.list()
             assert len(remaining_issues) == 0