import importlib import os import sys import unittest from unittest.mock import MagicMock, patch CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) if CURRENT_DIR not in sys.path: sys.path.insert(0, CURRENT_DIR) api_app = importlib.import_module("app") TEST_SCHEMA = { "pessoas": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "nome": {"name": "nome", "raw_type": "varchar", "kind": "text"}, "codigo": {"name": "codigo", "raw_type": "varchar", "kind": "text"}, "federacao": {"name": "federacao", "raw_type": "int", "kind": "number"}, "status": {"name": "status", "raw_type": "int", "kind": "number"}, "email": {"name": "email", "raw_type": "varchar", "kind": "text"}, "data_cadastro": { "name": "data_cadastro", "raw_type": "datetime", "kind": "date", }, }, "competicoes": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "nome": {"name": "nome", "raw_type": "varchar", "kind": "text"}, "codigo": {"name": "codigo", "raw_type": "varchar", "kind": "text"}, "inicio": {"name": "inicio", "raw_type": "date", "kind": "date"}, "status": {"name": "status", "raw_type": "int", "kind": "number"}, }, "ranking": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "atleta": {"name": "atleta", "raw_type": "int", "kind": "number"}, "federacao": {"name": "federacao", "raw_type": "int", "kind": "number"}, "clube": {"name": "clube", "raw_type": "int", "kind": "number"}, "categoria": {"name": "categoria", "raw_type": "int", "kind": "number"}, "classe": {"name": "classe", "raw_type": "int", "kind": "number"}, "ano": {"name": "ano", "raw_type": "year", "kind": "number"}, "sexo": {"name": "sexo", "raw_type": "int", "kind": "number"}, }, "eventos": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "nome": {"name": "nome", "raw_type": "varchar", "kind": "text"}, "inicio": {"name": "inicio", "raw_type": "date", "kind": "date"}, "status": {"name": "status", "raw_type": "int", "kind": "number"}, }, "categorias": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "nome": {"name": "nome", "raw_type": "varchar", "kind": "text"}, "ordem": {"name": "ordem", "raw_type": "int", "kind": "number"}, }, "classes": { "id": {"name": "id", "raw_type": "int", "kind": "number"}, "nome": {"name": "nome", "raw_type": "varchar", "kind": "text"}, "ordem": {"name": "ordem", "raw_type": "int", "kind": "number"}, }, } class FlaskAppTestCase(unittest.TestCase): def setUp(self): api_app.app.config["TESTING"] = True self.client = api_app.app.test_client() api_app.API_TOKEN = "token-api-valido" self._load_registry() self.auth_headers = {"Authorization": "Bearer token-api-valido"} def _load_registry(self): api_app.SCHEMA = TEST_SCHEMA ( api_app.RESOURCE_REGISTRY, api_app.CATALOG_REGISTRY, ) = api_app.build_resource_registry(TEST_SCHEMA) 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)) @patch("app.pymysql.connect") @patch("app.pd.Timestamp.now") def test_health_check_returns_online_status(self, mock_now, mock_connect): mock_conn = MagicMock() mock_connect.return_value = mock_conn mock_now.return_value.strftime.return_value = "2026-03-18 10:30:00" response = self.client.get("/health") self.assertEqual(response.status_code, 200) self.assertEqual( response.get_json(), { "status": "online", "database": "connected", "version": "3.1.0", "timestamp": "2026-03-18 10:30:00", }, ) mock_connect.assert_called_once() mock_conn.close.assert_called_once() def test_protected_endpoint_requires_bearer_token(self): response = self.client.get("/api/meta/overview") self.assertEqual(response.status_code, 401) self.assertEqual( response.get_json(), {"error": "Acesso negado. Token de autenticação ausente."}, ) def test_protected_endpoint_rejects_malformed_bearer_token(self): response = self.client.get( "/api/meta/overview", headers={"Authorization": "Bearer"}, ) self.assertEqual(response.status_code, 401) self.assertEqual( response.get_json(), {"error": "Token Bearer mal formatado."}, ) def test_protected_endpoint_rejects_invalid_token(self): response = self.client.get( "/api/meta/overview", headers={"Authorization": "Bearer token-invalido"}, ) self.assertEqual(response.status_code, 401) self.assertEqual( response.get_json(), {"error": "Acesso negado. Token inválido ou expirado."}, ) def test_meta_overview_returns_grouped_metadata(self): response = self.client.get("/api/meta/overview", headers=self.auth_headers) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertIn("groups", payload) self.assertIn("cadastros", payload["groups"]) self.assertIn("competicoes", payload["groups"]) self.assertIn("catalog_count", payload) self.assertIn("query_operators", payload) def test_meta_groups_returns_available_groups(self): response = self.client.get("/api/meta/groups", headers=self.auth_headers) self.assertEqual(response.status_code, 200) payload = response.get_json() groups = {item["group"] for item in payload} self.assertIn("cadastros", groups) self.assertIn("competicoes", groups) self.assertIn("ranking", groups) def test_meta_resources_can_filter_by_group(self): response = self.client.get( "/api/meta/resources?group=cadastros", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertTrue(payload) self.assertTrue(all(item["group"] == "cadastros" for item in payload)) def test_meta_resource_detail_returns_schema_information(self): response = self.client.get( "/api/meta/resources/pessoas", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["resource"], "pessoas") self.assertEqual(payload["group"], "cadastros") self.assertIn("columns", payload) self.assertIn("searchable_fields", payload) @patch("app.load_schema_from_database", return_value=TEST_SCHEMA) def test_registry_can_be_loaded_lazily_from_database(self, mock_load_schema): api_app.RESOURCE_REGISTRY = {} api_app.CATALOG_REGISTRY = {} api_app.SCHEMA = {} response = self.client.get("/api/meta/overview", headers=self.auth_headers) self.assertEqual(response.status_code, 200) self.assertTrue(mock_load_schema.called) self.assertIn("pessoas", api_app.RESOURCE_REGISTRY) @patch("app.execute_query") def test_resource_listing_uses_generic_query_layer(self, mock_execute_query): mock_execute_query.return_value = [{"id": 1, "nome": "João"}] response = self.client.get( "/api/resources/pessoas?limit=10&nome__like=Jo", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["resource"], "pessoas") self.assertEqual(payload["data"], [{"id": 1, "nome": "João"}]) self.assertEqual(payload["pagination"]["returned"], 1) mock_execute_query.assert_called_once() def test_resource_listing_rejects_invalid_fields(self): response = self.client.get( "/api/resources/pessoas?fields=id,nome,inexistente", headers=self.auth_headers, ) self.assertEqual(response.status_code, 400) self.assertIn("Campos inválidos", response.get_json()["error"]) def test_resource_listing_rejects_invalid_sort_field(self): response = self.client.get( "/api/resources/pessoas?sort_by=inexistente", headers=self.auth_headers, ) self.assertEqual(response.status_code, 400) self.assertIn("Campo de ordenação inválido", response.get_json()["error"]) def test_resource_listing_rejects_invalid_filter_field(self): response = self.client.get( "/api/resources/pessoas?naoexiste=123", headers=self.auth_headers, ) self.assertEqual(response.status_code, 400) self.assertIn("Filtros inválidos", response.get_json()["error"]) @patch("app.execute_query") def test_resource_detail_returns_404_when_not_found(self, mock_execute_query): mock_execute_query.return_value = None response = self.client.get( "/api/resources/pessoas/999", headers=self.auth_headers, ) self.assertEqual(response.status_code, 404) self.assertEqual(response.get_json(), {"error": "Registro não encontrado."}) @patch("app.execute_query") def test_resource_summary_returns_quantitative_blocks(self, mock_execute_query): mock_execute_query.side_effect = [ {"total": 12}, [{"valor": 1, "total": 7}], [{"valor": 2, "total": 5}], [{"valor": 3, "total": 4}], [{"valor": 4, "total": 3}], [{"valor": 2026, "total": 12}], [{"valor": 5, "total": 2}], [{"valor": 6, "total": 1}], ] response = self.client.get( "/api/resources/ranking/stats/summary", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["resource"], "ranking") self.assertEqual(payload["total_registros"], 12) self.assertIn("groupings", payload) @patch("app.execute_query") def test_group_by_returns_aggregated_counts(self, mock_execute_query): mock_execute_query.return_value = [{"valor": "SP", "total": 20}] response = self.client.get( "/api/resources/pessoas/stats/group-by/federacao?top=5", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) self.assertEqual( response.get_json(), { "resource": "pessoas", "field": "federacao", "data": [{"valor": "SP", "total": 20}], }, ) def test_group_by_rejects_invalid_field(self): response = self.client.get( "/api/resources/pessoas/stats/group-by/inexistente", headers=self.auth_headers, ) self.assertEqual(response.status_code, 400) self.assertIn("não existe no recurso", response.get_json()["error"]) @patch("app.execute_query") def test_time_series_returns_grouped_periods(self, mock_execute_query): mock_execute_query.return_value = [{"periodo": "2026-03", "total": 8}] response = self.client.get( "/api/resources/competicoes/stats/time-series?date_field=inicio&granularity=month", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["resource"], "competicoes") self.assertEqual(payload["date_field"], "inicio") self.assertEqual(payload["granularity"], "month") def test_time_series_rejects_invalid_granularity(self): response = self.client.get( "/api/resources/competicoes/stats/time-series?date_field=inicio&granularity=week", headers=self.auth_headers, ) self.assertEqual(response.status_code, 400) self.assertIn("granularity deve ser", response.get_json()["error"]) def test_group_resources_returns_only_resources_of_group(self): response = self.client.get( "/api/groups/cadastros/resources", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertTrue(payload) self.assertTrue(all(item["group"] == "cadastros" for item in payload)) @patch("app.execute_query") def test_group_resource_list_reuses_resource_listing(self, mock_execute_query): mock_execute_query.return_value = [{"id": 1, "nome": "João"}] response = self.client.get( "/api/groups/cadastros/pessoas?limit=5", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["group"], "cadastros") self.assertEqual(payload["resource"], "pessoas") def test_group_resource_route_rejects_wrong_group(self): response = self.client.get( "/api/groups/ranking/pessoas", headers=self.auth_headers, ) self.assertEqual(response.status_code, 404) self.assertIn("não pertence ao grupo", response.get_json()["error"]) def test_catalogs_endpoint_lists_available_catalogs(self): response = self.client.get("/api/catalogs", headers=self.auth_headers) self.assertEqual(response.status_code, 200) payload = response.get_json() catalog_names = {item["catalog"] for item in payload} self.assertIn("categorias", catalog_names) self.assertIn("classes", catalog_names) @patch("app.execute_query") def test_catalog_detail_returns_catalog_rows(self, mock_execute_query): mock_execute_query.return_value = [{"id": 1, "nome": "Senior"}] response = self.client.get( "/api/catalogs/classes", headers=self.auth_headers, ) self.assertEqual(response.status_code, 200) payload = response.get_json() self.assertEqual(payload["catalog"], "classes") self.assertEqual(payload["data"], [{"id": 1, "nome": "Senior"}]) if __name__ == "__main__": unittest.main()