example_mock_openai_client_endpoint.py

example_mock_openai_client_endpoint.py

@pytest.mark.asyncio
async def test_get_openai_models(simple_test_settings: Settings):
    async with create_test_server() as server:
        # Create Mock OpenAI Models endpoint
        @server.app.get("/openai/models", status_code=200)
        async def get_models():
            # Create models list with random timestamps
            models_data = []
            end_date = datetime(2024, 3, 1)

            for supported_model in AssistantSupportedModels:
                random_days = random.randint(0, 1520)
                random_date = end_date - timedelta(days=random_days)
                timestamp = int(random_date.timestamp())

                models_data.append(
                    ModelModel(
                        id=supported_model.value,
                        created=timestamp,
                        object=Object29.model,
                        owned_by="azure",
                    )
                )

            # Add custom DeepSeek-R1 model
            models_data.append(
                ModelModel(
                    id="DeepSeek-R1",
                    created=int(datetime(2023, 11, 15).timestamp()),  # Fixed date for custom model
                    object=Object29.model,
                    owned_by="deepseek"
                )
            )

            return ListModelsResponse(
                object=Object20.list,
                data=models_data
            )

        # Deal with response
        response = await server.client.get("/openai/models")
        _, data = await asyncio.gather(
            response.expect_status(200),
            response.json()
        )
        # logger.info(f"\n{json.dumps(data, indent=4)}")

        # Try with Azure OpenAI client
        azure_openai = AsyncAzureOpenAI(
            azure_endpoint=server.base_url,
            api_key="TEST_API_KEY",
            api_version="2023-05-15"
        )

        ai_model = OpenAIModel('DeepSeek-R1', openai_client=azure_openai)

        # Get models list
        models_page: AsyncPage[Model] = await ai_model.client.models.list()

        # Collect all models from the paginator
        all_models: List[dict] = []
        async for model in models_page:
            # Cast to Model to make typechecker happy
            model_: Model = Model.model_validate(model.__dict__)  # todo: suspect pydantic v1 here from azure openai
            all_models.append(model_.model_dump())

        # Create the expected structure
        response_data = {
            "object": "list",
            "data": all_models
        }

        logger.info(f"\nAzure supported models:\n{json.dumps(response_data, indent=4)}")