Method test_validation

inference/test/test_validation.py:13–43

(self, test_data)

    @pytest.mark.asyncio
    @pytest.mark.test_id("inference_016-017")
    async def test_validation(self, test_data):
        model_schema_id = test_data["model_schema_id"]
        print("model_schema_id: ", model_schema_id)
        model_type = test_data["model_type"]

        credentials = {
            key: provider_credentials.aes_decrypt(test_data["credentials"][key])
            for key in test_data["credentials"].keys()
        }

        request_data = {"model_schema_id": model_schema_id, "model_type": model_type, "credentials": credentials}
        if "custom_host" in model_schema_id:
            request_data.update(
                {
                    "model_schema_id": "custom_host/openai-text-embedding",
                    "model_type": "text_embedding",
                    "credentials": {
                        "CUSTOM_HOST_ENDPOINT_URL": "https://api.openai.com/v1/embeddings",
                        "CUSTOM_HOST_MODEL_ID": "text-embedding-3-small",
                        "CUSTOM_HOST_API_KEY": Config.CUSTOM_HOST_API_KEY,
                    },
                    "properties": {"embedding_size": 1536},
                }
            )
        try:
            res = await asyncio.wait_for(verify_credentials(request_data), timeout=120)
        except asyncio.TimeoutError:
            pytest.skip("Skipping test due to timeout after 2 minutes.")
        assert res.status_code == 200, f"test_validation failed: result={res.json()}"
        assert res.json()["status"] == "success", f"test_validation failed: result={res.json()}"
        await asyncio.sleep(1)

    @pytest.mark.parametrize(
        "test_data", generate_wildcard_test_case_for_validation(), ids=lambda d: d["model_schema_id"]