test(together_ai): move request-shape checks to the mapped file, drop the live ones

tests/_live_test_helpers.py

@@ -1,10 +1,64 @@
 import os
+from datetime import date
 
 import pytest
+from pydantic import BaseModel, ConfigDict
 
 
 def _skip_live_prompt_caching_test():
     if os.environ.get("LITELLM_RUN_LIVE_PROMPT_CACHING_TESTS") != "1":
         pytest.skip("Live prompt-caching E2E tests are opt-in")
     if os.environ.get("CASSETTE_REDIS_URL"):
         pytest.skip("Live prompt-caching E2E tests cannot run under VCR replay")
+
+
+
+class TogetherCostEntry(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="ignore")
+
+    litellm_provider: str | None = None
+    mode: str | None = None
+    deprecation_date: str | None = None
+    input_cost_per_token: float | None = None
+    output_cost_per_token: float | None = None
+    supports_function_calling: bool | None = None
+    supports_response_schema: bool | None = None
+
+
+def cheapest_together_chat_model(
+    *, function_calling: bool = False, response_schema: bool = False
+) -> str:
+    import litellm
+
+    today = date.today().isoformat()
+
+    def qualifies(name: str, entry: TogetherCostEntry) -> bool:
+        return (
+            name.startswith("together_ai/")
+            and entry.litellm_provider == "together_ai"
+            and entry.mode == "chat"
+            and (entry.deprecation_date is None or entry.deprecation_date > today)
+            and (entry.input_cost_per_token or 0.0) > 0
+            and (entry.output_cost_per_token or 0.0) > 0
+            and (not function_calling or bool(entry.supports_function_calling))
+            and (not response_schema or bool(entry.supports_response_schema))
+        )
+
+    registry: dict[str, TogetherCostEntry] = {
+        name: TogetherCostEntry.model_validate(raw)
+        for name, raw in litellm.model_cost.items()
+        if isinstance(raw, dict) and name.startswith("together_ai/")
+    }
+    candidates = sorted(
+        (name for name, entry in registry.items() if qualifies(name, entry)),
+        key=lambda name: (
+            registry[name].input_cost_per_token or 0.0,
+            registry[name].output_cost_per_token or 0.0,
+            name,
+        ),
+    )
+    assert candidates, (
+        "no live together_ai chat model in the cost map satisfies "
+        f"function_calling={function_calling} response_schema={response_schema}"
+    )
+    return candidates[0]

tests/llm_translation/test_together_ai.py

@@ -3,6 +3,7 @@
 """
 
 from base_llm_unit_tests import BaseLLMChatTest
+from tests._live_test_helpers import cheapest_together_chat_model
 import json
 import os
 from datetime import datetime
@@ -16,7 +17,11 @@
 class TestTogetherAI(BaseLLMChatTest):
     def get_base_completion_call_args(self) -> dict:
         litellm.set_verbose = True
-        return {"model": "together_ai/openai/gpt-oss-20b"}
+        return {
+            "model": cheapest_together_chat_model(
+                function_calling=True, response_schema=True
+            )
+        }
 
     def test_tool_call_no_arguments(self, tool_call_no_arguments):
         """Test that tool calls with no arguments is translated correctly. Relevant issue: https://github.com/BerriAI/litellm/issues/6833"""

tests/local_testing/test_completion.py

@@ -57,23 +57,6 @@ def test_response_model_none():
     assert isinstance(x, litellm.ModelResponse)
 
 
-def test_completion_custom_provider_model_name():
-    try:
-        litellm.cache = None
-        response = completion(
-            model="together_ai/openai/gpt-oss-20b",
-            messages=messages,
-            logger_fn=logger_fn,
-        )
-        # Add assertions here to check the-response
-        print(response)
-        print(response["choices"][0]["finish_reason"])
-    except litellm.Timeout as e:
-        pass
-    except Exception as e:
-        pytest.fail(f"Error occurred: {e}")
-
-
 def _openai_mock_response(*args, **kwargs) -> litellm.ModelResponse:
     new_response = MagicMock()
     new_response.headers = {"hello": "world"}
@@ -2803,41 +2786,6 @@ def test_completion_together_ai_llama():
 
 
 # test_completion_together_ai()
-def test_customprompt_together_ai():
-    try:
-        litellm.set_verbose = False
-        litellm.num_retries = 0
-        print("in test_customprompt_together_ai")
-        print(litellm.success_callback)
-        print(litellm._async_success_callback)
-        response = completion(
-            model="together_ai/openai/gpt-oss-20b",
-            messages=messages,
-            roles={
-                "system": {
-                    "pre_message": "<|im_start|>system\n",
-                    "post_message": "<|im_end|>",
-                },
-                "assistant": {
-                    "pre_message": "<|im_start|>assistant\n",
-                    "post_message": "<|im_end|>",
-                },
-                "user": {
-                    "pre_message": "<|im_start|>user\n",
-                    "post_message": "<|im_end|>",
-                },
-            },
-        )
-        print(response)
-    except litellm.exceptions.Timeout as e:
-        print(f"Timeout Error")
-        pass
-    except Exception as e:
-        print(f"ERROR TYPE {type(e)}")
-        pytest.fail(f"Error occurred: {e}")
-
-
-# test_customprompt_together_ai()
 
 
 def response_format_tests(response: litellm.ModelResponse):
@@ -3644,28 +3592,6 @@ async def test_acompletion_stream_watsonx():
 # test_maritalk()
 
 
-def test_completion_together_ai_stream():
-    litellm.set_verbose = True
-    user_message = "Write 1pg about YC & litellm"
-    messages = [{"content": user_message, "role": "user"}]
-    try:
-        response = completion(
-            model="together_ai/openai/gpt-oss-20b",
-            messages=messages,
-            stream=True,
-            max_tokens=5,
-        )
-        print(response)
-        for chunk in response:
-            print(chunk)
-        # print(string_response)
-    except Exception as e:
-        pytest.fail(f"Error occurred: {e}")
-
-
-# test_completion_together_ai_stream()
-
-
 def test_moderation():
     response = litellm.moderation(input="i'm ishaan cto of litellm")
     print(response)

tests/local_testing/test_text_completion.py

@@ -12,6 +12,7 @@
 import pytest
 
 import litellm
+from tests._live_test_helpers import cheapest_together_chat_model
 from litellm import (
     RateLimitError,
     TextCompletionResponse,
@@ -4030,7 +4031,7 @@ def test_async_text_completion_together_ai():
     async def test_get_response():
         try:
             response = await litellm.atext_completion(
-                model="together_ai/openai/gpt-oss-20b",
+                model=cheapest_together_chat_model(),
                 prompt="good morning",
                 max_tokens=10,
             )

tests/test_litellm/llms/together_ai/chat/test_together_ai_chat_transformation.py

@@ -1108,3 +1108,52 @@ def test_get_optional_params_preserves_max_for_declared_levels_model():
     )
 
     assert optional_params["reasoning_effort"] == "max"
+
+
+def _together_chat_transport() -> tuple[HTTPHandler, list[httpx.Request]]:
+    captured_requests: list[httpx.Request] = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        captured_requests.append(request)
+        return httpx.Response(
+            200,
+            json={
+                "id": "chatcmpl-together",
+                "object": "chat.completion",
+                "created": 1234567890,
+                "model": TOOL_CALLING_MODEL,
+                "choices": [
+                    {
+                        "index": 0,
+                        "message": {"role": "assistant", "content": "Hello!"},
+                        "finish_reason": "stop",
+                    }
+                ],
+                "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
+            },
+        )
+
+    client = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
+    return client, captured_requests
+
+
+def test_custom_role_wrappers_never_reach_the_request():
+    client, captured_requests = _together_chat_transport()
+    messages = [{"role": "user", "content": "Hello!"}]
+
+    litellm.completion(
+        model=f"together_ai/{TOOL_CALLING_MODEL}",
+        messages=messages,
+        roles={
+            "system": {"pre_message": "<|im_start|>system\n", "post_message": "<|im_end|>"},
+            "assistant": {"pre_message": "<|im_start|>assistant\n", "post_message": "<|im_end|>"},
+            "user": {"pre_message": "<|im_start|>user\n", "post_message": "<|im_end|>"},
+        },
+        api_key="fake-key",
+        client=client,
+    )
+
+    request_body = json.loads(captured_requests[0].content)
+    assert request_body["messages"] == messages
+    assert "prompt" not in request_body
+    assert "roles" not in request_body