Set up the transport for reposting failed calls and bump supported python version (#668)

Ugly requirement coming from the way the workflows transport is set up and how the murfey transport is defined. Will want revisiting in future.

Leads to a bump in the python version as this requires workflows 3

Author: stephen-riggs

.github/workflows/ci.yml

@@ -14,7 +14,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.9", "3.10", "3.11"]
+        python-version: ["3.10", "3.11", "3.12"]
     services:
       mariadb:
         image: mariadb:11.7.2 # released 2024-05-06

pyproject.toml

@@ -16,15 +16,14 @@ keywords = [
 license = {file = "LICENSE"}
 maintainers = []
 authors = []
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Environment :: Console",
     "Intended Audience :: Science/Research",
     "License :: OSI Approved :: BSD License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",

src/murfey/cli/repost_failed_calls.py

@@ -10,22 +10,8 @@
 from sqlmodel import Session, create_engine
 from workflows.transport.pika_transport import PikaTransport
 
-import murfey.server.api.auth
-import murfey.server.api.bootstrap
-import murfey.server.api.clem
-import murfey.server.api.display
-import murfey.server.api.file_io_frontend
-import murfey.server.api.file_io_instrument
-import murfey.server.api.hub
-import murfey.server.api.instrument
-import murfey.server.api.mag_table
-import murfey.server.api.processing_parameters
-import murfey.server.api.prometheus
-import murfey.server.api.session_control
-import murfey.server.api.session_info
-import murfey.server.api.websocket
-import murfey.server.api.workflow
 from murfey.server.murfey_db import url
+from murfey.server.run import _set_up_transport
 from murfey.util.config import security_from_file
 
 
@@ -104,6 +90,23 @@ def handle_dlq_messages(messages_path: list[Path], rabbitmq_credentials: Path):
 
 def handle_failed_posts(messages_path: list[Path], murfey_db: Session):
     """Deal with any messages that have been sent as failed client posts"""
+    # These imports need to happen after transport object is configured
+    import murfey.server.api.auth
+    import murfey.server.api.bootstrap
+    import murfey.server.api.clem
+    import murfey.server.api.display
+    import murfey.server.api.file_io_frontend
+    import murfey.server.api.file_io_instrument
+    import murfey.server.api.hub
+    import murfey.server.api.instrument
+    import murfey.server.api.mag_table
+    import murfey.server.api.processing_parameters
+    import murfey.server.api.prometheus
+    import murfey.server.api.session_control
+    import murfey.server.api.session_info
+    import murfey.server.api.websocket
+    import murfey.server.api.workflow
+
     for json_file in messages_path:
         with open(json_file, "r") as json_data:
             message = json.load(json_data)
@@ -181,6 +184,15 @@ def run():
     # Read the security config file
     security_config = security_from_file(args.config)
 
+    # Configure the transport
+    PikaTransport().load_configuration_file(security_config.rabbitmq_credentials)
+    _set_up_transport("PikaTransport")
+
+    # Now import transport object which was set up in the above step
+    from murfey.server import _transport_object
+
+    _transport_object.feedback_queue = security_config.feedback_queue
+
     # Purge the queue and repost/reinject any messages found
     dlq_dump_path = Path(args.dir)
     dlq_dump_path.mkdir(parents=True, exist_ok=True)

tests/cli/test_repost_failed_calls.py

@@ -187,6 +187,8 @@ def test_handle_failed_posts(tmp_path):
     assert (tmp_path / "msg6").is_file()  # function does not exist
 
 
+@mock.patch("workflows.transport.pika_transport.PikaTransport")
+@mock.patch("murfey.cli.repost_failed_calls.PikaTransport")
 @mock.patch("murfey.cli.repost_failed_calls.dlq_purge")
 @mock.patch("murfey.cli.repost_failed_calls.handle_failed_posts")
 @mock.patch("murfey.cli.repost_failed_calls.handle_dlq_messages")
@@ -200,6 +202,8 @@ def test_run_repost_failed_calls(
     mock_reinject,
     mock_repost,
     mock_purge,
+    mock_pika,
+    mock_workflows,
     mock_security_configuration,
 ):
     mock_session = mock.MagicMock()
@@ -227,6 +231,11 @@ def test_run_repost_failed_calls(
     mock_db_engine.assert_called_with("db_url")
     mock_db_session.assert_called_with("db_engine")
 
+    mock_pika().load_configuration_file.assert_called_with(
+        Path("/path/to/rabbitmq.yaml")
+    )
+    mock_workflows().connect.assert_called_once()
+
     mock_purge.assert_called_once_with(
         Path("DLQ_dir"),
         "murfey_feedback",

tests/server/test_main.py

@@ -24,7 +24,7 @@ def login(test_user):
             f"{url_path_for('auth.router', 'generate_token')}",
             data=test_user,
         )
-        assert mock_validate.called_once()
+        mock_validate.assert_called_once()
         assert response.status_code == 200
         token = response.json()["access_token"]
         assert token is not None
@@ -35,10 +35,9 @@ def login(test_user):
 def test_read_main(mock_check, test_user):
     token = login(test_user)
     response = client.get(
-        "/session_info/connections", 
-        headers={"Authorization": f"Bearer {token}"}
+        "/session_info/connections", headers={"Authorization": f"Bearer {token}"}
     )
-    assert mock_check.called_once()
+    mock_check.assert_called_once()
     assert response.status_code == 200