(self, input_data, test_data)
| 89 | @pytest.mark.parametrize("input_data", error_input_list) |
| 90 | @pytest.mark.parametrize("test_data", generate_test_cases("rerank"), ids=lambda d: d["model_schema_id"]) |
| 91 | async def test_error_rerank(self, input_data, test_data): |
| 92 | request_data = {} |
| 93 | if input_data.get("model_schema_id") is None: |
| 94 | request_data.update({"model_schema_id": test_data["model_schema_id"]}) |
| 95 | if input_data.get("query") is None: |
| 96 | request_data.update({"query": self.query}) |
| 97 | if input_data.get("documents") is None: |
| 98 | request_data.update({"documents": self.documents}) |
| 99 | if input_data.get("top_n") is None: |
| 100 | request_data.update({"top_n": self.top_n}) |
| 101 | try: |
| 102 | res = await asyncio.wait_for(rerank(request_data), timeout=120) |
| 103 | except asyncio.TimeoutError: |
| 104 | pytest.skip("Skipping test due to timeout after 2 minutes.") |
| 105 | if is_provider_service_error(res): |
| 106 | pytest.skip(f"Skip the test case with provider service error.") |
| 107 | res_json = res.json() |
| 108 | assert res.status_code == 422, res.json() |
| 109 | assert res_json.get("status") == "error" |
| 110 | assert res_json.get("error").get("code") == "REQUEST_VALIDATION_ERROR" |
| 111 | |
| 112 | @pytest.mark.asyncio |
| 113 | @pytest.mark.test_id("inference_033") |
nothing calls this directly
no test coverage detected