Fix SA2.0 syntax (query->select) in integration tests

test/integration/objectstore/test_selection_with_resource_parameters.py

@@ -3,6 +3,8 @@
 import os
 import string
 
+from sqlalchemy import select
+
 from galaxy.model import Dataset
 from galaxy.model.unittest_utils.store_fixtures import (
     deferred_hda_model_store_dict,
@@ -87,7 +89,7 @@ def _assert_file_counts(self, default, static, dynamic_ebs, dynamic_s3):
 
     def _assert_no_external_filename(self):
         # Should maybe be its own test case ...
-        for external_filename_tuple in self._app.model.session.query(Dataset.external_filename).all():
+        for external_filename_tuple in self._app.model.session.scalars(select(Dataset.external_filename)).all():
             assert external_filename_tuple[0] is None
 
     def test_objectstore_selection(self):
@@ -181,5 +183,7 @@ def _run_tool(tool_id, inputs):
 
     @property
     def _latest_dataset(self):
-        latest_dataset = self._app.model.session.query(Dataset).order_by(Dataset.table.c.id.desc()).first()
+        latest_dataset = self._app.model.session.scalars(
+            select(Dataset).order_by(Dataset.table.c.id.desc()).limit(1)
+        ).first()
         return latest_dataset

test/integration/objectstore/test_selection_with_user_preferred_object_store.py

@@ -8,6 +8,8 @@
     Optional,
 )
 
+from sqlalchemy import select
+
 from galaxy.model import Dataset
 from galaxy_test.base.populators import WorkflowPopulator
 from galaxy_test.base.workflow_fixtures import (
@@ -482,5 +484,7 @@ def _create_hda_get_storage_info(self, history_id: str):
 
     @property
     def _latest_dataset(self):
-        latest_dataset = self._app.model.session.query(Dataset).order_by(Dataset.table.c.id.desc()).first()
+        latest_dataset = self._app.model.session.scalars(
+            select(Dataset).order_by(Dataset.table.c.id.desc()).limit(1)
+        ).first()
         return latest_dataset

test/integration/test_celery_tasks.py

@@ -1,6 +1,7 @@
 import tarfile
 
 from celery import shared_task
+from sqlalchemy import select
 
 from galaxy.celery import galaxy_task
 from galaxy.celery.tasks import (
@@ -108,9 +109,5 @@ def test_import_export_history_contents(self):
 
     @property
     def _latest_hda(self):
-        latest_hda = (
-            self._app.model.session.query(HistoryDatasetAssociation)
-            .order_by(HistoryDatasetAssociation.table.c.id.desc())
-            .first()
-        )
-        return latest_hda
+        stmt = select(HistoryDatasetAssociation).order_by(HistoryDatasetAssociation.table.c.id.desc()).limit(1)
+        return self._app.model.session.scalars(stmt).first()

test/integration/test_job_files.py

@@ -19,6 +19,7 @@
 import tempfile
 
 import requests
+from sqlalchemy import select
 
 from galaxy import model
 from galaxy.model.base import transaction
@@ -50,10 +51,11 @@ def setUp(self):
         if not TestJobFilesIntegration.initialized:
             history_id = self.dataset_populator.new_history()
             sa_session = self.sa_session
-            assert len(sa_session.query(model.HistoryDatasetAssociation).all()) == 0
+            stmt = select(model.HistoryDatasetAssociation)
+            assert len(sa_session.scalars(stmt).all()) == 0
             self.dataset_populator.new_dataset(history_id, content=TEST_INPUT_TEXT, wait=True)
-            assert len(sa_session.query(model.HistoryDatasetAssociation).all()) == 1
-            self.input_hda = sa_session.query(model.HistoryDatasetAssociation).all()[0]
+            assert len(sa_session.scalars(stmt).all()) == 1
+            self.input_hda = sa_session.scalars(stmt).all()[0]
             TestJobFilesIntegration.initialized = True
 
     def test_read_by_state(self):
@@ -119,11 +121,11 @@ def sa_session(self):
     def create_static_job_with_state(self, state):
         """Create a job with unknown handler so its state won't change."""
         sa_session = self.sa_session
-        hda = sa_session.query(model.HistoryDatasetAssociation).all()[0]
+        hda = sa_session.scalars(select(model.HistoryDatasetAssociation)).all()[0]
         assert hda
-        history = sa_session.query(model.History).all()[0]
+        history = sa_session.scalars(select(model.History)).all()[0]
         assert history
-        user = sa_session.query(model.User).all()[0]
+        user = sa_session.scalars(select(model.User)).all()[0]
         assert user
         output_hda = model.HistoryDatasetAssociation(history=history, create_dataset=True, flush=False)
         output_hda.hid = 2

test/integration/test_kubernetes_runner.py

@@ -269,8 +269,8 @@ def test_kill_process(self) -> None:
             job_dict = running_response["jobs"][0]
 
             app = self._app
-            sa_session = app.model.context.current
-            job = sa_session.query(app.model.Job).get(app.security.decode_id(job_dict["id"]))
+            sa_session = app.model.session
+            job = sa_session.get(app.model.Job, app.security.decode_id(job_dict["id"]))
 
             self._wait_for_external_state(sa_session, job, app.model.Job.states.RUNNING)
             assert not job.finished
@@ -306,8 +306,8 @@ def test_external_job_delete(self) -> None:
             job_dict = running_response.json()["jobs"][0]
 
             app = self._app
-            sa_session = app.model.context.current
-            job = sa_session.query(app.model.Job).get(app.security.decode_id(job_dict["id"]))
+            sa_session = app.model.session
+            job = sa_session.get(app.model.Job, app.security.decode_id(job_dict["id"]))
 
             self._wait_for_external_state(sa_session, job, app.model.Job.states.RUNNING)
 
@@ -340,8 +340,8 @@ def test_exit_code_127(self, history_id: str) -> None:
         # check that logs are also available in job logs
         app = self._app
         job_id = app.security.decode_id(running_response.json()["jobs"][0]["id"])
-        sa_session = app.model.context
-        job = sa_session.query(app.model.Job).get(job_id)
+        sa_session = app.model.session
+        job = sa_session.get(app.model.Job, job_id)
         self._wait_for_external_state(sa_session=sa_session, job=job, expected=app.model.Job.states.RUNNING)
 
         external_id = job.job_runner_external_id

test/integration/test_page_revision_json_encoding.py

@@ -5,6 +5,8 @@
 exported API values are encoded though.
 """
 
+from sqlalchemy import select
+
 from galaxy import model
 from galaxy_test.base import api_asserts
 from galaxy_test.base.populators import DatasetPopulator
@@ -26,8 +28,8 @@ def test_page_encoding(self, history_id: str):
         )
         page_response = self._post("pages", request, json=True)
         api_asserts.assert_status_code_is_ok(page_response)
-        sa_session = self._app.model.context
-        page_revision = sa_session.query(model.PageRevision).filter_by(content_format="html").all()[0]
+        sa_session = self._app.model.session
+        page_revision = sa_session.scalars(select(model.PageRevision).filter_by(content_format="html")).all()[0]
         assert '''id="History-1"''' in page_revision.content, page_revision.content
         assert f'''id="History-{history_id}"''' not in page_revision.content, page_revision.content
 
@@ -52,8 +54,8 @@ def test_page_encoding_markdown(self, history_id: str):
         )
         page_response = self._post("pages", request, json=True)
         api_asserts.assert_status_code_is_ok(page_response)
-        sa_session = self._app.model.context
-        page_revision = sa_session.query(model.PageRevision).filter_by(content_format="markdown").all()[0]
+        sa_session = self._app.model.session
+        page_revision = sa_session.scalars(select(model.PageRevision).filter_by(content_format="markdown")).all()[0]
         assert (
             """```galaxy
 history_dataset_display(history_dataset_id=1)

test/integration/test_repository_operations.py

@@ -1,6 +1,8 @@
 import os
 from collections import namedtuple
 
+from sqlalchemy import select
+
 from galaxy.model.base import transaction
 from galaxy_test.base.populators import DatasetPopulator
 from galaxy_test.driver import integration_util
@@ -86,7 +88,7 @@ def test_repository_update(self):
         hg_util.update_repository(repository_path, ctx_rev="3")
         # change repo to revision 3 in database
         model = self._app.install_model
-        tsr = model.context.query(model.ToolShedRepository).first()
+        tsr = model.session.scalars(select(model.ToolShedRepository).limit(1)).first()
         assert tsr.name == REPO.name
         assert tsr.changeset_revision == latest_revision
         assert int(tsr.ctx_rev) >= 4

test/integration/test_save_job_id_on_datasets.py

@@ -8,6 +8,7 @@
 """
 
 import pytest
+from sqlalchemy import select
 
 from galaxy import model
 from galaxy_test.driver.driver_util import GalaxyTestDriver
@@ -39,8 +40,8 @@ def test_driver():
 def test_tool_datasets(tool_id, test_driver):
     test_driver.run_tool_test(tool_id)
     session = test_driver.app.model.context.current
-    job = session.query(model.Job).order_by(model.Job.id.desc()).first()
-    datasets = session.query(model.Dataset).filter(model.Dataset.job_id == job.id).all()
+    job = session.scalars(select(model.Job).order_by(model.Job.id.desc()).limit(1)).first()
+    datasets = session.scalars(select(model.Dataset).filter(model.Dataset.job_id == job.id)).all()
 
     if tool_id == "boolean_conditional":
         assert len(datasets) == 1

test/integration/test_user_preferences.py

@@ -7,6 +7,7 @@
     get,
     put,
 )
+from sqlalchemy import select
 
 from galaxy_test.driver import integration_util
 
@@ -18,7 +19,8 @@ def test_user_theme(self):
         user = self._setup_user(TEST_USER_EMAIL)
         url = self._api_url(f"users/{user['id']}/theme/test_theme", params=dict(key=self.master_api_key))
         app = cast(Any, self._test_driver.app if self._test_driver else None)
-        db_user = app.model.context.query(app.model.User).filter(app.model.User.email == user["email"]).first()
+        stmt = select(app.model.User).filter(app.model.User.email == user["email"]).limit(1)
+        db_user = app.model.session.scalars(stmt).first()
 
         # create some initial data
         put(url)

test/integration/test_workflow_handler_configuration.py

@@ -127,8 +127,7 @@ def _get_workflow_invocations(self, history_id: str):
         # into Galaxy's internal state.
         app = self._app
         history_id = app.security.decode_id(history_id)
-        sa_session = app.model.context.current
-        history = sa_session.query(app.model.History).get(history_id)
+        history = app.model.session.get(app.model.History, history_id)
         workflow_invocations = history.workflow_invocations
         return workflow_invocations