Handle the async case

Author: stephen-riggs

src/murfey/cli/repost_failed_calls.py

@@ -1,8 +1,9 @@
 import argparse
+import asyncio
 import json
 from datetime import datetime
 from functools import partial
-from inspect import getfullargspec
+from inspect import getfullargspec, iscoroutinefunction
 from pathlib import Path
 from queue import Empty, Queue
 
@@ -106,9 +107,9 @@ def handle_failed_posts(messages_path: list[Path], murfey_db: Session):
     for json_file in messages_path:
         with open(json_file, "r") as json_data:
             message = json.load(json_data)
-        router_name = message.get("router_name", "")
+        router_name = message.get("message", {}).get("router_name", "")
         router_base = router_name.split(".")[0]
-        function_name = message.get("function_name", "")
+        function_name = message.get("message", {}).get("function_name", "")
         if not router_name or not function_name:
             print(
                 f"Cannot repost {json_file} as it does not have a router or function name"
@@ -122,22 +123,31 @@ def handle_failed_posts(messages_path: list[Path], murfey_db: Session):
         except AttributeError:
             print(f"Cannot repost {json_file} as {function_name} does not exist")
             continue
-        expected_args = getfullargspec(function_to_call).args
+        expected_args = getfullargspec(function_to_call)
 
-        call_kwargs = message.get("kwargs", {})
-        call_data = message.get("data", {})
+        call_kwargs = message.get("message", {}).get("kwargs", {})
+        call_data = message.get("message", {}).get("data", {})
         function_call_dict = {}
 
-        for call_arg in expected_args:
-            if call_arg in call_kwargs.keys():
-                function_call_dict[call_arg] = call_kwargs[call_arg]
-            elif call_arg == "db":
-                function_call_dict["db"] = murfey_db
-            else:
-                function_call_dict[call_arg] = call_data
+        try:
+            for call_arg in expected_args.args:
+                call_arg_type = expected_args.annotations.get(call_arg, str)
+                if call_arg in call_kwargs.keys():
+                    function_call_dict[call_arg] = call_arg_type(call_kwargs[call_arg])
+                elif call_arg == "db":
+                    function_call_dict["db"] = murfey_db
+                else:
+                    print(call_data, call_arg_type, call_arg)
+                    function_call_dict[call_arg] = call_arg_type(**call_data)
+        except TypeError as e:
+            print(f"Cannot repost {json_file} due to argument error: {e}")
+            continue
 
         try:
-            function_to_call(**function_call_dict)
+            if iscoroutinefunction(function_to_call):
+                asyncio.run(function_to_call(**function_call_dict))
+            else:
+                function_to_call(**function_call_dict)
             print(f"Reposted {json_file}")
             json_file.unlink()
         except Exception as e:

tests/cli/test_repost_failed_calls.py

@@ -6,6 +6,8 @@
 from unittest import mock
 
 from murfey.cli import repost_failed_calls
+from murfey.util.config import security_from_file
+from murfey.util.db import Tilt
 
 
 @mock.patch("murfey.cli.repost_failed_calls.PikaTransport")
@@ -104,101 +106,124 @@ def test_handle_dlq_messages(mock_transport, tmp_path):
     mock_transport().disconnect.assert_called_once()
 
 
-@mock.patch("murfey.cli.repost_failed_calls.requests")
-def test_handle_failed_posts(mock_requests, tmp_path):
+def test_handle_failed_posts(tmp_path):
     """Test that the API is called with any failed client post messages"""
     # Create some sample messages
     messages_paths_list: list[Path] = []
     messages_dict: dict[str, dict] = {
         "msg1": {
-            "message": {"url": "sample/url", "json": {"content": "msg1"}},
-        },
+            "message": {
+                "router_name": "workflow.tomo_router",
+                "function_name": "register_completed_tilt_series",
+                "kwargs": {"visit_name": "cm12345-1", "session_id": 1},
+                "data": {
+                    "tags": ["tag"],
+                    "source": "source",
+                    "tilt_series_lengths": [10],
+                },
+            },
+        },  # normal example
         "msg2": {
-            "message": {"url": "sample/url", "json": {"content": "msg2"}},
-        },
+            "message": {
+                "router_name": "workflow.tomo_router",
+                "function_name": "register_tilt",
+                "kwargs": {"visit_name": "cm12345-1", "session_id": 1},
+                "data": {
+                    "tilt_series_tag": "tag",
+                    "source": "source",
+                    "movie_path": "path",
+                },
+            },
+        },  # async example
         "msg3": {
-            "message": {"content": "msg3"},  # not a failed client post
+            "message": {
+                "router_name": "workflow.tomo_router",
+                "function_name": "register_completed_tilt_series",
+                "data": {"tags": ["tag"]},
+            }
         },
         "msg4": {
-            "header": {"content": "msg3"},  # does not have a message
+            "message": {"function_name": "dummy"},  # does not have a router
+        },
+        "msg5": {
+            "message": {"router_name": "workflow"},  # does not have a function
+        },
+        "msg6": {
+            "message": {
+                "router_name": "workflow",
+                "function_name": "dummy",
+            },  # function does not exist
         },
     }
     for file_name, message in messages_dict.items():
         messages_paths_list.append(tmp_path / file_name)
         with open(tmp_path / file_name, "w") as msg_file:
             json.dump(message, msg_file)
 
-    class Response:
-        def __init__(self, status_code):
-            self.status_code = status_code
-
-    mock_requests.post.side_effect = [Response(200), Response(300)]
-
-    repost_failed_calls.handle_failed_posts(messages_paths_list, "dummy_token")
+    mock_db = mock.Mock()
+    mock_exec_return = mock.Mock()
+    mock_exec_return.all.return_value = []
+    mock_db.exec.return_value = mock_exec_return
+    repost_failed_calls.handle_failed_posts(messages_paths_list, mock_db)
 
     # Check the failed posts were resent
-    assert mock_requests.post.call_count == 2
-    mock_requests.post.assert_any_call(
-        "sample/url",
-        json={"content": "msg1"},
-        headers={"Authorization": "Bearer dummy_token"},
-    )
-    mock_requests.post.assert_any_call(
-        "sample/url",
-        json={"content": "msg2"},
-        headers={"Authorization": "Bearer dummy_token"},
+    assert mock_db.exec.call_count == 3
+    assert mock_db.exec().one.call_count == 1
+    assert mock_db.exec().all.call_count == 2
+    assert mock_exec_return.one.call_count == 1
+    assert mock_exec_return.all.call_count == 2
+    assert mock_db.commit.call_count == 3
+    mock_db.add.assert_called_once_with(
+        Tilt(movie_path="path", tilt_series_id=mock.ANY, motion_corrected=False)
     )
 
     # Check only the failed post which was successfully reinjected got deleted
     assert not (tmp_path / "msg1").is_file()  # got resent
-    assert (tmp_path / "msg2").is_file()  # failed reinjection
-    assert (tmp_path / "msg3").is_file()  # not a failed client post
-    assert (tmp_path / "msg4").is_file()  # does not have a message
+    assert not (tmp_path / "msg2").is_file()  # got resent
+    assert (tmp_path / "msg3").is_file()  # failed reinjection
+    assert (tmp_path / "msg4").is_file()  # does not have a router
+    assert (tmp_path / "msg5").is_file()  # does not have a function
+    assert (tmp_path / "msg6").is_file()  # function does not exist
 
 
 @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")
-@mock.patch("murfey.cli.repost_failed_calls.jwt")
+@mock.patch("murfey.cli.repost_failed_calls.get_murfey_db_session")
 def test_run_repost_failed_calls(
-    mock_jwt,
+    mock_db,
     mock_reinject,
     mock_repost,
     mock_purge,
     mock_security_configuration,
 ):
-    mock_jwt.encode.return_value = "dummy_token"
+    mock_db.return_value = "db"
     mock_purge.return_value = ["/path/to/msg1"]
 
     config_file = mock_security_configuration
     with open(config_file) as f:
-        security_config = json.load(f)
+        security_config_dict = json.load(f)
 
     sys.argv = [
         "murfey.repost_failed_calls",
         "--config",
         str(config_file),
-        "--username",
-        "user",
         "--dir",
         "DLQ_dir",
     ]
     repost_failed_calls.run()
 
-    mock_jwt.encode.assert_called_with(
-        {"user": "user"},
-        security_config["auth_key"],
-        algorithm=security_config["auth_algorithm"],
-    )
+    security_config_class = security_from_file(config_file)
+    mock_db.assert_called_with(security_config_class)
 
     mock_purge.assert_called_once_with(
         Path("DLQ_dir"),
         "murfey_feedback",
-        Path(security_config["rabbitmq_credentials"]),
+        Path(security_config_dict["rabbitmq_credentials"]),
     )
-    mock_repost.assert_called_once_with(["/path/to/msg1"], "dummy_token")
+    mock_repost.assert_called_once_with(["/path/to/msg1"], "db")
     mock_reinject.assert_called_once_with(
-        ["/path/to/msg1"], Path(security_config["rabbitmq_credentials"])
+        ["/path/to/msg1"], Path(security_config_dict["rabbitmq_credentials"])
     )
 
 
@@ -218,6 +243,4 @@ def test_repost_failed_calls_exists():
     cleaned_help_line = (
         stdout_as_string.split("\n\n")[0].replace("\n", "").replace(" ", "")
     )
-    assert cleaned_help_line == (
-        "usage:murfey.repost_failed_calls[-h]-cCONFIG-uUSERNAME[-dDIR]"
-    )
+    assert cleaned_help_line == ("usage:murfey.repost_failed_calls[-h]-cCONFIG[-dDIR]")

tests/client/test_context.py

@@ -26,6 +26,7 @@ def test_tomography_context_add_tomo_tilt(mock_post, mock_get, tmp_path):
         default_destinations={tmp_path: str(tmp_path)},
         instrument_name="",
         visit="test",
+        murfey_session=1,
     )
     context = TomographyContext("tomo", tmp_path)
     (tmp_path / "Position_1_001_[30.0]_date_time_fractions.tiff").touch()
@@ -82,6 +83,7 @@ def test_tomography_context_add_tomo_tilt_out_of_order(mock_post, mock_get, tmp_
         default_destinations={tmp_path: str(tmp_path)},
         instrument_name="",
         visit="test",
+        murfey_session=1,
     )
     context = TomographyContext("tomo", tmp_path)
     (tmp_path / "Position_1_001_[30.0]_date_time_fractions.tiff").touch()
@@ -166,6 +168,7 @@ def test_tomography_context_add_tomo_tilt_delayed_tilt(mock_post, mock_get, tmp_
         default_destinations={tmp_path: str(tmp_path)},
         instrument_name="",
         visit="test",
+        murfey_session=1,
     )
     context = TomographyContext("tomo", tmp_path)
     (tmp_path / "Position_1_001_[30.0]_date_time_fractions.tiff").touch()
@@ -230,6 +233,7 @@ def test_setting_tilt_series_size_and_completion_from_mdoc_parsing(
         default_destinations={tmp_path: str(tmp_path)},
         instrument_name="",
         visit="test",
+        murfey_session=1,
     )
     context = TomographyContext("tomo", tmp_path)
     assert len(context._tilt_series_sizes) == 0