[Serving] Fail RemoteStep on bad status code also in async mode (#1474)

mlrun/serving/remote.py

@@ -116,6 +116,12 @@ async def _process_event(self, event):
 
     async def _handle_completed(self, event, response):
         response_body = await response.read()
+        if response.status >= 400:
+            raise ValueError(
+                f"For event {event}, RemoteStep {self.name} got an unexpected response "
+                f"status {response.status}: {response_body}"
+            )
+
         body = self._get_data(response_body, response.headers)
 
         if body is not None:
@@ -138,7 +144,7 @@ def do_event(self, event):
         except OSError as err:
             raise OSError(f"error: cannot invoke url: {url}, {err}")
         if not resp.ok:
-            raise RuntimeError(f"bad http response {resp.text}")
+            raise RuntimeError(f"bad http response {resp.status_code}: {resp.text}")
 
         result = self._get_data(resp.content, resp.headers)
         event.body = _update_result_body(self._result_path, event.body, result)

requirements.txt

@@ -63,7 +63,7 @@ fsspec~=2021.8.1
 v3iofs~=0.1.7
 # 3.4 and above failed builidng in some images - see https://github.com/pyca/cryptography/issues/5771
 cryptography~=3.0, <3.4
-storey~=0.8.6; python_version >= '3.7'
+storey~=0.8.7; python_version >= '3.7'
 deepdiff~=5.0
 pymysql~=1.0
 inflection~=0.5.0

tests/serving/test_remote.py

@@ -67,6 +67,39 @@ def test_remote_step(httpserver, engine):
     assert resp == {"foo": "ok"}
 
 
+@pytest.mark.parametrize("engine", ["async"])
+def test_remote_step_bad_status_code(httpserver, engine):
+    httpserver.expect_request("/", method="GET").respond_with_json({"get": "ok"})
+    httpserver.expect_request("/foo", method="GET").respond_with_json({}, status=400)
+
+    # verify the remote step added headers for the event path and id
+    expected_headers = {
+        event_id_key: "123",
+        event_path_key: "/datapath",
+    }
+    httpserver.expect_request(
+        "/data", method="POST", data="req text", headers=expected_headers
+    ).respond_with_data("my str")
+    httpserver.expect_request("/json", method="POST", json={"x": 5}).respond_with_json(
+        {"post": "ok"}
+    )
+    url = httpserver.url_for("/")
+
+    for params, request, expected in tests_map:
+        print(f"test params: {params}")
+        server = _new_server(url, engine, **params)
+        resp = server.test(**request)
+        server.wait_for_completion()
+        assert resp == expected
+
+    # test with url generated with expression (from the event)
+    server = _new_server(None, engine, method="GET", url_expression="event['myurl']")
+    with pytest.raises(RuntimeError):
+        server.test(body={"myurl": httpserver.url_for("/foo")})
+    with pytest.raises(ValueError):
+        server.wait_for_completion()
+
+
 @pytest.mark.parametrize("engine", ["sync", "async"])
 def test_remote_class(httpserver, engine):
     from mlrun.serving.remote import RemoteStep