EidanThen/TransferData-main
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
Eidan T.
2026-05-09 14:00:04 -04:00
commit e0ec11977c
42 changed files with 4596 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/IA.py
+115
View File
@@ -0,0 +1,115 @@
import requests
import json
from datetime import datetime
# ---------------------------
# CONFIG
# ---------------------------
OLLAMA_URL = "http://136.115.239.38:11434/api/generate"
MODEL = "qwen:14b"
pg_config = PGConfig()
database = DatabaseManager(pg_config.database, pg_config.databaseID)
# ---------------------------
# 1. TU QUERY (ya la tienes)
# ---------------------------
query = f"""
SELECT TOP (100)-- [ID]
-- ,[FieldID]
[ExecutionID]
,[ProcedureID]
,[ProcedureName]
,[ProcedureType]
,[TargetDate]
,[Status]
,[ExecutedBy]
,[MessageDescription]
,[Source]
,[ExtraInfo]
,[ExecutionDate]
FROM [glm_pg].[dbo].[ExecutionLog]
where [Status]='FAILED' AND [TargetDate]>'2026-04-01'
ORDER BY [ExecutionDate] DESC
"""
print("Query ejecutada:")
print(query)
# ---------------------------
# 2. EJECUTAR QUERY
# ---------------------------
data_sql_report = database.get_execute_query(query)
# ---------------------------
# 3. FORMATEAR DATOS
# ---------------------------
def format_data(data):
try:
return json.dumps(data, indent=2, default=str)
except:
return str(data)
data_formatted = format_data(data_sql_report)
# ---------------------------
# 4. CREAR PROMPT
# ---------------------------
prompt = f"""
Eres un analista de datos experto.
Tengo los siguientes datos obtenidos desde SQL:
QUERY:
{query}
RESULTADO:
{data_formatted}
TAREA:
1. Analiza los datos
2. Identifica patrones importantes
3. Detecta posibles errores o anomalías
4. Da recomendaciones claras
Responde de forma clara y estructurada.
"""
# ---------------------------
# 5. GUARDAR PROMPT EN TXT
# ---------------------------
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
file_name = f"prompt_{timestamp}.txt"
with open(file_name, "w", encoding="utf-8") as f:
f.write(prompt)
print(f"\nPrompt guardado en: {file_name}")
# ---------------------------
# 6. (OPCIONAL) ENVIAR A IA
# ---------------------------
send_to_ai = True # cambia a False si solo quieres el TXT
print(datetime.now())
if send_to_ai:
response = requests.post(
OLLAMA_URL,
json={
"model": MODEL,
"prompt": prompt,
"stream": False
}
)
result = response.json()
print("\nRespuesta IA:\n")
print(result["response"])
print(datetime.now())
print(datetime.now())
tests/test_db_utils.py
+332
View File
@@ -0,0 +1,332 @@
# Tests para App/Utils/db_utils.py
import pytest
from unittest.mock import Mock, MagicMock, patch, call
from App.Utils.db_utils import DatabaseManager, get_db_connection
class TestDatabaseManager:
"""Tests para la clase DatabaseManager"""
@patch('App.Utils.db_utils.ConfigDatabase')
def test_init(self, mock_config):
"""Test que DatabaseManager se inicializa correctamente con credenciales"""
# Arrange
mock_config.db_user = 'test_user'
mock_config.db_password = 'test_pass'
mock_config.db_host = 'test_host'
# Act
manager = DatabaseManager('test_db', 'test_id')
# Assert
assert manager.db_database == 'test_db'
assert manager.db_databaseID == 'test_id'
assert manager.db_user == 'test_user'
assert manager.db_password == 'test_pass'
assert manager.db_host == 'test_host'
@patch('App.Utils.db_utils.pyodbc.connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_connect_success(self, mock_config, mock_pyodbc_connect):
"""Test que _connect establece conexión exitosa"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
mock_connection = MagicMock()
mock_pyodbc_connect.return_value = mock_connection
manager = DatabaseManager('test_db', 'test_id')
# Act
manager._connect()
# Assert
assert manager.connection == mock_connection
mock_pyodbc_connect.assert_called_once()
assert 'ODBC Driver 17 for SQL Server' in manager.conn_str
@patch('App.Utils.db_utils.pyodbc.connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_connect_failure(self, mock_config, mock_pyodbc_connect):
"""Test que _connect maneja errores de conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
mock_pyodbc_connect.side_effect = Exception("Connection failed")
manager = DatabaseManager('test_db', 'test_id')
# Act
manager._connect()
# Assert
assert manager.connection is None
@patch('App.Utils.db_utils.ConfigDatabase')
def test_disconnect(self, mock_config):
"""Test que _disconnect cierra la conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
# Act
manager._disconnect()
# Assert
manager.connection.close.assert_called_once()
@patch('App.Utils.db_utils.ConfigDatabase')
def test_disconnect_no_connection(self, mock_config):
"""Test que _disconnect no falla si no hay conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = None
# Act & Assert (no debe lanzar excepción)
manager._disconnect()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_get_execute_query_success(self, mock_config, mock_connect, mock_disconnect):
"""Test que get_execute_query devuelve datos como lista de dicts"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
manager.cursor = MagicMock()
# Simula resultado de query
manager.cursor.description = [('id',), ('name',)]
manager.cursor.fetchall.return_value = [(1, 'Alice'), (2, 'Bob')]
manager.connection.cursor.return_value = manager.cursor
# Act
results, error = manager.get_execute_query("SELECT * FROM users")
# Assert
assert error is None
assert len(results) == 2
assert results[0] == {'id': 1, 'name': 'Alice'}
assert results[1] == {'id': 2, 'name': 'Bob'}
manager.cursor.execute.assert_called_once_with("SELECT * FROM users")
mock_disconnect.assert_called_once()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_get_execute_query_no_connection(self, mock_config, mock_connect, mock_disconnect):
"""Test que get_execute_query devuelve error si no hay conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = None
# Act
results, error = manager.get_execute_query("SELECT * FROM users")
# Assert
assert results is None
assert error == "Error de conexión"
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_get_execute_query_error(self, mock_config, mock_connect, mock_disconnect):
"""Test que get_execute_query maneja errores de query"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
manager.cursor = MagicMock()
manager.cursor.execute.side_effect = Exception("invalid column")
manager.connection.cursor.return_value = manager.cursor
# Act
results, error = manager.get_execute_query("SELECT invalid_col FROM users")
# Assert
assert results is None
assert "invalid column" in error
mock_disconnect.assert_called_once()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_set_execute_query_success(self, mock_config, mock_connect, mock_disconnect):
"""Test que set_execute_query hace commit correctamente"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
cursor_mock = MagicMock()
manager.connection.cursor.return_value = cursor_mock
# Act
result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")
# Assert
assert result is True
cursor_mock.execute.assert_called_once_with("INSERT INTO users VALUES (1, 'Alice')")
manager.connection.commit.assert_called_once()
cursor_mock.close.assert_called_once()
mock_disconnect.assert_called_once()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_set_execute_query_no_connection(self, mock_config, mock_connect, mock_disconnect):
"""Test que set_execute_query devuelve False si no hay conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = None
# Act
result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")
# Assert
assert result is False
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_set_execute_query_error(self, mock_config, mock_connect, mock_disconnect):
"""Test que set_execute_query maneja errores"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
cursor_mock = MagicMock()
cursor_mock.execute.side_effect = Exception("Constraint violation")
manager.connection.cursor.return_value = cursor_mock
# Act
result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")
# Assert
assert result is False
mock_disconnect.assert_called_once()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_log_execution_success(self, mock_config, mock_connect, mock_disconnect):
"""Test que log_execution ejecuta el SP correctamente"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = MagicMock()
manager.cursor = MagicMock()
manager.connection.cursor.return_value = manager.cursor
# Act
manager.log_execution(
procedure_id=1,
procedure_name='test_proc',
status='Success',
message='Test completed',
executed_by='test_user',
extra_info='None',
target_date='2023-01-01',
date='2023-01-01'
)
# Assert
assert manager.cursor.execute.called
manager.connection.commit.assert_called_once()
mock_disconnect.assert_called_once()
@patch.object(DatabaseManager, '_disconnect')
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_log_execution_no_connection(self, mock_config, mock_connect, mock_disconnect):
"""Test que log_execution maneja falta de conexión"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
manager = DatabaseManager('test_db', 'test_id')
manager.connection = None
# Act & Assert (no debe lanzar excepción)
manager.log_execution(
procedure_id=1,
procedure_name='test_proc',
status='Error',
message='Connection failed',
executed_by='test_user',
extra_info='None',
target_date='2023-01-01',
date='2023-01-01'
)
class TestGetDbConnection:
"""Tests para la función get_db_connection"""
@patch.object(DatabaseManager, '_connect')
@patch('App.Utils.db_utils.ConfigDatabase')
def test_get_db_connection_returns_connection(self, mock_config, mock_connect):
"""Test que get_db_connection devuelve una conexión pyodbc"""
# Arrange
mock_config.db_user = 'user'
mock_config.db_password = 'pass'
mock_config.db_host = 'server'
mock_connection = MagicMock()
with patch.object(DatabaseManager, '__init__', return_value=None):
with patch.object(DatabaseManager, '_connect'):
manager = DatabaseManager.__new__(DatabaseManager)
manager.connection = mock_connection
# Act (simulamos el comportamiento)
conn = manager.connection
# Assert
assert conn == mock_connection
@patch('App.Utils.db_utils.DatabaseManager')
def test_get_db_connection_params(self, mock_db_manager_class):
"""Test que get_db_connection pasa parámetros correctamente"""
# Arrange
mock_instance = MagicMock()
mock_instance.connection = MagicMock()
mock_db_manager_class.return_value = mock_instance
# Act
conn = get_db_connection(database='custom_db', databaseID='custom_id')
# Assert
mock_db_manager_class.assert_called_once_with('custom_db', 'custom_id')
mock_instance._connect.assert_called_once()
tests/test_utils.py
+19
View File
@@ -0,0 +1,19 @@
import requests
url = "https://automation.glmanalytics.com/webhook/new-task-clickup"
payload = {
"id": "12345",
"title": "Prueba ticket desde Python",
"description": "Este es un ticket de prueba enviado manualmente",
"priority": "3",
"status": "Nuevo",
"requester": "Juan Perez",
"category": "Soporte",
"created_at": "2026-04-09 10:00:00"
}
response = requests.post(url, json=payload)
print(response.status_code)
print(response.text)