""" Tests for the MinimaxTranslationProvider. Validates Bug 1 fix: - default base_url is the public ``api.minimax.io`` host (NOT ``api.minimax.chat``) - default model is ``MiniMax-M3`` - ``is_available()`` / ``health_check()`` tolerate a missing ``/models`` path (Minimax does not document it) and only mark the provider down on 401/network error - translation success, 429 retry, 401 handling Batch mode (ported from the OpenAI provider): - same language pair + metadata chunk → ONE numbered-JSON request for all texts - unparseable/failed batch answer → per-item fallback, every text still translated - mixed metadata or a single text → no batch (individual requests) """ import json import pytest from unittest.mock import patch, MagicMock from requests.exceptions import Timeout from services.providers.minimax_provider import ( MinimaxTranslationProvider, MinimaxProviderError, MINIMAX_RATE_LIMITED, MINIMAX_INVALID_KEY, MINIMAX_TIMEOUT, MINIMAX_SERVICE_ERROR, ) from services.providers.schemas import TranslationRequest class TestMinimaxProviderConfig: """Defaults must point at the real public endpoint.""" def test_default_base_url_is_public_host(self): provider = MinimaxTranslationProvider(api_key="k", max_retries=0) assert provider._base_url == "https://api.minimax.io/v1" # Regression guard: the old broken host must never come back. assert "minimax.chat" not in provider._base_url def test_default_model_is_m3(self): provider = MinimaxTranslationProvider(api_key="k", max_retries=0) assert provider._model == "MiniMax-M3" def test_custom_base_url_respected(self): provider = MinimaxTranslationProvider( api_key="k", base_url="https://proxy.example.com/v1", max_retries=0 ) assert provider._base_url == "https://proxy.example.com/v1" def test_get_name(self): provider = MinimaxTranslationProvider(api_key="k", max_retries=0) assert provider.get_name() == "minimax" class TestMinimaxAvailabilityProbe: """is_available/health_check must not fail just because /models 404s.""" @pytest.fixture def provider(self): return MinimaxTranslationProvider(api_key="k", max_retries=0) def _mock_get(self, status_code: int): mock_response = MagicMock() mock_response.status_code = status_code return mock_response @patch("requests.get") def test_available_when_models_returns_200(self, mock_get, provider): mock_get.return_value = self._mock_get(200) assert provider.is_available() is True @patch("requests.get") def test_available_when_models_404(self, mock_get, provider): # Minimax does not document /models; a 404 must NOT mark it unavailable. mock_get.return_value = self._mock_get(404) assert provider.is_available() is True @patch("requests.get") def test_unavailable_on_401(self, mock_get, provider): mock_get.return_value = self._mock_get(401) assert provider.is_available() is False @patch("requests.get") def test_unavailable_on_network_error(self, _mock_get, provider): def _raise(*a, **kw): raise Timeout("boom") with patch("requests.get", side_effect=_raise): assert provider.is_available() is False @patch("requests.get") def test_health_check_tolerates_404(self, mock_get, provider): mock_get.return_value = self._mock_get(404) status = provider.health_check() assert status.available is True assert status.name == "minimax" @patch("requests.get") def test_health_check_marks_down_on_401(self, mock_get, provider): mock_get.return_value = self._mock_get(401) status = provider.health_check() assert status.available is False class TestMinimaxTranslateText: @pytest.fixture def provider(self): return MinimaxTranslationProvider(api_key="k", model="MiniMax-M3", max_retries=0) def _mock_post(self, payload, status_code=200): mock_response = MagicMock() mock_response.status_code = status_code mock_response.json.return_value = payload mock_response.text = "" return mock_response @patch("requests.post") def test_success(self, mock_post, provider): mock_post.return_value = self._mock_post( {"choices": [{"message": {"content": "Bonjour"}}], "usage": {}} ) resp = provider.translate_text(TranslationRequest(text="Hello", target_language="fr")) assert resp.translated_text == "Bonjour" assert resp.provider_name == "minimax" # Verify we hit the public host on the OpenAI-compatible path. called_url = mock_post.call_args[0][0] assert called_url == "https://api.minimax.io/v1/chat/completions" def test_empty_text_short_circuits(self, provider): resp = provider.translate_text(TranslationRequest(text="", target_language="fr")) assert resp.translated_text == "" @patch("requests.post") def test_invalid_key_returns_error(self, mock_post, provider): mock_post.return_value = self._mock_post({"error": "bad key"}, status_code=401) resp = provider.translate_text(TranslationRequest(text="Hello", target_language="fr")) assert resp.error_code == MINIMAX_INVALID_KEY # Original text returned on failure. assert resp.translated_text == "Hello" @patch("time.sleep") @patch("requests.post") def test_rate_limit_then_success(self, mock_post, mock_sleep): provider = MinimaxTranslationProvider(api_key="k", max_retries=2, retry_delay=0.01) mock_post.side_effect = [ self._mock_post({"error": "slow down"}, status_code=429), self._mock_post({"choices": [{"message": {"content": "Hola"}}], "usage": {}}), ] resp = provider.translate_text(TranslationRequest(text="Hello", target_language="es")) assert resp.translated_text == "Hola" assert mock_sleep.called # backoff happened @patch("requests.post") def test_service_error_when_empty_choices(self, mock_post, provider): mock_post.return_value = self._mock_post({"choices": []}) resp = provider.translate_text(TranslationRequest(text="Hello", target_language="fr")) assert resp.error_code == MINIMAX_SERVICE_ERROR class TestMinimaxProviderError: def test_error_carries_code_and_message(self): err = MinimaxProviderError(MINIMAX_TIMEOUT, "timed out", details={"wait": 1}) assert err.code == MINIMAX_TIMEOUT assert err.message == "timed out" assert err.details == {"wait": 1} class TestMinimaxBatchTranslation: """Batch mode: one numbered-JSON request per same pair/metadata chunk.""" @pytest.fixture def provider(self): return MinimaxTranslationProvider(api_key="k", max_retries=0) def _mock_post(self, payload, status_code=200): mock_response = MagicMock() mock_response.status_code = status_code mock_response.json.return_value = payload mock_response.text = "" return mock_response def _batch_response(self, translations): """Mock a 200 response whose content is a numbered JSON array.""" content = json.dumps( [{"id": i, "translation": t} for i, t in enumerate(translations)] ) return self._mock_post({"choices": [{"message": {"content": content}}], "usage": {}}) @patch("requests.post") def test_batch_of_three_uses_single_request(self, mock_post, provider): mock_post.return_value = self._batch_response(["Bonjour", "Au revoir", "Merci"]) requests = [ TranslationRequest(text="Hello", target_language="fr"), TranslationRequest(text="Goodbye", target_language="fr"), TranslationRequest(text="Thank you", target_language="fr"), ] responses = provider.translate_batch(requests) assert mock_post.call_count == 1 assert [r.translated_text for r in responses] == ["Bonjour", "Au revoir", "Merci"] assert all(r.provider_name == "minimax" for r in responses) assert all(r.error is None for r in responses) # The single call went to the public chat completions endpoint. assert mock_post.call_args[0][0] == "https://api.minimax.io/v1/chat/completions" @patch("requests.post") def test_batch_request_payload_is_numbered_json_with_instructions(self, mock_post, provider): mock_post.return_value = self._batch_response(["Bonjour", "Au revoir"]) requests = [ TranslationRequest( text="Hello", target_language="fr", source_language="en", metadata={"custom_prompt": "use a formal tone"}, ), TranslationRequest( text="Goodbye", target_language="fr", source_language="en", metadata={"custom_prompt": "use a formal tone"}, ), ] provider.translate_batch(requests) payload = mock_post.call_args[1]["json"] system = payload["messages"][0]["content"] user = payload["messages"][1]["content"] # System prompt = base instructions + custom prompt + strict JSON rule. assert "professional translator" in system assert "ADDITIONAL CONTEXT AND INSTRUCTIONS" in system assert "use a formal tone" in system assert "BATCH MODE" in system assert "ONLY a JSON array" in system # User message = the numbered JSON list of the texts. parsed = json.loads(user) assert [item["text"] for item in parsed] == ["Hello", "Goodbye"] assert [item["id"] for item in parsed] == [0, 1] @patch("requests.post") def test_invalid_batch_json_falls_back_to_individual(self, mock_post, provider): broken_batch = self._mock_post( {"choices": [{"message": {"content": "désolé, pas de JSON ici"}}], "usage": {}} ) individuals = [ self._mock_post({"choices": [{"message": {"content": "Bonjour"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Au revoir"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Merci"}}], "usage": {}}), ] mock_post.side_effect = [broken_batch] + individuals requests = [ TranslationRequest(text="Hello", target_language="fr"), TranslationRequest(text="Goodbye", target_language="fr"), TranslationRequest(text="Thank you", target_language="fr"), ] responses = provider.translate_batch(requests) # 1 failed batch attempt + 3 individual calls. assert mock_post.call_count == 4 assert [r.translated_text for r in responses] == ["Bonjour", "Au revoir", "Merci"] assert all(r.error is None for r in responses) @patch("requests.post") def test_batch_api_error_falls_back_to_individual(self, mock_post, provider): # The batch call itself fails (500) — per-item calls still translate everything. failed_batch = self._mock_post({"error": "boom"}, status_code=500) individuals = [ self._mock_post({"choices": [{"message": {"content": "Bonjour"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Au revoir"}}], "usage": {}}), ] mock_post.side_effect = [failed_batch] + individuals requests = [ TranslationRequest(text="Hello", target_language="fr"), TranslationRequest(text="Goodbye", target_language="fr"), ] responses = provider.translate_batch(requests) assert mock_post.call_count == 3 assert [r.translated_text for r in responses] == ["Bonjour", "Au revoir"] @patch("requests.post") def test_wrong_batch_length_falls_back_to_individual(self, mock_post, provider): # A JSON answer with a missing item is NOT trusted → fallback. short_batch = self._batch_response(["Bonjour", "Au revoir"]) individuals = [ self._mock_post({"choices": [{"message": {"content": "Bonjour"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Au revoir"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Merci"}}], "usage": {}}), ] mock_post.side_effect = [short_batch] + individuals requests = [ TranslationRequest(text="Hello", target_language="fr"), TranslationRequest(text="Goodbye", target_language="fr"), TranslationRequest(text="Thank you", target_language="fr"), ] responses = provider.translate_batch(requests) assert mock_post.call_count == 4 assert [r.translated_text for r in responses] == ["Bonjour", "Au revoir", "Merci"] @patch("requests.post") def test_mixed_metadata_skips_batch(self, mock_post, provider): mock_post.side_effect = [ self._mock_post({"choices": [{"message": {"content": "Bonjour"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Au revoir"}}], "usage": {}}), ] requests = [ TranslationRequest( text="Hello", target_language="fr", metadata={"custom_prompt": "formal"} ), TranslationRequest( text="Goodbye", target_language="fr", metadata={"custom_prompt": "casual"} ), ] responses = provider.translate_batch(requests) # Two individual calls, no numbered-JSON batch attempt. assert mock_post.call_count == 2 assert [r.translated_text for r in responses] == ["Bonjour", "Au revoir"] for call in mock_post.call_args_list: assert call[1]["json"]["messages"][1]["content"] in ("Hello", "Goodbye") @patch("requests.post") def test_mixed_language_pair_skips_batch(self, mock_post, provider): mock_post.side_effect = [ self._mock_post({"choices": [{"message": {"content": "Bonjour"}}], "usage": {}}), self._mock_post({"choices": [{"message": {"content": "Hola"}}], "usage": {}}), ] requests = [ TranslationRequest(text="Hello", target_language="fr"), TranslationRequest(text="Hello", target_language="es"), ] responses = provider.translate_batch(requests) assert mock_post.call_count == 2 assert [r.translated_text for r in responses] == ["Bonjour", "Hola"] @patch("requests.post") def test_single_request_is_individual(self, mock_post, provider): mock_post.return_value = self._mock_post( {"choices": [{"message": {"content": "Bonjour"}}], "usage": {}} ) requests = [TranslationRequest(text="Hello", target_language="fr")] responses = provider.translate_batch(requests) assert mock_post.call_count == 1 assert responses[0].translated_text == "Bonjour" # The user message is the raw text, not a numbered JSON list. user_message = mock_post.call_args[1]["json"]["messages"][1]["content"] assert user_message == "Hello" def test_empty_batch_returns_empty(self, provider): assert provider.translate_batch([]) == []