chore: py3.12 compatibility

ckanext/ap_cron/logic/validators.py

@@ -37,7 +37,7 @@ def cron_job_exists(job_id: str, context: types.Context) -> Any:
 
     session = context["session"]
 
-    if not session.query(cron_model.CronJob).get(job_id):
+    if not session.get(cron_model.CronJob, job_id):
         raise tk.Invalid("The cron job not found.")
 
     return job_id

ckanext/ap_cron/tests/conftest.py

@@ -2,7 +2,6 @@
 from faker import Faker
 from pytest_factoryboy import register
 
-from ckan.plugins import load_all
 
 import ckanext.ap_cron.tests.factories as local_factories
 
@@ -14,17 +13,4 @@
 @pytest.fixture()
 def clean_db(reset_db, migrate_db_for):
     reset_db()
-    _migrate_plugins(migrate_db_for)
-
-
-@pytest.fixture(scope="session")
-def reset_db_once(reset_db, migrate_db_for):
-    """Dependency of `non_clean_db`"""
-    load_all()
-    reset_db()
-    _migrate_plugins(migrate_db_for)
-
-
-def _migrate_plugins(migrate_db_for):
-    migrate_db_for("admin_panel")
     migrate_db_for("admin_panel_cron")

ckanext/ap_cron/tests/logic/test_action.py

@@ -10,7 +10,7 @@
 from ckanext.ap_cron.types import DictizedCronJob
 
 
-@pytest.mark.usefixtures("with_plugins", "reset_db_once")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestCronJobCreate:
     def test_basic_create(self, cron_job_factory):
         job: DictizedCronJob = cron_job_factory()
@@ -28,7 +28,7 @@ def test_basic_create(self, cron_job_factory):
         assert isinstance(job["actions"], list)
 
 
-@pytest.mark.usefixtures("with_plugins", "reset_db_once")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestCronJobUpdate:
     def test_basic_update(self, cron_job: DictizedCronJob):
         assert cron_job["last_run"] is None
@@ -51,7 +51,7 @@ def test_basic_update(self, cron_job: DictizedCronJob):
         assert result["updated_at"] != cron_job["updated_at"]
 
 
-@pytest.mark.usefixtures("with_plugins", "reset_db_once")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestCronJobGet:
     def test_basic_get(self, cron_job: DictizedCronJob):
         result: DictizedCronJob = call_action("ap_cron_get_cron_job", id=cron_job["id"])
@@ -63,7 +63,7 @@ def test_basic_get_missing(self):
             call_action("ap_cron_get_cron_job", id="xxx")
 
 
-@pytest.mark.usefixtures("with_plugins", "reset_db_once")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestCronJobRemove:
     def test_basic_remove(self, cron_job: DictizedCronJob):
         result: DictizedCronJob = call_action(
@@ -98,7 +98,7 @@ def test_basic_list_filter_state(self, cron_job_factory):
         assert len(result) == 1
 
 
-@pytest.mark.usefixtures("with_plugins", "reset_db_once")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestCronJobRun:
     @mock.patch("ckan.plugins.toolkit.enqueue_job")
     def test_basic_run(self, enqueue_mock, cron_job: DictizedCronJob):

ckanext/ap_cron/tests/test_interface.py

@@ -6,7 +6,7 @@
 from ckan.tests import factories
 
 
-@pytest.mark.usefixtures("non_clean_db", "with_plugins")
+@pytest.mark.usefixtures("with_plugins", "clean_db")
 class TestApCronInterace(object):
     def test_action_without_exclude(self, app, sysadmin):
         url = tk.h.url_for("ap_cron.action_autocomplete")
@@ -15,9 +15,9 @@ def test_action_without_exclude(self, app, sysadmin):
 
         result = app.get(
             url=url,
-            headers={"Authorization": user_token["token"]},  # type: ignore
+            headers={"Authorization": user_token["token"]},
             query_string={"incomplete": "test"},
         ).json
 
         # I've excluded "ap_cron_test_action_2" in ApCronTestPlugin
-        assert len(result["ResultSet"]["Result"]) == 2
+        assert len(result["ResultSet"]["Result"]) == 1

ckanext/ap_log/log_handlers.py

@@ -2,7 +2,7 @@
 
 import logging
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import sessionmaker
 
 from ckan.plugins import toolkit as tk
@@ -19,8 +19,8 @@ def __init__(self, db_uri: str):
         super().__init__()
 
         engine = create_engine(db_uri)
-
-        if not engine.has_table(ApLogs.__tablename__):
+        inspector = inspect(engine)
+        if not inspector.has_table(ApLogs.__tablename__):
             self.not_ready = True
             log.error("The ApLogs table is not initialized")
 

pyproject.toml

@@ -14,7 +14,6 @@ sections = "FUTURE,STDLIB,FIRSTPARTY,THIRDPARTY,CKAN,CKANEXT,SELF,LOCALFOLDER"
 [tool.pytest.ini_options]
 addopts = "--ckan-ini test.ini"
 filterwarnings = [
-               "ignore::sqlalchemy.exc.SADeprecationWarning",
                "ignore::sqlalchemy.exc.SAWarning",
                "ignore::DeprecationWarning",
 ]