refactor_Standardize_code_quality_fix_tests_and_archive_legacy_code

Author: DysTelefonica-legacy

.github/workflows/python-ci.yml

@@ -0,0 +1,9 @@
+# Herramientas de desarrollo y testing
+pytest>=7.4.0
+pytest-cov>=4.1.0
+pytest-mock>=3.12.0
+
+# Calidad de código
+black>=23.0.0
+ruff  # Reemplaza flake8 y añade más funcionalidades
+mypy>=1.7.0

coverage.json

@@ -0,0 +1,83 @@
+"""Integration-style test for unified EmailServicesTask processing both sources.
+Simulates DB pools for 'correos' and 'tareas' without touching real Access files.
+"""
+from unittest.mock import MagicMock, patch
+
+from email_services.email_task import EmailServicesTask
+
+
+class DummyPool:
+    def __init__(self, rows):
+        self.rows = rows
+        self.queries = []
+        self.updates = []
+
+    def execute_query(self, q):  # simplistic capture
+        self.queries.append(q)
+        return list(self.rows)
+
+    def update_record(self, table, data, where):
+        self.updates.append((table, data, where))
+        return True
+
+
+@patch("email_services.email_manager.smtplib.SMTP")
+@patch("email_services.email_manager.get_tareas_connection_pool")
+@patch("email_services.email_manager.get_correos_connection_pool")
+@patch("email_services.email_manager.config")
+def test_email_services_task_full_cycle(
+    mock_config, mock_get_correos_pool, mock_get_tareas_pool, mock_smtp
+):
+    # Configure fake config
+    mock_config.smtp_server = "localhost"
+    mock_config.smtp_port = 25
+    mock_config.smtp_user = "user@example.com"
+    mock_config.smtp_password = None
+    mock_config.smtp_tls = False
+    mock_config.get_db_correos_connection_string.return_value = "c_conn"
+    mock_config.get_db_connection_string.return_value = "t_conn"
+
+    # Two pending in correos, one in tareas
+    correos_rows = [
+        {
+            "IDCorreo": 1,
+            "Aplicacion": "App",
+            "Destinatarios": "a@example.com",
+            "Asunto": "A1",
+            "Cuerpo": "Body1",
+        },
+        {
+            "IDCorreo": 2,
+            "Aplicacion": "App",
+            "Destinatarios": "b@example.com",
+            "Asunto": "A2",
+            "Cuerpo": "Body2",
+        },
+    ]
+    tareas_rows = [
+        {
+            "IDCorreo": 10,
+            "Aplicacion": "Task",
+            "Destinatarios": "c@example.com",
+            "Asunto": "T1",
+            "Cuerpo": "TB",
+        },
+    ]
+
+    correos_pool = DummyPool(correos_rows)
+    tareas_pool = DummyPool(tareas_rows)
+    mock_get_correos_pool.return_value = correos_pool
+    mock_get_tareas_pool.return_value = tareas_pool
+
+    smtp_inst = MagicMock()
+    mock_smtp.return_value.__enter__.return_value = smtp_inst
+
+    task = EmailServicesTask()
+    ok = task.execute_specific_logic()
+
+    assert ok is True
+    # We expect three sendmail calls total
+    assert smtp_inst.sendmail.call_count == 3
+    # Each pool should have update calls matching its rows
+    assert len(correos_pool.updates) == 2
+    assert len(tareas_pool.updates) == 1

legacy_archive.zip

@@ -0,0 +1,54 @@
+import pytest
+
+from common.config import Config
+from common.config import config as global_config
+
+
+@pytest.mark.parametrize(
+    "db_type,alias_method,kwargs",
+    [
+        ("agedys", "get_db_agedys_connection_string", {}),
+        ("brass", "get_db_brass_connection_string", {}),
+        ("tareas", "get_db_tareas_connection_string", {}),
+        ("correos", "get_db_correos_connection_string", {"with_password": True}),
+        ("correos", "get_db_correos_connection_string", {"with_password": False}),
+        ("riesgos", "get_db_riesgos_connection_string", {}),
+        ("expedientes", "get_db_expedientes_connection_string", {}),
+        ("no_conformidades", "get_db_no_conformidades_connection_string", {}),
+    ],
+)
+def test_legacy_alias_methods_match_generic(db_type, alias_method, kwargs):
+    """Cada método legacy debe producir exactamente la misma cadena que el método genérico."""
+    # Usar una nueva instancia para evitar que otros tests hayan mutado rutas
+    cfg = Config()
+
+    # Resultado alias
+    alias_func = getattr(cfg, alias_method)
+    alias_value = alias_func(**kwargs) if kwargs else alias_func()
+
+    # Resultado genérico (con posible control de password)
+    with_password = kwargs.get("with_password", True)
+    generic_value = cfg.get_db_connection_string(db_type, with_password=with_password)
+
+    assert alias_value == generic_value, (
+        f"Alias {alias_method} difiere del genérico para {db_type}.\n"
+        f"Alias:   {alias_value}\nGenérico: {generic_value}"
+    )
+
+
+def test_global_config_instance_alias_equivalence():
+    """Verifica también contra la instancia global por si se usa en producción directamente."""
+    # Asegurar que la instancia global se comporta igual al crear otra nueva
+    fresh = Config()
+    assert (
+        global_config.get_db_brass_connection_string()
+        == fresh.get_db_brass_connection_string()
+    )
+    assert global_config.get_db_connection_string(
+        "brass"
+    ) == fresh.get_db_connection_string("brass")
+    # Consistencia cruzada alias vs genérico en la instancia global
+    assert (
+        global_config.get_db_brass_connection_string()
+        == global_config.get_db_connection_string("brass")
+    )

requirements-dev.txt

@@ -0,0 +1,71 @@
+from unittest.mock import MagicMock
+
+from common.utils import execute_task_with_standard_boilerplate
+
+
+class DummyTask:
+    """Mock sencillo de una Task con interfaz mínima utilizada por el helper."""
+
+    def __init__(self):
+        self.debe_ejecutarse = MagicMock(return_value=True)
+        self.execute_specific_logic = MagicMock(return_value=True)
+        self.marcar_como_completada = MagicMock()
+        # Soporta context manager
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+def test_no_force_not_due_skips_logic(tmp_path, monkeypatch):
+    task = DummyTask()
+    task.debe_ejecutarse.return_value = False
+
+    exit_code = execute_task_with_standard_boilerplate(
+        "Dummy", task_obj=task, log_file=tmp_path / "dummy.log"
+    )
+
+    task.execute_specific_logic.assert_not_called()
+    task.marcar_como_completada.assert_not_called()
+    assert exit_code == 0
+
+
+def test_no_force_due_executes_and_marks(tmp_path):
+    task = DummyTask()
+    task.debe_ejecutarse.return_value = True
+
+    exit_code = execute_task_with_standard_boilerplate(
+        "Dummy", task_obj=task, log_file=tmp_path / "dummy.log"
+    )
+
+    task.execute_specific_logic.assert_called_once()
+    task.marcar_como_completada.assert_called_once()
+    assert exit_code == 0
+
+
+def test_force_executes_without_mark(tmp_path):
+    task = DummyTask()
+
+    exit_code = execute_task_with_standard_boilerplate(
+        "Dummy", task_obj=task, force=True, log_file=tmp_path / "dummy.log"
+    )
+
+    task.execute_specific_logic.assert_called_once()
+    task.marcar_como_completada.assert_not_called()
+    assert exit_code == 0
+
+
+def test_execute_specific_logic_exception_returns_error_code(tmp_path):
+    task = DummyTask()
+    task.execute_specific_logic.side_effect = RuntimeError("boom")
+
+    exit_code = execute_task_with_standard_boilerplate(
+        "Dummy", task_obj=task, log_file=tmp_path / "dummy.log"
+    )
+
+    task.execute_specific_logic.assert_called_once()
+    # Como falla, no se debería marcar
+    task.marcar_como_completada.assert_not_called()
+    assert exit_code == 1