Convert test_theme as new style test, cleanup

geoportal/tests/functional/__init__.py

@@ -32,22 +32,25 @@
 
 import logging
 from configparser import ConfigParser
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import pyramid.registry
 import pyramid.request
-import sqlalchemy.exc
 import tests
 import transaction
 import webob.acceptparse
 from c2c.template.config import config as configuration
 from pyramid import testing
+from sqlalchemy.orm.session import Session
 
 import c2cgeoportal_geoportal
 import c2cgeoportal_geoportal.lib
 from c2cgeoportal_commons import models
 from c2cgeoportal_geoportal.lib import caching
 
+if TYPE_CHECKING:
+    from c2cgeoportal_commons.models import main, static
+
 _LOG = logging.getLogger(__name__)
 mapserv_url = "http://mapserver:8080/"
 config = None
@@ -76,27 +79,27 @@ def cleanup_db() -> None:
     )
     from c2cgeoportal_commons.models.static import OAuth2Client, Shorturl, User
 
-    try:
-        transaction.commit()
-    except sqlalchemy.exc.InvalidRequestError:
-        models.DBSession.rollback()
-
-    for ra in models.DBSession.query(RestrictionArea).all():
-        ra.roles = []
-        models.DBSession.delete(ra)
-    for ti in models.DBSession.query(TreeItem).all():
-        models.DBSession.delete(ti)
-    models.DBSession.query(OGCServer).delete()
-    models.DBSession.query(Interface).delete()
-    for r in models.DBSession.query(Role).all():
-        r.functionalities = []
-        models.DBSession.delete(r)
-    models.DBSession.query(User).delete()
-    models.DBSession.query(Functionality).delete()
-    models.DBSession.query(FullTextSearch).delete()
-    models.DBSession.query(Shorturl).delete()
-    models.DBSession.query(OAuth2Client).delete()
-    transaction.commit()
+    with models.DBSession() as session:
+        for ra in session.query(RestrictionArea).all():
+            ra.roles = []
+            session.delete(ra)
+        for ti in session.query(TreeItem).all():
+            session.delete(ti)
+        session.query(OGCServer).delete()
+        session.query(Interface).delete()
+        for r in session.query(Role).all():
+            r.functionalities = []
+            session.delete(r)
+        session.query(User).delete()
+        session.query(Functionality).delete()
+        session.query(FullTextSearch).delete()
+        session.query(Shorturl).delete()
+        session.query(OAuth2Client).delete()
+        try:
+            transaction.commit()
+        except Exception as e:
+            _LOG.error(e)
+            transaction.abort()
 
 
 def setup_db() -> None:
@@ -108,9 +111,13 @@ def setup_db() -> None:
     from c2cgeoportal_commons.models import DBSession
     from c2cgeoportal_commons.models.main import Role
 
-    DBSession.add_all([Role(name) for name in ("anonymous", "registered", "intranet")])
+    with DBSession() as session:
+        for role_name in ("anonymous", "registered", "intranet"):
+            role = session.query(Role).filter(Role.name == role_name).one_or_none()
+            if role is None:
+                session.add(Role(name=role_name))
 
-    transaction.commit()
+        transaction.commit()
 
     c2cgeoportal_geoportal.lib.ogc_server_wms_url_ids = None
     c2cgeoportal_geoportal.lib.ogc_server_wfs_url_ids = None
@@ -170,21 +177,21 @@ def teardown_common() -> None:
     models.DBSessions = {}
 
 
-def create_default_ogcserver():
-    from c2cgeoportal_commons.models import DBSession
+def create_default_ogcserver(session: Session) -> "main.OGCServer":
     from c2cgeoportal_commons.models.main import OGCServer
 
-    transaction.commit()
-    ogcserver = OGCServer(name="__test_ogc_server")
+    ogcserver = session.query(OGCServer).filter(OGCServer.name == "__test_ogc_server").one_or_none()
+    if ogcserver is None:
+        ogcserver = OGCServer(name="__test_ogc_server")
     ogcserver.url = mapserv_url
-    DBSession.add(ogcserver)
-    transaction.commit()
+    session.add(ogcserver)
     caching.invalidate_region()
+    session.flush()
 
     return ogcserver
 
 
-def _get_user(username: str):
+def _get_user(username: str) -> "static.User":
     from c2cgeoportal_commons.models import DBSession
     from c2cgeoportal_commons.models.static import User
 
@@ -285,12 +292,10 @@ def create_dummy_request(
     return request
 
 
-def fill_tech_user_functionality(name, functionalities):
-    from c2cgeoportal_commons.models import DBSession
+def fill_tech_user_functionality(name, functionalities, session: Session) -> None:
     from c2cgeoportal_commons.models.main import Functionality, Role
 
-    role = DBSession.query(Role).filter_by(name=name).one()
+    role = session.query(Role).filter_by(name=name).one()
     role.functionalities = [Functionality(name, value) for name, value in functionalities]
-    DBSession.add(role)
-    transaction.commit()
+    session.add(role)
     caching.invalidate_region()

geoportal/tests/functional/conftest.py

@@ -25,19 +25,29 @@
 # of the authors and should not be interpreted as representing official policies,
 # either expressed or implied, of the FreeBSD Project.
 
+import logging
+from typing import TYPE_CHECKING, Any
 from unittest.mock import patch
 
+import pyramid.request
 import pytest
 import sqlalchemy.orm
 import transaction
 from c2c.template.config import config as configuration
 from pyramid.testing import DummyRequest
+from sqlalchemy.orm.session import Session, SessionTransaction
 from tests.functional import setup_common as setup_module
 
+from c2cgeoportal_commons.testing import generate_mappers, get_engine, get_session_factory, get_tm_session
+from c2cgeoportal_commons.testing.initializedb import truncate_tables
 from c2cgeoportal_geoportal.lib import caching
 
+_LOG = logging.getLogger(__name__)
 mapserv_url = "http://mapserver:8080/"
 
+if TYPE_CHECKING:
+    from c2cgeoportal_commons.models import main, static
+
 
 @pytest.fixture
 def settings():
@@ -46,23 +56,51 @@ def settings():
 
 
 @pytest.fixture
-def dbsession(settings):
+def dbsession_old(settings: dict[str, Any]) -> Session:
     from c2cgeoportal_commons.models import DBSession
 
+    truncate_tables(DBSession)
     with patch("c2cgeoportal_geoportal.views.vector_tiles.DBSession", new=DBSession):
         yield DBSession
+    truncate_tables(DBSession)
+
+
+@pytest.fixture
+def dbsession(settings: dict[str, Any]) -> Session:
+    generate_mappers()
+    engine = get_engine(settings)
+    session_factory = get_session_factory(engine)
+    session = get_tm_session(session_factory, transaction.manager)
+    truncate_tables(session)
+    with patch("c2cgeoportal_commons.models.DBSession", new=session):
+        yield session
+    truncate_tables(session)
+
+
+@pytest.fixture
+def transact_old(dbsession_old) -> SessionTransaction:
+    t = dbsession_old.begin_nested()
+    yield t
+    try:
+        t.rollback()
+    except sqlalchemy.exc.ResourceClosedError:  # pragma: no cover
+        _LOG.warning("Transaction already closed")
+    dbsession_old.expire_all()
 
 
 @pytest.fixture
-def transact(dbsession):
+def transact(dbsession) -> SessionTransaction:
     t = dbsession.begin_nested()
     yield t
-    t.rollback()
+    try:
+        t.rollback()
+    except sqlalchemy.exc.ResourceClosedError:  # pragma: no cover
+        _LOG.warning("Transaction already closed")
     dbsession.expire_all()
 
 
 @pytest.fixture
-def dummy_request(dbsession):
+def dummy_request(dbsession_old) -> pyramid.request.Request:
     """
     A lightweight dummy request.
 
@@ -76,48 +114,67 @@ def dummy_request(dbsession):
     """
     request = DummyRequest()
     request.host = "example.com"
-    request.dbsession = dbsession
+    request.dbsession = dbsession_old
 
     return request
 
 
 @pytest.fixture
-def default_ogcserver(dbsession, transact):
+def default_ogcserver(dbsession_old, transact_old) -> "main.OGCServer":
     from c2cgeoportal_commons.models.main import OGCServer
 
-    del transact
+    del transact_old
 
-    ogcserver = OGCServer(name="__test_ogc_server")
+    dbsession_old.flush()
+    ogcserver = dbsession_old.query(OGCServer).filter(OGCServer.name == "__test_ogc_server").one_or_none()
+    if ogcserver is None:
+        ogcserver = OGCServer(name="__test_ogc_server")
     ogcserver.url = mapserv_url
-    dbsession.add(ogcserver)
-    dbsession.flush()
+    dbsession_old.add(ogcserver)
+    dbsession_old.flush()
 
     yield ogcserver
 
 
 @pytest.fixture
-def some_user(dbsession, transact):
+def some_user(dbsession_old, transact_old) -> "static.User":
     from c2cgeoportal_commons.models.static import User
 
-    del transact
+    del transact_old
 
     user = User(username="__test_user", password="__test_user")
-    dbsession.add(user)
-    dbsession.flush()
+    dbsession_old.add(user)
+    dbsession_old.flush()
 
     yield user
 
 
 @pytest.fixture
-def admin_user(dbsession, transact):
+def admin_user(dbsession_old, transact_old) -> "static.User":
     from c2cgeoportal_commons.models.main import Role
     from c2cgeoportal_commons.models.static import User
 
-    del transact
+    del transact_old
 
     role = Role(name="role_admin")
     user = User(username="__test_user", password="__test_user", settings_role=role, roles=[role])
-    dbsession.add_all([role, user])
-    dbsession.flush()
+    dbsession_old.add_all([role, user])
+    dbsession_old.flush()
 
     yield user
+
+
+@pytest.fixture()
+@pytest.mark.usefixtures("dbsession")
+def default_ogcserver(dbsession: Session) -> "main.OGCServer":
+    from c2cgeoportal_commons.models.main import OGCServer
+
+    dbsession.flush()
+    ogcserver = dbsession.query(OGCServer).filter(OGCServer.name == "__test_ogc_server").one_or_none()
+    if ogcserver is None:
+        ogcserver = OGCServer(name="__test_ogc_server")
+    ogcserver.url = mapserv_url
+    dbsession.add(ogcserver)
+    caching.invalidate_region()
+
+    return ogcserver

geoportal/tests/functional/test_entry.py

@@ -96,7 +96,7 @@ def setup_method(self, _):
         a_geo_table.drop(checkfirst=True, bind=engine)
         a_geo_table.create(bind=engine)
 
-        ogcserver = create_default_ogcserver()
+        ogcserver = create_default_ogcserver(DBSession)
 
         private_layer_edit = LayerWMS(name="__test_private_layer_edit", public=False)
         private_layer_edit.layer = "__test_private_layer"

geoportal/tests/functional/test_functionalities.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2013-2023, Camptocamp SA
+# Copyright (c) 2013-2024, Camptocamp SA
 # All rights reserved.
 
 # Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@ def setup_method(self, _):
         from c2cgeoportal_commons.models.main import Functionality, Role
         from c2cgeoportal_commons.models.static import User
 
-        create_default_ogcserver()
+        create_default_ogcserver(DBSession)
         role1 = Role(name="__test_role1")
         user1 = User(username="__test_user1", password="__test_user1", settings_role=role1, roles=[role1])
         role2 = Role(name="__test_role2")
@@ -128,7 +128,7 @@ def test_functionalities(self):
         self.assertEqual(get_functionality("__test_b", request3, False), [])
 
         fill_tech_user_functionality(
-            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2"))
+            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2")), DBSession
         )
         settings = {
             "admin_interface": {"available_functionalities": [{"name": "__test_a"}, {"name": "__test_s"}]}
@@ -146,9 +146,9 @@ def test_functionalities(self):
         self.assertEqual(get_functionality("__test_s", request2, False), ["db"])
         self.assertEqual(set(get_functionality("__test_a", request2, False)), {"db1", "db2"})
 
-        fill_tech_user_functionality("registered", [])
+        fill_tech_user_functionality("registered", [], DBSession)
         fill_tech_user_functionality(
-            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2"))
+            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2")), DBSession
         )
         settings = {
             "admin_interface": {"available_functionalities": [{"name": "__test_a"}, {"name": "__test_s"}]}
@@ -167,10 +167,10 @@ def test_functionalities(self):
         self.assertEqual(set(get_functionality("__test_a", request2, False)), {"db1", "db2"})
 
         fill_tech_user_functionality(
-            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2"))
+            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2")), DBSession
         )
         fill_tech_user_functionality(
-            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2"))
+            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2")), DBSession
         )
         settings = {
             "admin_interface": {"available_functionalities": [{"name": "__test_a"}, {"name": "__test_s"}]}
@@ -204,10 +204,10 @@ def test_web_client_functionalities(self):
         request2.user = DBSession.query(User).filter(User.username == "__test_user2").one()
 
         fill_tech_user_functionality(
-            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2"))
+            "registered", (("__test_s", "registered"), ("__test_a", "r1"), ("__test_a", "r2")), DBSession
         )
         fill_tech_user_functionality(
-            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2"))
+            "anonymous", (("__test_s", "anonymous"), ("__test_a", "a1"), ("__test_a", "a2")), DBSession
         )
         settings = {
             "functionalities": {"available_in_templates": ["__test_s", "__test_a"]},

geoportal/tests/functional/test_layers.py

@@ -71,7 +71,7 @@ def setup_method(self, _: Any) -> None:
         DBSession.add(self.role)
         DBSession.add(self.main)
 
-        create_default_ogcserver()
+        create_default_ogcserver(DBSession)
 
         transaction.commit()