feat(infra): conexion y consulta directa a SQL Server (Navision) via pymssql

Grupo de capacidad nuevo 'sql-connect' (3 funciones) para conectar a un
Microsoft SQL Server (donde corre Navision) y consultar directamente, en
lugar del ida y vuelta manual de pegar CSVs.

- mssql_connect_py_infra: abre conexion pymssql (login_timeout acotado,
  credenciales por argumento, RuntimeError claro si falla).
- mssql_query_py_infra: SELECT parametrizada con binding seguro (sin
  inyeccion) sobre conexion abierta; devuelve {columns, rows, row_count};
  0 filas -> lista vacia; max_rows con fetchmany; read-only.
- run_mssql_query_py_pipelines: one-shot que compone connect+query y cierra
  siempre; CLI imprime JSON o CSV; contrasena desde env var (pass).

Pagina madre docs/capabilities/sql-connect.md + fila en INDEX.md.
Dependencia pymssql>=2.3.13 anadida a python/pyproject.toml + uv.lock.
Tests mock-based (11) verdes; error path verificado end-to-end contra el
driver real (host inalcanzable -> RuntimeError, acotado por login_timeout).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-22 11:29:49 +02:00
parent c1f355ffa5
commit 86d68dc9f0
13 changed files with 930 additions and 0 deletions
+133
View File
@@ -0,0 +1,133 @@
"""Tests para mssql_query usando un doble de prueba (sin servidor real)."""
from __future__ import annotations
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", ".."))
from functions.infra.mssql_query import mssql_query
def _desc(*names):
"""Construye una description estilo DB-API: una tupla 7-elem por columna."""
return [(name, None, None, None, None, None, None) for name in names]
class FakeCursor:
"""Doble de prueba de un cursor DB-API (pymssql-like)."""
def __init__(self, description=None, rows=None):
self.description = description
self._rows = list(rows or [])
self.executed = None # (sql, params) de la ultima execute
self.fetchmany_calls = [] # tamaños pedidos a fetchmany
self.closed = False
def execute(self, sql, params=None):
self.executed = (sql, params)
def fetchall(self):
return list(self._rows)
def fetchmany(self, size):
self.fetchmany_calls.append(size)
return list(self._rows[:size])
def close(self):
self.closed = True
class FakeConn:
"""Doble de prueba de una conexion: devuelve un FakeCursor fijo."""
def __init__(self, cursor):
self._cursor = cursor
def cursor(self):
return self._cursor
def test_golden_maps_rows_to_dicts():
cur = FakeCursor(
description=_desc("No_", "Amount"),
rows=[("CLI-1", 100), ("CLI-2", 200)],
)
conn = FakeConn(cur)
result = mssql_query(conn, "SELECT No_, Amount FROM Cartera")
assert result == {
"columns": ["No_", "Amount"],
"rows": [
{"No_": "CLI-1", "Amount": 100},
{"No_": "CLI-2", "Amount": 200},
],
"row_count": 2,
}
assert cur.closed is True
def test_binding_passes_params_to_driver():
cur = FakeCursor(description=_desc("No_"), rows=[("CLI-0001",)])
conn = FakeConn(cur)
sql = "SELECT No_ FROM Cartera WHERE [Customer No_] = %s"
mssql_query(conn, sql, params=("CLI-0001",))
# El SQL y los params llegan al driver tal cual: binding, no interpolacion.
assert cur.executed == (sql, ("CLI-0001",))
def test_zero_rows_no_error():
cur = FakeCursor(description=_desc("No_", "Amount"), rows=[])
conn = FakeConn(cur)
result = mssql_query(conn, "SELECT No_, Amount FROM Cartera WHERE 1 = 0")
assert result["rows"] == []
assert result["row_count"] == 0
assert result["columns"] == ["No_", "Amount"]
def test_max_rows_uses_fetchmany():
cur = FakeCursor(
description=_desc("No_"),
rows=[("CLI-1",), ("CLI-2",), ("CLI-3",)],
)
conn = FakeConn(cur)
result = mssql_query(conn, "SELECT No_ FROM Cartera", max_rows=1)
assert cur.fetchmany_calls == [1]
assert result["row_count"] == 1
assert result["rows"] == [{"No_": "CLI-1"}]
def test_description_none_empty_columns():
cur = FakeCursor(description=None, rows=[])
conn = FakeConn(cur)
result = mssql_query(conn, "SET NOCOUNT ON")
assert result["columns"] == []
assert result["rows"] == []
assert result["row_count"] == 0
def test_execution_error_raises_runtimeerror():
class BoomCursor(FakeCursor):
def execute(self, sql, params=None):
raise ValueError("boom")
cur = BoomCursor()
conn = FakeConn(cur)
with pytest.raises(RuntimeError, match="mssql_query failed executing query"):
mssql_query(conn, "SELECT 1")
# El cursor se cierra incluso en error (try/finally).
assert cur.closed is True