more

.scripts/community_split/libs/partners/openai/tests/integration_tests/chat_models/test_azure.py

@@ -0,0 +1,145 @@
+"""Test AzureChatOpenAI wrapper."""
+import os
+from typing import Any
+
+import pytest
+from langchain_core.messages import BaseMessage, HumanMessage
+from langchain_core.outputs import ChatGeneration, ChatResult, LLMResult
+
+from langchain_openai.chat_models import AzureChatOpenAI
+
+OPENAI_API_VERSION = os.environ.get("AZURE_OPENAI_API_VERSION", "")
+OPENAI_API_BASE = os.environ.get("AZURE_OPENAI_API_BASE", "")
+OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_API_KEY", "")
+DEPLOYMENT_NAME = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME", "")
+
+
+def _get_llm(**kwargs: Any) -> AzureChatOpenAI:
+    return AzureChatOpenAI(
+        deployment_name=DEPLOYMENT_NAME,
+        openai_api_version=OPENAI_API_VERSION,
+        openai_api_base=OPENAI_API_BASE,
+        openai_api_key=OPENAI_API_KEY,
+        **kwargs,
+    )
+
+
+@pytest.mark.scheduled
+@pytest.fixture
+def llm() -> AzureChatOpenAI:
+    return _get_llm(
+        max_tokens=10,
+    )
+
+
+def test_chat_openai(llm: AzureChatOpenAI) -> None:
+    """Test AzureChatOpenAI wrapper."""
+    message = HumanMessage(content="Hello")
+    response = llm([message])
+    assert isinstance(response, BaseMessage)
+    assert isinstance(response.content, str)
+
+
+@pytest.mark.scheduled
+def test_chat_openai_generate() -> None:
+    """Test AzureChatOpenAI wrapper with generate."""
+    chat = _get_llm(max_tokens=10, n=2)
+    message = HumanMessage(content="Hello")
+    response = chat.generate([[message], [message]])
+    assert isinstance(response, LLMResult)
+    assert len(response.generations) == 2
+    for generations in response.generations:
+        assert len(generations) == 2
+        for generation in generations:
+            assert isinstance(generation, ChatGeneration)
+            assert isinstance(generation.text, str)
+            assert generation.text == generation.message.content
+
+
+@pytest.mark.scheduled
+def test_chat_openai_multiple_completions() -> None:
+    """Test AzureChatOpenAI wrapper with multiple completions."""
+    chat = _get_llm(max_tokens=10, n=5)
+    message = HumanMessage(content="Hello")
+    response = chat._generate([message])
+    assert isinstance(response, ChatResult)
+    assert len(response.generations) == 5
+    for generation in response.generations:
+        assert isinstance(generation.message, BaseMessage)
+        assert isinstance(generation.message.content, str)
+
+
+@pytest.mark.scheduled
+async def test_async_chat_openai() -> None:
+    """Test async generation."""
+    chat = _get_llm(max_tokens=10, n=2)
+    message = HumanMessage(content="Hello")
+    response = await chat.agenerate([[message], [message]])
+    assert isinstance(response, LLMResult)
+    assert len(response.generations) == 2
+    for generations in response.generations:
+        assert len(generations) == 2
+        for generation in generations:
+            assert isinstance(generation, ChatGeneration)
+            assert isinstance(generation.text, str)
+            assert generation.text == generation.message.content
+
+
+@pytest.mark.scheduled
+def test_openai_streaming(llm: AzureChatOpenAI) -> None:
+    """Test streaming tokens from OpenAI."""
+
+    for token in llm.stream("I'm Pickle Rick"):
+        assert isinstance(token.content, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_astream(llm: AzureChatOpenAI) -> None:
+    """Test streaming tokens from OpenAI."""
+    async for token in llm.astream("I'm Pickle Rick"):
+        assert isinstance(token.content, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_abatch(llm: AzureChatOpenAI) -> None:
+    """Test streaming tokens from AzureChatOpenAI."""
+
+    result = await llm.abatch(["I'm Pickle Rick", "I'm not Pickle Rick"])
+    for token in result:
+        assert isinstance(token.content, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_abatch_tags(llm: AzureChatOpenAI) -> None:
+    """Test batch tokens from AzureChatOpenAI."""
+
+    result = await llm.abatch(
+        ["I'm Pickle Rick", "I'm not Pickle Rick"], config={"tags": ["foo"]}
+    )
+    for token in result:
+        assert isinstance(token.content, str)
+
+
+@pytest.mark.scheduled
+def test_openai_batch(llm: AzureChatOpenAI) -> None:
+    """Test batch tokens from AzureChatOpenAI."""
+
+    result = llm.batch(["I'm Pickle Rick", "I'm not Pickle Rick"])
+    for token in result:
+        assert isinstance(token.content, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_ainvoke(llm: AzureChatOpenAI) -> None:
+    """Test invoke tokens from AzureChatOpenAI."""
+
+    result = await llm.ainvoke("I'm Pickle Rick", config={"tags": ["foo"]})
+    assert isinstance(result.content, str)
+
+
+@pytest.mark.scheduled
+def test_openai_invoke(llm: AzureChatOpenAI) -> None:
+    """Test invoke tokens from AzureChatOpenAI."""
+
+    result = llm.invoke("I'm Pickle Rick", config=dict(tags=["foo"]))
+    assert isinstance(result.content, str)

.scripts/community_split/libs/partners/openai/tests/integration_tests/llms/test_azure.py

@@ -0,0 +1,141 @@
+"""Test AzureOpenAI wrapper."""
+import os
+from typing import Any, Generator
+
+import pytest
+from langchain_core.outputs import LLMResult
+
+from langchain_openai.llms import AzureOpenAI
+
+OPENAI_API_VERSION = os.environ.get("AZURE_OPENAI_API_VERSION", "")
+OPENAI_API_BASE = os.environ.get("AZURE_OPENAI_API_BASE", "")
+OPENAI_API_KEY = os.environ.get("AZURE_OPENAI_API_KEY", "")
+DEPLOYMENT_NAME = os.environ.get("AZURE_OPENAI_DEPLOYMENT_NAME", "")
+
+
+def _get_llm(**kwargs: Any) -> AzureOpenAI:
+    return AzureOpenAI(
+        deployment_name=DEPLOYMENT_NAME,
+        openai_api_version=OPENAI_API_VERSION,
+        openai_api_base=OPENAI_API_BASE,
+        openai_api_key=OPENAI_API_KEY,
+        **kwargs,
+    )
+
+
+@pytest.mark.scheduled
+@pytest.fixture
+def llm() -> AzureOpenAI:
+    return _get_llm(
+        max_tokens=10,
+    )
+
+
+@pytest.mark.scheduled
+def test_openai_call(llm: AzureOpenAI) -> None:
+    """Test valid call to openai."""
+    output = llm("Say something nice:")
+    assert isinstance(output, str)
+
+
+@pytest.mark.scheduled
+def test_openai_streaming(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    generator = llm.stream("I'm Pickle Rick")
+
+    assert isinstance(generator, Generator)
+
+    full_response = ""
+    for token in generator:
+        assert isinstance(token, str)
+        full_response += token
+    assert full_response
+
+
+@pytest.mark.scheduled
+async def test_openai_astream(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    async for token in llm.astream("I'm Pickle Rick"):
+        assert isinstance(token, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_abatch(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    result = await llm.abatch(["I'm Pickle Rick", "I'm not Pickle Rick"])
+    for token in result:
+        assert isinstance(token, str)
+
+
+async def test_openai_abatch_tags(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    result = await llm.abatch(
+        ["I'm Pickle Rick", "I'm not Pickle Rick"], config={"tags": ["foo"]}
+    )
+    for token in result:
+        assert isinstance(token, str)
+
+
+@pytest.mark.scheduled
+def test_openai_batch(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    result = llm.batch(["I'm Pickle Rick", "I'm not Pickle Rick"])
+    for token in result:
+        assert isinstance(token, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_ainvoke(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    result = await llm.ainvoke("I'm Pickle Rick", config={"tags": ["foo"]})
+    assert isinstance(result, str)
+
+
+@pytest.mark.scheduled
+def test_openai_invoke(llm: AzureOpenAI) -> None:
+    """Test streaming tokens from AzureOpenAI."""
+    result = llm.invoke("I'm Pickle Rick", config=dict(tags=["foo"]))
+    assert isinstance(result, str)
+
+
+@pytest.mark.scheduled
+def test_openai_multiple_prompts(llm: AzureOpenAI) -> None:
+    """Test completion with multiple prompts."""
+    output = llm.generate(["I'm Pickle Rick", "I'm Pickle Rick"])
+    assert isinstance(output, LLMResult)
+    assert isinstance(output.generations, list)
+    assert len(output.generations) == 2
+
+
+def test_openai_streaming_best_of_error() -> None:
+    """Test validation for streaming fails if best_of is not 1."""
+    with pytest.raises(ValueError):
+        _get_llm(best_of=2, streaming=True)
+
+
+def test_openai_streaming_n_error() -> None:
+    """Test validation for streaming fails if n is not 1."""
+    with pytest.raises(ValueError):
+        _get_llm(n=2, streaming=True)
+
+
+def test_openai_streaming_multiple_prompts_error() -> None:
+    """Test validation for streaming fails if multiple prompts are given."""
+    with pytest.raises(ValueError):
+        _get_llm(streaming=True).generate(["I'm Pickle Rick", "I'm Pickle Rick"])
+
+
+@pytest.mark.scheduled
+def test_openai_streaming_call() -> None:
+    """Test valid call to openai."""
+    llm = _get_llm(max_tokens=10, streaming=True)
+    output = llm("Say foo:")
+    assert isinstance(output, str)
+
+
+@pytest.mark.scheduled
+async def test_openai_async_generate() -> None:
+    """Test async generation."""
+    llm = _get_llm(max_tokens=10)
+    output = await llm.agenerate(["Hello, how are you?"])
+    assert isinstance(output, LLMResult)