# -*- coding: utf-8 -*- # test.py import json import unittest from unittest.mock import MagicMock, patch import os # Importa as aplicações locais import app as mcp_app import mcp_server class MockResponse: """Simula a resposta da biblioteca 'requests' para não bater na API real durante o teste.""" def __init__(self, status_code=200, json_data=None, text=""): self.status_code = status_code self._json_data = json_data if json_data is not None else {} self.text = text def json(self): return self._json_data def _return_asyncio_result(result): """Auxiliar para mockar funções assíncronas do LangChain/MCP.""" def _runner(coro): coro.close() return result return _runner class FlaskAppTestCase(unittest.TestCase): def setUp(self): mcp_app.app.config["TESTING"] = True self.client = mcp_app.app.test_client() # Simula o token JWT/Bearer enviado pelo Chatbot self.valid_token = "meu-token-seguro-123" self.auth_header = {"Authorization": f"Bearer {self.valid_token}"} # Se você estiver usando o token_required com API_SECRET_KEY: mcp_app.settings.API_SECRET_KEY = self.valid_token def test_health_check_returns_online_status(self): response = self.client.get("/health") self.assertEqual(response.status_code, 200) self.assertEqual( response.get_json(), {"status": "online", "service": "Gateway IA Zempo (CBJ)"}, ) def test_apidocs_requires_basic_auth(self): response = self.client.get("/apidocs/") self.assertEqual(response.status_code, 401) self.assertIn("Acesso restrito", response.get_data(as_text=True)) def test_chat_requires_bearer_token(self): response = self.client.post("/chat", json={"pergunta": "Quais os atletas?"}) self.assertEqual(response.status_code, 401) # O gateway deve bloquear acessos não autenticados self.assertIn("Acesso negado", response.get_json()["error"]) def test_chat_rejects_malformed_bearer_token(self): response = self.client.post( "/chat", json={"pergunta": "teste"}, headers={"Authorization": "Bearer"}, ) self.assertEqual(response.status_code, 401) self.assertIn("Token mal formatado", response.get_json().get("error", "")) @patch("app.asyncio.run", side_effect=_return_asyncio_result("Resposta gerada pela IA Zempo")) def test_chat_processes_valid_request(self, mock_asyncio_run): response = self.client.post( "/chat", json={"pergunta": "Quais os clubes da federação SP?"}, headers=self.auth_header, ) self.assertEqual(response.status_code, 200) self.assertEqual( response.get_json(), {"status": "sucesso", "resposta": "Resposta gerada pela IA Zempo"}, ) mock_asyncio_run.assert_called_once() def test_chat_rejects_invalid_json_payload_with_400(self): response = self.client.post( "/chat", data='{"pergunta": ', # JSON propositalmente quebrado content_type="application/json", headers=self.auth_header, ) self.assertEqual(response.status_code, 400) self.assertEqual( response.get_json(), {"status": "erro", "mensagem": "JSON inválido na requisição."}, ) class MCPServerTestCase(unittest.TestCase): @patch("mcp_server.requests.get") def test_zempo_get_uses_bearer_query_params_and_ignores_nulls(self, mock_get): mock_get.return_value = MockResponse(status_code=200, json_data={"resultado": "ok"}) with patch.object(mcp_server.settings, "API_BASE_URL", "https://api.exemplo.com"): with patch.object(mcp_server.settings, "API_TOKEN", "token-padrao-cbj"): response = mcp_server.zempo_get( "/api/resources/competicoes_categorias", params={"codigo": "jj23", "id_categoria": None} ) self.assertEqual(json.loads(response), {"resultado": "ok"}) mock_get.assert_called_once() args, kwargs = mock_get.call_args self.assertEqual(args[0], "https://api.exemplo.com/api/resources/competicoes_categorias") self.assertEqual(kwargs["headers"]["Authorization"], "Bearer token-padrao-cbj") self.assertEqual(kwargs["headers"]["Accept"], "application/json") self.assertEqual(kwargs["params"]["codigo"], "jj23") self.assertNotIn("id_categoria", kwargs["params"]) self.assertEqual(kwargs["timeout"], mcp_server.REQUEST_TIMEOUT) @patch("mcp_server.requests.get") def test_zempo_get_preserves_explicit_id_param(self, mock_get): """Testa se o parâmetro id é enviado diretamente para a API nova.""" mock_get.return_value = MockResponse(status_code=200, json_data={"ok": True}) with patch.object(mcp_server.settings, "API_BASE_URL", "https://api.exemplo.com"): with patch.object(mcp_server.settings, "API_TOKEN", "token-padrao-cbj"): mcp_server.zempo_get( "/api/resources/pessoas", params={"id": "999", "nome": "Judoca"}, ) args, kwargs = mock_get.call_args self.assertIn("id", kwargs["params"]) self.assertEqual(kwargs["params"]["id"], "999") self.assertEqual(kwargs["params"]["nome"], "Judoca") @patch("mcp_server.requests.get") def test_zempo_get_handles_http_errors_gracefully(self, mock_get): mock_get.return_value = MockResponse(status_code=500, text="Erro Interno CBJ") with patch.object(mcp_server.settings, "API_BASE_URL", "https://api.exemplo.com"): with patch.object(mcp_server.settings, "API_TOKEN", "token-padrao-cbj"): response = mcp_server.zempo_get("/api/resources/pessoas") parsed = json.loads(response) self.assertEqual(parsed["error"], 500) self.assertEqual(parsed["message"], "Erro Interno CBJ") def test_zempo_get_returns_invalid_params_for_bad_filtros_json(self): with patch.object(mcp_server.settings, "API_TOKEN", "token-padrao-cbj"): response = mcp_server.zempo_get( "/api/resources/pessoas", filtros_json="{invalido", ) parsed = json.loads(response) self.assertEqual(parsed["error"], "invalid_params") self.assertIn("filtros_json", parsed["message"]) @patch("mcp_server._resource_list", return_value='{"mock": "data"}') def test_zempo_pessoas_tool_delegates_correctly(self, mock_resource_list): """Garante que a tool usa o recurso novo e passa filtros compatíveis com o schema.""" res = mcp_server.zempo_pessoas(nome="Alexia", tipo_atleta="1") self.assertEqual(res, '{"mock": "data"}') mock_resource_list.assert_called_once() args, kwargs = mock_resource_list.call_args self.assertEqual(args[0], "pessoas") self.assertEqual(kwargs["params"]["nome"], "Alexia") self.assertEqual(kwargs["params"]["tipo_atleta"], "1") @patch("mcp_server._resource_list", return_value='{"mock": "data"}') def test_zempo_pessoas_tool_forwards_filtros_json_explicitly(self, mock_resource_list): res = mcp_server.zempo_pessoas( nome="Alexia", filtros_json='{"data_cadastro__gte":"2025-01-01"}', ) self.assertEqual(res, '{"mock": "data"}') mock_resource_list.assert_called_once() _, kwargs = mock_resource_list.call_args self.assertEqual(kwargs["filtros_json"], '{"data_cadastro__gte":"2025-01-01"}') @patch("mcp_server.zempo_get", return_value='{"mock": "meta"}') def test_zempo_api_recursos_tool_uses_new_meta_endpoint(self, mock_zempo_get): res = mcp_server.zempo_api_recursos(group="competicoes") self.assertEqual(res, '{"mock": "meta"}') mock_zempo_get.assert_called_once_with( "/api/meta/resources", params={"group": "competicoes"}, ) class IAExecutionTestCase(unittest.TestCase): def test_executar_ia_simples_returns_friendly_error_on_internal_failure(self): async_context = MagicMock() # Força um erro crítico na comunicação do LangChain com o MCP async_context.__aenter__.side_effect = RuntimeError("Falha na ponte MCP") with patch("app.stdio_client", return_value=async_context): resposta = mcp_app.asyncio.run( mcp_app._executar_ia_simples("Quantos atletas?", "token-zempo-do-usuario") ) self.assertEqual( resposta, "Erro interno na IA: Falha na ponte MCP", ) if __name__ == "__main__": unittest.main()