(self, test_data)
| 55 | ) |
| 56 | @pytest.mark.flaky(reruns=3, reruns_delay=1) |
| 57 | async def test_rerank(self, test_data): |
| 58 | model_schema_id = test_data["model_schema_id"] |
| 59 | |
| 60 | request_data = { |
| 61 | "model_schema_id": model_schema_id, |
| 62 | "query": self.query, |
| 63 | "documents": self.documents, |
| 64 | "top_n": self.top_n, |
| 65 | } |
| 66 | try: |
| 67 | res = await asyncio.wait_for(rerank(request_data), timeout=120) |
| 68 | except asyncio.TimeoutError: |
| 69 | pytest.skip("Skipping test due to timeout after 2 minutes.") |
| 70 | if is_provider_service_error(res): |
| 71 | pytest.skip(f"Skip the test case with provider service error.") |
| 72 | res_json = res.json() |
| 73 | assert res.status_code == 200, res.json() |
| 74 | assert res_json.get("status") == "success" |
| 75 | results = res_json.get("data").get("results") |
| 76 | assert len(results) == self.top_n |
| 77 | assert check_order(results, "relevance_score") |
| 78 | for result in results: |
| 79 | assert result.get("document").get("text") in self.documents |
| 80 | assert result.get("relevance_score") >= 0.0 |
| 81 | if "jina-colbert-v1-en" in model_schema_id: |
| 82 | assert result.get("relevance_score") <= 10.0 |
| 83 | else: |
| 84 | assert result.get("relevance_score") <= 1.0 |
| 85 | assert result.get("index") == self.documents.index(result.get("document").get("text")) |
| 86 | |
| 87 | @pytest.mark.test_id("inference_032") |
| 88 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected