egutierrez
86d68dc9f0
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>
95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
"""Tests for run_mssql_query: composition of mssql_connect + mssql_query.
|
|
|
|
Mock-based, no real SQL Server. The pipeline binds `mssql_connect` and
|
|
`mssql_query` as module-level names, so we monkeypatch them in the pipeline's
|
|
namespace and assert the orchestration (connect -> query -> always close).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import pipelines.run_mssql_query as mod
|
|
from pipelines.run_mssql_query import run_mssql_query, _to_csv
|
|
|
|
|
|
class FakeConn:
|
|
def __init__(self):
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
def test_run_mssql_query_composes_connect_and_query(monkeypatch):
|
|
fake_conn = FakeConn()
|
|
connect_calls = {}
|
|
query_calls = {}
|
|
|
|
def fake_connect(host, database, user, password, **kwargs):
|
|
connect_calls.update(
|
|
host=host, database=database, user=user, password=password, **kwargs
|
|
)
|
|
return fake_conn
|
|
|
|
sentinel = {"columns": ["No_"], "rows": [{"No_": "CLI-1"}], "row_count": 1}
|
|
|
|
def fake_query(conn, sql, params=None, max_rows=None):
|
|
query_calls.update(conn=conn, sql=sql, params=params, max_rows=max_rows)
|
|
return sentinel
|
|
|
|
monkeypatch.setattr(mod, "mssql_connect", fake_connect)
|
|
monkeypatch.setattr(mod, "mssql_query", fake_query)
|
|
|
|
result = run_mssql_query(
|
|
"10.0.0.5", "navdb", "sa", "pw",
|
|
"SELECT [No_] FROM [dbo].[Cartera] WHERE [Customer No_] = %s",
|
|
params=("CLI-0001",), port=1433, max_rows=5,
|
|
)
|
|
|
|
# Returns exactly what mssql_query produced.
|
|
assert result is sentinel
|
|
# Connection was opened with the supplied params.
|
|
assert connect_calls["host"] == "10.0.0.5"
|
|
assert connect_calls["database"] == "navdb"
|
|
assert connect_calls["port"] == 1433
|
|
# Query borrowed the open connection and got the bound params (no interpolation).
|
|
assert query_calls["conn"] is fake_conn
|
|
assert query_calls["params"] == ("CLI-0001",)
|
|
assert query_calls["max_rows"] == 5
|
|
# Connection is always closed.
|
|
assert fake_conn.closed is True
|
|
|
|
|
|
def test_run_mssql_query_closes_connection_on_error(monkeypatch):
|
|
fake_conn = FakeConn()
|
|
|
|
monkeypatch.setattr(mod, "mssql_connect", lambda *a, **k: fake_conn)
|
|
|
|
def boom(conn, sql, params=None, max_rows=None):
|
|
raise RuntimeError("mssql_query failed executing query: boom")
|
|
|
|
monkeypatch.setattr(mod, "mssql_query", boom)
|
|
|
|
with pytest.raises(RuntimeError, match="failed executing query"):
|
|
run_mssql_query("10.0.0.5", "navdb", "sa", "pw", "SELECT 1")
|
|
|
|
# Even on a query error, the connection is closed (try/finally).
|
|
assert fake_conn.closed is True
|
|
|
|
|
|
def test_to_csv_renders_header_and_rows():
|
|
result = {
|
|
"columns": ["No_", "Amount"],
|
|
"rows": [
|
|
{"No_": "CLI-1", "Amount": 100},
|
|
{"No_": "CLI-2", "Amount": 200},
|
|
],
|
|
"row_count": 2,
|
|
}
|
|
csv_text = _to_csv(result)
|
|
lines = csv_text.strip().splitlines()
|
|
assert lines[0] == "No_,Amount"
|
|
assert lines[1] == "CLI-1,100"
|
|
assert lines[2] == "CLI-2,200"
|