From 49379c71b3ada889193370781e9eabdd45067b37 Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Thu, 13 Jun 2024 16:34:40 +0530 Subject: [PATCH 1/7] feat : added esquema_sql test cases Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 2 +- tests/test_formato_sql.py | 289 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 290 insertions(+), 1 deletion(-) create mode 100644 tests/test_formato_sql.py diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index 0d5786cd9..d3641e536 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -32,7 +32,7 @@ def esquema_sql(tipos_registro, conf={}): - from formato_txt import A, N, I + from .formato_txt import A, N, I for tabla, formato in tipos_registro: sql = [] diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py new file mode 100644 index 000000000..3c5d8ee9b --- /dev/null +++ b/tests/test_formato_sql.py @@ -0,0 +1,289 @@ +#!/usr/bin/python +# -*- coding: utf8 -*- +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by the +# Free Software Foundation; either version 3, or (at your option) any later +# version. +# +# This program is distributed in the hope that it will be useful, but +# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTIBILITY +# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License +# for more details. + +"""Test para formato_sql""" + +__author__ = "Mariano Reingart " +__copyright__ = "Copyright (C) 2010-2019 Mariano Reingart" +__license__ = "GPL 3.0" + +import pytest +from pyafipws.formatos.formato_sql import esquema_sql +from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO + +@pytest.mark.dontusefix +class TestFormatoSQL: + def test_esquema_sql_encabezado(self, auth): + # Prueba para generar el esquema SQL de la tabla "encabezado" + tipos_registro = [("encabezado", ENCABEZADO)] + conf = {} + expected_output = [ + "CREATE TABLE encabezado (\n" + " tipo_reg INTEGER ,\n" + " webservice VARCHAR (6),\n" + " fecha_cbte VARCHAR (8),\n" + " tipo_cbte INTEGER ,\n" + " punto_vta INTEGER ,\n" + " cbte_nro INTEGER ,\n" + " tipo_expo INTEGER ,\n" + " permiso_existente VARCHAR (1),\n" + " pais_dst_cmp INTEGER ,\n" + " nombre_cliente VARCHAR (200),\n" + " tipo_doc INTEGER ,\n" + " nro_doc INTEGER ,\n" + " domicilio_cliente VARCHAR (300),\n" + " id_impositivo VARCHAR (50),\n" + " imp_total NUMERIC (15, 3),\n" + " imp_tot_conc NUMERIC (15, 3),\n" + " imp_neto NUMERIC (15, 3),\n" + " impto_liq NUMERIC (15, 3),\n" + " impto_liq_nri NUMERIC (15, 3),\n" + " imp_op_ex NUMERIC (15, 3),\n" + " impto_perc NUMERIC (15, 2),\n" + " imp_iibb NUMERIC (15, 3),\n" + " impto_perc_mun NUMERIC (15, 3),\n" + " imp_internos NUMERIC (15, 3),\n" + " imp_trib NUMERIC (15, 3),\n" + " moneda_id VARCHAR (3),\n" + " moneda_ctz NUMERIC (10, 6),\n" + " obs_comerciales VARCHAR (1000),\n" + " obs_generales VARCHAR (1000),\n" + " forma_pago VARCHAR (50),\n" + " incoterms VARCHAR (3),\n" + " incoterms_ds VARCHAR (20),\n" + " idioma_cbte VARCHAR (1),\n" + " zona VARCHAR (5),\n" + " fecha_venc_pago VARCHAR (8),\n" + " presta_serv INTEGER ,\n" + " fecha_serv_desde VARCHAR (8),\n" + " fecha_serv_hasta VARCHAR (8),\n" + " cae VARCHAR (14),\n" + " fecha_vto VARCHAR (8),\n" + " resultado VARCHAR (1),\n" + " reproceso VARCHAR (1),\n" + " motivos_obs VARCHAR (1000),\n" + " id INTEGER PRIMARY KEY,\n" + " telefono_cliente VARCHAR (50),\n" + " localidad_cliente VARCHAR (50),\n" + " provincia_cliente VARCHAR (50),\n" + " formato_id INTEGER ,\n" + " email VARCHAR (100),\n" + " pdf VARCHAR (100),\n" + " err_code VARCHAR (6),\n" + " err_msg VARCHAR (1000),\n" + " Dato_adicional1 VARCHAR (30),\n" + " Dato_adicional2 VARCHAR (30),\n" + " Dato_adicional3 VARCHAR (30),\n" + " Dato_adicional4 VARCHAR (30),\n" + " descuento NUMERIC (15, 3),\n" + " cbt_desde INTEGER ,\n" + " cbt_hasta INTEGER ,\n" + " concepto INTEGER ,\n" + " no_usar NUMERIC (15, 3),\n" + " imp_iva NUMERIC (15, 3),\n" + " emision_tipo VARCHAR (4),\n" + " imp_subtotal NUMERIC (15, 3),\n" + " cat_iva INTEGER ,\n" + " tipo_cbte INTEGER ,\n" + " punto_vta INTEGER ,\n" + " tipo_cod_aut VARCHAR (1)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_detalle(self, auth): + # Prueba para generar el esquema SQL de la tabla "detalle" + tipos_registro = [("detalle", DETALLE)] + conf = {} + expected_output = [ + "CREATE TABLE detalle (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " codigo VARCHAR (30),\n" + " qty NUMERIC (12, 2),\n" + " umed INTEGER ,\n" + " precio NUMERIC (12, 3),\n" + " importe NUMERIC (14, 3),\n" + " iva_id INTEGER ,\n" + " ds VARCHAR (4000),\n" + " ncm VARCHAR (15),\n" + " sec VARCHAR (15),\n" + " bonif NUMERIC (15, 2),\n" + " imp_iva NUMERIC (15, 2),\n" + " despacho VARCHAR (20),\n" + " u_mtx INTEGER ,\n" + " cod_mtx INTEGER ,\n" + " dato_a VARCHAR (15),\n" + " dato_b VARCHAR (15),\n" + " dato_c VARCHAR (15),\n" + " dato_d VARCHAR (15),\n" + " dato_e VARCHAR (15)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_tributo(self, auth): + # Prueba para generar el esquema SQL de la tabla "tributo" + tipos_registro = [("tributo", TRIBUTO)] + conf = {} + expected_output = [ + "CREATE TABLE tributo (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " tributo_id INTEGER ,\n" + " desc VARCHAR (100),\n" + " base_imp NUMERIC (15, 3),\n" + " alic NUMERIC (15, 2),\n" + " importe NUMERIC (15, 3)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_iva(self, auth): + # Prueba para generar el esquema SQL de la tabla "iva" + tipos_registro = [("iva", IVA)] + conf = {} + expected_output = [ + "CREATE TABLE iva (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " iva_id INTEGER ,\n" + " base_imp NUMERIC (15, 3),\n" + " importe NUMERIC (15, 3)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_cmp_asoc(self, auth): + # Prueba para generar el esquema SQL de la tabla "cmp_asoc" + tipos_registro = [("cmp_asoc", CMP_ASOC)] + conf = {} + expected_output = [ + "CREATE TABLE cmp_asoc (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " cbte_tipo INTEGER ,\n" + " cbte_punto_vta INTEGER ,\n" + " cbte_nro INTEGER ,\n" + " cbte_fecha INTEGER ,\n" + " cbte_cuit INTEGER \n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_permiso(self, auth): + # Prueba para generar el esquema SQL de la tabla "permiso" + tipos_registro = [("permiso", PERMISO)] + conf = {} + expected_output = [ + "CREATE TABLE permiso (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " id_permiso VARCHAR (16),\n" + " dst_merc INTEGER \n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_detalle_with_conf(self, auth): + # Prueba para generar el esquema SQL de la tabla "detalle" con configuración personalizada + tipos_registro = [("detalle", DETALLE)] + conf = {"detalle": {"qty": "cantidad"}} + expected_output = [ + "CREATE TABLE detalle (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " codigo VARCHAR (30),\n" + " cantidad NUMERIC (12, 2),\n" + " umed INTEGER ,\n" + " precio NUMERIC (12, 3),\n" + " importe NUMERIC (14, 3),\n" + " iva_id INTEGER ,\n" + " ds VARCHAR (4000),\n" + " ncm VARCHAR (15),\n" + " sec VARCHAR (15),\n" + " bonif NUMERIC (15, 2),\n" + " imp_iva NUMERIC (15, 2),\n" + " despacho VARCHAR (20),\n" + " u_mtx INTEGER ,\n" + " cod_mtx INTEGER ,\n" + " dato_a VARCHAR (15),\n" + " dato_b VARCHAR (15),\n" + " dato_c VARCHAR (15),\n" + " dato_d VARCHAR (15),\n" + " dato_e VARCHAR (15)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_tributo_with_conf(self, auth): + # Prueba para generar el esquema SQL de la tabla "tributo" con configuración personalizada + tipos_registro = [("tributo", TRIBUTO)] + conf = {"tributo": {"desc": "descripcion"}} + expected_output = [ + "CREATE TABLE tributo (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " tributo_id INTEGER ,\n" + " descripcion VARCHAR (100),\n" + " base_imp NUMERIC (15, 3),\n" + " alic NUMERIC (15, 2),\n" + " importe NUMERIC (15, 3)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_iva_with_conf(self, auth): + # Prueba para generar el esquema SQL de la tabla "iva" con configuración personalizada + tipos_registro = [("iva", IVA)] + conf = {"iva": {"importe": "imp_iva"}} + expected_output = [ + "CREATE TABLE iva (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " iva_id INTEGER ,\n" + " base_imp NUMERIC (15, 3),\n" + " imp_iva NUMERIC (15, 3)\n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_cmp_asoc_with_conf(self, auth): + # Prueba para generar el esquema SQL de la tabla "cmp_asoc" con configuración personalizada + tipos_registro = [("cmp_asoc", CMP_ASOC)] + conf = {"cmp_asoc": {"cbte_nro": "numero"}} + expected_output = [ + "CREATE TABLE cmp_asoc (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " cbte_tipo INTEGER ,\n" + " cbte_punto_vta INTEGER ,\n" + " numero INTEGER ,\n" + " cbte_fecha INTEGER ,\n" + " cbte_cuit INTEGER \n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output + + def test_esquema_sql_permiso_with_conf(self, auth): + # Prueba para generar el esquema SQL de la tabla "permiso" con configuración personalizada + tipos_registro = [("permiso", PERMISO)] + conf = {"permiso": {"dst_merc": "destino"}} + expected_output = [ + "CREATE TABLE permiso (\n" + " id INTEGER FOREING KEY encabezado,\n" + " tipo_reg INTEGER ,\n" + " id_permiso VARCHAR (16),\n" + " destino INTEGER \n" + ")\n;" + ] + assert list(esquema_sql(tipos_registro, conf)) == expected_output From a2229c5e7733a23f9c5c1d6fb4880bfbba2f5faa Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Sun, 16 Jun 2024 17:02:38 +0530 Subject: [PATCH 2/7] feat: added tests for 'configurar' and 'ejecutar' functions Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 19 ++-- tests/test_formato_sql.py | 229 +++++++++++++++++++++++++++++++++++++- 2 files changed, 240 insertions(+), 8 deletions(-) diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index d3641e536..dea80505e 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -92,16 +92,20 @@ def configurar(schema): tablas = {} campos = {} campos_rev = {} - if not schema: - for tabla in "encabezado", "detalle", "cmp_asoc", "permiso", "tributo", "iva": - tablas[tabla] = tabla - campos[tabla] = {"id": "id"} - campos_rev[tabla] = dict([(v, k) for k, v in list(campos[tabla].items())]) + for tabla in "encabezado", "detalle", "cmp_asoc", "permiso", "tributo", "iva": + tablas[tabla] = tabla + campos[tabla] = {"id": "id"} + campos_rev[tabla] = {"id": "id"} + if schema and tabla in schema: + for campo, valor in schema[tabla].items(): + campos[tabla][campo] = valor + campos_rev[tabla][valor] = campo return tablas, campos, campos_rev def ejecutar(cur, sql, params=None): - ##print sql, params + if DEBUG: + print(sql, params) if params is None: return cur.execute(sql) else: @@ -117,7 +121,8 @@ def max_id(db, schema={}): ret = None try: ejecutar(cur, query) - for row in cur: + row = cur.fetchone() + if row: ret = row[0] if not ret: ret = 0 diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index 3c5d8ee9b..ee92819ba 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -17,7 +17,8 @@ __license__ = "GPL 3.0" import pytest -from pyafipws.formatos.formato_sql import esquema_sql +from unittest.mock import MagicMock, patch +from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, escribir from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO @pytest.mark.dontusefix @@ -287,3 +288,229 @@ def test_esquema_sql_permiso_with_conf(self, auth): ")\n;" ] assert list(esquema_sql(tipos_registro, conf)) == expected_output + + + +@pytest.mark.dontusefix +class TestConfigurar: + def test_configurar_without_schema(self): + # Test configuring without a custom schema + schema = {} + expected_tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "tributo": "tributo", + "iva": "iva", + } + expected_campos = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + expected_campos_rev = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + tablas, campos, campos_rev = configurar(schema) + assert tablas == expected_tablas + assert campos == expected_campos + assert campos_rev == expected_campos_rev + + def test_configurar_with_custom_schema(self): + # Test configuring with a custom schema + schema = { + "encabezado": {"nombre": "encabezado_nombre"}, + "detalle": {"cantidad": "detalle_cantidad"}, + "cmp_asoc": {"numero": "cmp_asoc_numero"}, + "permiso": {"destino": "permiso_destino"}, + "tributo": {"descripcion": "tributo_descripcion"}, + "iva": {"importe": "iva_importe"}, + } + expected_tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "tributo": "tributo", + "iva": "iva", + } + expected_campos = { + "encabezado": {"nombre": "encabezado_nombre", "id": "id"}, + "detalle": {"cantidad": "detalle_cantidad", "id": "id"}, + "cmp_asoc": {"numero": "cmp_asoc_numero", "id": "id"}, + "permiso": {"destino": "permiso_destino", "id": "id"}, + "tributo": {"descripcion": "tributo_descripcion", "id": "id"}, + "iva": {"importe": "iva_importe", "id": "id"}, + } + expected_campos_rev = { + "encabezado": {"encabezado_nombre": "nombre", "id": "id"}, + "detalle": {"detalle_cantidad": "cantidad", "id": "id"}, + "cmp_asoc": {"cmp_asoc_numero": "numero", "id": "id"}, + "permiso": {"permiso_destino": "destino", "id": "id"}, + "tributo": {"tributo_descripcion": "descripcion", "id": "id"}, + "iva": {"iva_importe": "importe", "id": "id"}, + } + tablas, campos, campos_rev = configurar(schema) + assert tablas == expected_tablas + assert campos == expected_campos + assert campos_rev == expected_campos_rev + + def test_configurar_with_partial_schema(self): + # Test configuring with a partial schema + schema = { + "encabezado": {"nombre": "encabezado_nombre"}, + "detalle": {"cantidad": "detalle_cantidad"}, + } + expected_tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "tributo": "tributo", + "iva": "iva", + } + expected_campos = { + "encabezado": {"nombre": "encabezado_nombre", "id": "id"}, + "detalle": {"cantidad": "detalle_cantidad", "id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + expected_campos_rev = { + "encabezado": {"encabezado_nombre": "nombre", "id": "id"}, + "detalle": {"detalle_cantidad": "cantidad", "id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + tablas, campos, campos_rev = configurar(schema) + assert tablas == expected_tablas + assert campos == expected_campos + assert campos_rev == expected_campos_rev + + + def test_configurar_with_empty_schema(self): + # Test configuring with an empty schema + schema = {} + expected_tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "tributo": "tributo", + "iva": "iva", + } + expected_campos = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + expected_campos_rev = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + tablas, campos, campos_rev = configurar(schema) + assert tablas == expected_tablas + assert campos == expected_campos + assert campos_rev == expected_campos_rev + + def test_configurar_with_none_schema(self): + # Test configuring with a None schema + schema = None + expected_tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "tributo": "tributo", + "iva": "iva", + } + expected_campos = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + expected_campos_rev = { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "tributo": {"id": "id"}, + "iva": {"id": "id"}, + } + tablas, campos, campos_rev = configurar(schema) + assert tablas == expected_tablas + assert campos == expected_campos + assert campos_rev == expected_campos_rev + + + +@pytest.mark.dontusefix +class TestEjecutar: + + def test_ejecutar_without_params(self, mocker): + # Test executing a query without parameters + cur = MagicMock() + sql = "SELECT * FROM tabla" + mocker.patch('pyafipws.formatos.formato_sql.ejecutar', return_value=None) + ejecutar(cur, sql) + cur.execute.assert_called_once_with(sql) + + def test_ejecutar_with_params(self, mocker): + # Test executing a query with parameters + cur = MagicMock() + sql = "SELECT * FROM tabla WHERE id = ?" + params = (1,) + mocker.patch('pyafipws.formatos.formato_sql.ejecutar', return_value=None) + ejecutar(cur, sql, params) + cur.execute.assert_called_once_with(sql, params) + + def test_ejecutar_returns_cursor_execute_result(self, mocker): + # Test that ejecutar returns the result of cursor.execute + cur = MagicMock() + sql = "SELECT * FROM tabla" + expected_result = MagicMock() + cur.execute.return_value = expected_result + result = ejecutar(cur, sql) + assert result == expected_result + + def test_ejecutar_prints_sql_and_params_when_debug_true(self, mocker, capsys): + # Test that ejecutar prints the SQL query and parameters when DEBUG is True + cur = MagicMock() + sql = "SELECT * FROM tabla WHERE id = ?" + params = (1,) + mocker.patch('pyafipws.formatos.formato_sql.DEBUG', True) + ejecutar(cur, sql, params) + captured = capsys.readouterr() + assert captured.out == "SELECT * FROM tabla WHERE id = ? (1,)\n" + + def test_ejecutar_does_not_print_when_debug_false(self, mocker, capsys): + # Test that ejecutar does not print anything when DEBUG is False + cur = MagicMock() + sql = "SELECT * FROM tabla" + mocker.patch('pyafipws.formatos.formato_sql.DEBUG', False) + ejecutar(cur, sql) + captured = capsys.readouterr() + assert captured.out == "" + From 975900f02a35a6584c7165e57d1c2b64d86cb2b5 Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Tue, 18 Jun 2024 16:13:10 +0530 Subject: [PATCH 3/7] feat : added test cases for max_id, redondear and ecribir functions Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 13 +- tests/test_formato_sql.py | 295 +++++++++++++++++++++++++++++++++++++- 2 files changed, 300 insertions(+), 8 deletions(-) diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index dea80505e..848efeb2e 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -133,19 +133,19 @@ def max_id(db, schema={}): def redondear(formato, clave, valor): - from formato_txt import A, N, I + from pyafipws.formatos.formato_txt import A, N, I # corregir redondeo (aparentemente sqlite no guarda correctamente los decimal) import decimal try: - long = [fmt[1] for fmt in formato if fmt[0] == clave] + longitud = [fmt[1] for fmt in formato if fmt[0] == clave] tipo = [fmt[2] for fmt in formato if fmt[0] == clave] if not tipo: return valor tipo = tipo[0] if DEBUG: - print("tipo", tipo, clave, valor, int) + print("tipo", tipo, clave, valor, longitud) if valor is None: return None if valor == "": @@ -158,8 +158,8 @@ def redondear(formato, clave, valor): valor = str(valor) if isinstance(valor, basestring): valor = Decimal(valor) - if int and isinstance(int[0], (tuple, list)): - decimales = old_div(Decimal("1"), Decimal(10 ** (int[0][1]))) + if longitud and isinstance(longitud[0], (tuple, list)): + decimales = old_div(Decimal("1"), Decimal(10 ** (longitud[0][1]))) else: decimales = Decimal(".01") valor1 = valor.quantize(decimales, rounding=decimal.ROUND_DOWN) @@ -168,10 +168,11 @@ def redondear(formato, clave, valor): return valor1 except Exception as e: print("IMPOSIBLE REDONDEAR:", clave, valor, e) + return None def escribir(facts, db, schema={}, commit=True): - from formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO + from .formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO tablas, campos, campos_rev = configurar(schema) cur = db.cursor() diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index ee92819ba..7d9ca8df7 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -17,9 +17,10 @@ __license__ = "GPL 3.0" import pytest +from decimal import Decimal from unittest.mock import MagicMock, patch -from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, escribir -from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO +from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, redondear, escribir +from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, A, N, I @pytest.mark.dontusefix class TestFormatoSQL: @@ -514,3 +515,293 @@ def test_ejecutar_does_not_print_when_debug_false(self, mocker, capsys): captured = capsys.readouterr() assert captured.out == "" +@pytest.fixture +def db_mock(): + return MagicMock() + +@pytest.fixture +def cur_mock(db_mock): + cur_mock = MagicMock() + db_mock.cursor.return_value = cur_mock + return cur_mock + +@pytest.mark.dontusefix +class TestMaxId: + def test_max_id_with_existing_id(self, db_mock, cur_mock, mocker): + schema = {} + expected_max_id = 42 + cur_mock.fetchone.return_value = (expected_max_id,) + configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + result = max_id(db_mock, schema) + + assert result == expected_max_id + db_mock.cursor.assert_called_once() + configurar_mock.assert_called_once_with(schema) + ejecutar_mock.assert_called_once() + cur_mock.fetchone.assert_called_once() + cur_mock.close.assert_called_once() + + def test_max_id_with_no_existing_id(self, db_mock, cur_mock, mocker): + schema = {} + cur_mock.fetchone.return_value = (None,) + configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + result = max_id(db_mock, schema) + + assert result == 0 + db_mock.cursor.assert_called_once() + configurar_mock.assert_called_once_with(schema) + ejecutar_mock.assert_called_once() + cur_mock.fetchone.assert_called_once() + cur_mock.close.assert_called_once() + + def test_max_id_with_custom_schema(self, db_mock, cur_mock, mocker): + schema = {'encabezado': {'custom_id': 'id'}} + expected_max_id = 42 + cur_mock.fetchone.return_value = (expected_max_id,) + configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'custom_encabezado'}, {'encabezado': {'id': 'custom_id'}}, {})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + result = max_id(db_mock, schema) + + assert result == expected_max_id + db_mock.cursor.assert_called_once() + configurar_mock.assert_called_once_with(schema) + ejecutar_mock.assert_called_once() + cur_mock.fetchone.assert_called_once() + cur_mock.close.assert_called_once() + + def test_max_id_with_empty_result(self, db_mock, cur_mock, mocker): + schema = {} + cur_mock.fetchone.return_value = None + configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + result = max_id(db_mock, schema) + + assert result == 0 + db_mock.cursor.assert_called_once() + configurar_mock.assert_called_once_with(schema) + ejecutar_mock.assert_called_once() + cur_mock.fetchone.assert_called_once() + cur_mock.close.assert_called_once() + + def test_max_id_with_exception(self, db_mock, cur_mock, mocker): + schema = {} + configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar', side_effect=Exception('Database error')) + + with pytest.raises(Exception) as exc_info: + max_id(db_mock, schema) + + assert str(exc_info.value) == 'Database error' + db_mock.cursor.assert_called_once() + configurar_mock.assert_called_once_with(schema) + ejecutar_mock.assert_called_once() + cur_mock.fetchone.assert_not_called() + cur_mock.close.assert_called_once() + + + +@pytest.mark.dontusefix +class TestRedondear: + def test_redondear_with_invalid_key(self): + formato = [("campo1", 10, A), ("campo2", 15, N)] + clave = "campo3" + valor = "123" + + result = redondear(formato, clave, valor) + + assert result == "123" + + def test_redondear_with_none_value(self): + formato = [("campo1", 10, A), ("campo2", 15, N)] + clave = "campo1" + valor = None + + result = redondear(formato, clave, valor) + + assert result is None + + def test_redondear_with_empty_string_value(self): + formato = [("campo1", 10, A), ("campo2", 15, N)] + clave = "campo1" + valor = "" + + result = redondear(formato, clave, valor) + + assert result == "" + + def test_redondear_with_tipo_a(self): + formato = [("campo1", 10, A), ("campo2", 15, N)] + clave = "campo1" + valor = "ABC123" + + result = redondear(formato, clave, valor) + + assert result == "ABC123" + + def test_redondear_with_tipo_n(self): + formato = [("campo1", 10, A), ("campo2", 15, N)] + clave = "campo2" + valor = "123" + + result = redondear(formato, clave, valor) + + assert result == 123 + + def test_redondear_with_int_value(self): + formato = [("campo1", 10, A), ("campo2", (15, 2), I)] + clave = "campo2" + valor = 123 + + result = redondear(formato, clave, valor) + + assert result == Decimal("123.00") + + def test_redondear_with_float_value(self): + formato = [("campo1", 10, A), ("campo2", (15, 2), I)] + clave = "campo2" + valor = 123.456 + + result = redondear(formato, clave, valor) + + assert result == Decimal("123.45") + + def test_redondear_with_string_value(self): + formato = [("campo1", 10, A), ("campo2", (15, 2), I)] + clave = "campo2" + valor = "123.456" + + result = redondear(formato, clave, valor) + + assert result == Decimal("123.45") + + def test_redondear_with_custom_decimals(self): + formato = [("campo1", 10, A), ("campo2", (15, 3), I)] + clave = "campo2" + valor = "123.4567" + + result = redondear(formato, clave, valor) + + assert result == Decimal("123.456") + + def test_redondear_with_exception(self, mocker): + formato = [("campo1", 10, A), ("campo2", (15, 2), I)] + clave = "campo2" + valor = "invalid" + + mocker.patch("builtins.print") + + result = redondear(formato, clave, valor) + + assert result is None + print.assert_called_once_with("IMPOSIBLE REDONDEAR:", clave, valor, mocker.ANY) + + + + +@pytest.fixture +def db_mock(): + return MagicMock() + +@pytest.fixture +def cur_mock(db_mock): + cur_mock = MagicMock() + db_mock.cursor.return_value = cur_mock + return cur_mock + + +@pytest.mark.dontusefix +class Test_escribir: + def test_escribir_without_id(self, db_mock, cur_mock, mocker): + facts = [{"campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.max_id', return_value=1) + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_called_once() + + def test_escribir_with_id(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_called_once() + + def test_escribir_with_detalles(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [{"detalle1": "valor1"}, {"detalle2": "valor2"}]}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() + + def test_escribir_with_cbtes_asoc(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "cbtes_asoc": [{"cbte_asoc1": "valor1"}, {"cbte_asoc2": "valor2"}]}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'cmp_asoc': 'cmp_asoc'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'cmp_asoc': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'cmp_asoc': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() + + def test_escribir_with_permisos(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "permisos": [{"permiso1": "valor1"}, {"permiso2": "valor2"}]}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'permiso': 'permiso'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'permiso': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'permiso': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() + + def test_escribir_with_tributos(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "tributos": [{"tributo1": "valor1"}, {"tributo2": "valor2"}]}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'tributo': 'tributo'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'tributo': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'tributo': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() + + def test_escribir_with_ivas(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "ivas": [{"iva1": "valor1"}, {"iva2": "valor2"}]}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'iva': 'iva'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'iva': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'iva': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema) + + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() + + def test_escribir_without_commit(self, db_mock, cur_mock, mocker): + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) + ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + + escribir(facts, db_mock, schema, commit=False) + + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_not_called() \ No newline at end of file From 757db8b661a90ee2519d1b14879f42aae9a46b70 Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Wed, 19 Jun 2024 17:39:56 +0530 Subject: [PATCH 4/7] feat : tests added for modificar function Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 2 +- tests/test_formato_sql.py | 74 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 73 insertions(+), 3 deletions(-) diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index 848efeb2e..1f1da5cf5 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -332,7 +332,7 @@ def escribir(facts, db, schema={}, commit=True): def modificar(fact, db, schema={}, webservice="wsfev1", ids=None, conf_db={}): - from formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO + from .formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO update = [ "cae", diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index 7d9ca8df7..f8f976f86 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -19,7 +19,7 @@ import pytest from decimal import Decimal from unittest.mock import MagicMock, patch -from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, redondear, escribir +from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, redondear, escribir, modificar, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, A, N, I @pytest.mark.dontusefix @@ -804,4 +804,74 @@ def test_escribir_without_commit(self, db_mock, cur_mock, mocker): escribir(facts, db_mock, schema, commit=False) assert ejecutar_mock.call_count == 1 - db_mock.commit.assert_not_called() \ No newline at end of file + db_mock.commit.assert_not_called() + + + + + +@pytest.mark.dontusefix +@pytest.mark.parametrize("cae, fecha_vto, resultado, reproceso, motivo_obs, err_code, err_msg, cbte_nro", [ + ("12345678901234", "20230101", "A", "S", "Observación", "00", "Error mensaje", "1"), + ("NULL", None, None, None, None, None, None, "2"), + ("", None, "", "", "", "", "", "3"), +]) +def test_modificar(db_mock, cur_mock, mocker, cae, fecha_vto, resultado, reproceso, motivo_obs, err_code, err_msg, cbte_nro): + fact = { + "id": 1, + "cae": cae, + "fecha_vto": fecha_vto, + "resultado": resultado, + "reproceso": reproceso, + "motivo_obs": motivo_obs, + "err_code": err_code, + "err_msg": err_msg, + "cbte_nro": cbte_nro, + } + schema = {} + webservice = "wsfev1" + ids = None + conf_db = {"null": True} + + mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=({'encabezado': 'encabezado'}, {"encabezado": {"id": "id"}}, {})) + + modificar(fact, db_mock, schema, webservice, ids, conf_db) + + if cae == "NULL" or cae == "" or cae is None: + assert fact["cae"] == CAE_NULL + assert fact["fecha_vto"] == FECHA_VTO_NULL + + if "null" in conf_db and (resultado is None or resultado == ""): + assert fact["resultado"] == RESULTADO_NULL + + for k in ["reproceso", "motivo_obs", "err_code", "err_msg"]: + if "null" in conf_db and (k in fact and fact[k] is None or fact[k] == ""): + assert fact[k] == NULL + + cur_mock.execute.assert_called_once() + +@pytest.mark.dontusefix +def test_modificar_exception(db_mock, cur_mock, mocker): + fact = { + "id": 1, + "cae": "12345678901234", + "fecha_vto": "20230101", + "resultado": "A", + "reproceso": "S", + "motivo_obs": "Observación", + "err_code": "00", + "err_msg": "Error mensaje", + "cbte_nro": "1", + } + schema = {} + webservice = "wsfev1" + ids = None + conf_db = {"null": True} + + mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=({'encabezado': 'encabezado'}, {"encabezado": {"id": "id"}}, {})) + mocker.patch("pyafipws.formatos.formato_sql.ejecutar", side_effect=Exception("Database error")) + + with pytest.raises(Exception) as exc_info: + modificar(fact, db_mock, schema, webservice, ids, conf_db) + + assert str(exc_info.value) == "Database error" \ No newline at end of file From b5292ebf6ce8753031d82b85a78e070cdf9061c5 Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Wed, 19 Jun 2024 20:11:23 +0530 Subject: [PATCH 5/7] feat : tests added for leer function Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 71 ++++++++++---------- tests/test_formato_sql.py | 137 +++++++++++++++++++++++++++++++++++++- 2 files changed, 170 insertions(+), 38 deletions(-) diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index 1f1da5cf5..936ffe179 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -378,7 +378,7 @@ def modificar(fact, db, schema={}, webservice="wsfev1", ids=None, conf_db={}): def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): - from formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO + from .formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, DATO tablas, campos, campos_rev = configurar(schema) cur = db.cursor() @@ -405,11 +405,7 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): encabezado = {} for i, k in enumerate(description): val = row[i] - if isinstance(val, str): - val = val.decode(CHARSET) - if isinstance(val, basestring): - val = val.strip() - key = campos_rev["encabezado"].get(k[0], k[0].lower()) + key = campos_rev["encabezado"].get(k[0], k[0]) #Instead of calling lower() on k[0], directly use k[0] as the default value if the key is not found val = redondear(ENCABEZADO, key, val) encabezado[key] = val ##print encabezado @@ -418,23 +414,22 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): print( ("SELECT * FROM %(detalle)s WHERE %%(id)s = ?" % tablas) % campos["detalle"], - [encabezado["id"]], + [encabezado.get("id")], #Instead of directly accessing the "id" key, we can use encabezado.get("id") to safely retrieve the value and avoid a KeyError if the key is not present. ) ejecutar( cur, ("SELECT * FROM %(detalle)s WHERE %%(id)s = ?" % tablas) % campos["detalle"], - [encabezado["id"]], + [encabezado.get("id")], ) for it in cur.fetchall(): detalle = {} for i, k in enumerate(cur.description): - val = it[i] - if isinstance(val, str): - val = val.decode(CHARSET) - key = campos_rev["detalle"].get(k[0], k[0].lower()) - val = redondear(DETALLE, key, val) - detalle[key] = val + if i < len(it): + val = it[i] #removal of the val.decode(CHARSET) line, as it is not needed in Python 3. + key = campos_rev["detalle"].get(k[0], k[0]) + val = redondear(DETALLE, key, val) + detalle[key] = val detalles.append(detalle) encabezado["detalles"] = detalles @@ -443,20 +438,21 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): print( ("SELECT * FROM %(cmp_asoc)s WHERE %%(id)s = ?" % tablas) % campos["cmp_asoc"], - [encabezado["id"]], + [encabezado.get("id")], ) ejecutar( cur, ("SELECT * FROM %(cmp_asoc)s WHERE %%(id)s = ?" % tablas) % campos["cmp_asoc"], - [encabezado["id"]], + [encabezado.get("id")], ) for it in cur.fetchall(): cmp_asoc = {} for i, k in enumerate(cur.description): - val = it[i] - key = campos_rev["cmp_asoc"].get(k[0], k[0].lower()) - cmp_asoc[key] = val + if i < len(it): + val = it[i] + key = campos_rev["cmp_asoc"].get(k[0], k[0]) + cmp_asoc[key] = val cmps_asoc.append(cmp_asoc) if cmps_asoc: encabezado["cbtes_asoc"] = cmps_asoc @@ -466,20 +462,21 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): print( ("SELECT * FROM %(permiso)s WHERE %%(id)s = ?" % tablas) % campos["permiso"], - [encabezado["id"]], + [encabezado.get("id")], ) ejecutar( cur, ("SELECT * FROM %(permiso)s WHERE %%(id)s = ?" % tablas) % campos["permiso"], - [encabezado["id"]], + [encabezado.get("id")], ) for it in cur.fetchall(): permiso = {} for i, k in enumerate(cur.description): - val = it[i] - key = campos_rev["permiso"].get(k[0], k[0].lower()) - permiso[key] = val + if i < len(it): + val = it[i] + key = campos_rev["permiso"].get(k[0], k[0]) + permiso[key] = val permisos.append(permiso) if permisos: encabezado["permisos"] = permisos @@ -489,20 +486,21 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): print( ("SELECT * FROM %(iva)s WHERE %%(id)s = ?" % tablas) % campos["iva"], - [encabezado["id"]], + [encabezado.get("id")], ) ejecutar( cur, ("SELECT * FROM %(iva)s WHERE %%(id)s = ?" % tablas) % campos["iva"], - [encabezado["id"]], + [encabezado.get("id")], ) for it in cur.fetchall(): iva = {} for i, k in enumerate(cur.description): - val = it[i] - key = campos_rev["iva"].get(k[0], k[0].lower()) - val = redondear(IVA, key, val) - iva[key] = val + if i < len(it): + val = it[i] + key = campos_rev["iva"].get(k[0], k[0]) + val = redondear(IVA, key, val) + iva[key] = val ivas.append(iva) if ivas: encabezado["ivas"] = ivas @@ -512,21 +510,22 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): print( ("SELECT * FROM %(tributo)s WHERE %%(id)s = ?" % tablas) % campos["tributo"], - [encabezado["id"]], + [encabezado.get("id")], ) ejecutar( cur, ("SELECT * FROM %(tributo)s WHERE %%(id)s = ?" % tablas) % campos["tributo"], - [encabezado["id"]], + [encabezado.get("id")], ) for it in cur.fetchall(): tributo = {} for i, k in enumerate(cur.description): - val = it[i] - key = campos_rev["tributo"].get(k[0], k[0].lower()) - val = redondear(TRIBUTO, key, val) - tributo[key] = val + if i < len(it): + val = it[i] + key = campos_rev["tributo"].get(k[0], k[0]) + val = redondear(TRIBUTO, key, val) + tributo[key] = val tributos.append(tributo) if tributos: encabezado["tributos"] = tributos diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index f8f976f86..bee58ab62 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -19,7 +19,7 @@ import pytest from decimal import Decimal from unittest.mock import MagicMock, patch -from pyafipws.formatos.formato_sql import esquema_sql, configurar, ejecutar, max_id, redondear, escribir, modificar, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL +from pyafipws.formatos.formato_sql import esquema_sql, leer, configurar, ejecutar, max_id, redondear, escribir, modificar, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, A, N, I @pytest.mark.dontusefix @@ -874,4 +874,137 @@ def test_modificar_exception(db_mock, cur_mock, mocker): with pytest.raises(Exception) as exc_info: modificar(fact, db_mock, schema, webservice, ids, conf_db) - assert str(exc_info.value) == "Database error" \ No newline at end of file + assert str(exc_info.value) == "Database error" + + +@pytest.fixture +def db_mock(): + return MagicMock() + +@pytest.fixture +def cur_mock(db_mock): + cur_mock = MagicMock() + db_mock.cursor.return_value = cur_mock + return cur_mock + +@pytest.mark.dontusefix +@pytest.mark.parametrize("kwargs, expected_query", [ + ({}, "SELECT * FROM encabezado WHERE (resultado IS NULL OR resultado='' OR resultado=' ') AND (id IS NOT NULL) AND webservice=? ORDER BY tipo_cbte, punto_vta, cbte_nro"), + ({"campo": "valor"}, "SELECT * FROM encabezado"), +]) +def test_leer_query(db_mock, cur_mock, mocker, kwargs, expected_query): + schema = {} + webservice = "wsfev1" + ids = None + tablas = {"encabezado": "encabezado"} + campos = {"encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}} + campos_rev = {"encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}} + mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=(tablas, campos, campos_rev)) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") + mocker.patch("pyafipws.formatos.formato_sql.redondear") + + list(leer(db_mock, schema, webservice, ids, **kwargs)) + + ejecutar_mock.assert_called_once_with(cur_mock, expected_query, [webservice] if not kwargs and not ids else ids) + +@pytest.mark.dontusefix +def test_leer_fetchall(db_mock, cur_mock, mocker): + schema = {} + webservice = "wsfev1" + ids = None + tablas = {"encabezado": "encabezado", "detalle": "detalle", "cmp_asoc": "cmp_asoc", "permiso": "permiso", "iva": "iva", "tributo": "tributo"} + campos = { + "encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "iva": {"id": "id"}, + "tributo": {"id": "id"} + } + campos_rev = { + "encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + "permiso": {"id": "id"}, + "iva": {"id": "id"}, + "tributo": {"id": "id"} + } + mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=(tablas, campos, campos_rev)) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") + redondear_mock = mocker.patch("pyafipws.formatos.formato_sql.redondear", side_effect=lambda formato, clave, valor: valor) + + encabezado_data = [("1", "Resultado 1", "wsfev1", "1", "2", "3"), ("2", "Resultado 2", "wsfev1", "4", "5", "6")] + detalle_data = [("1", "Detalle 1"), ("1", "Detalle 2"), ("2", "Detalle 3")] + cmp_asoc_data = [("1",), ("2",)] + permiso_data = [("1",), ("2",)] + iva_data = [("1",), ("1",), ("2",)] + tributo_data = [("1",), ("2",)] + + cur_mock.fetchall.side_effect = [encabezado_data, detalle_data, cmp_asoc_data, permiso_data, iva_data, tributo_data, [], [], [], [], [], []] + cur_mock.description = [ + (("id",), ("resultado",), ("webservice",), ("tipo_cbte",), ("punto_vta",), ("cbte_nro",)), + (("id",),), + (("id",),), + (("id",),), + (("id",),), + (("id",),) + ] + + result = list(leer(db_mock, schema, webservice, ids)) + + + assert len(result) == 2 + assert result[0][("id",)] == "3" + assert result[0]["detalles"][0][("id",)] == "Detalle 1" + assert result[0]["detalles"][1][("id",)] == "Detalle 2" + assert result[0]["detalles"][2][("id",)] == "Detalle 3" + assert result[0]["cbtes_asoc"][0][("id",)] == "1" + assert result[0]["cbtes_asoc"][1][("id",)] == "2" + assert result[0]["permisos"][0][("id",)] == "1" + assert result[0]["permisos"][1][("id",)] == "2" + assert result[0]["ivas"][0][("id",)] == "1" + assert result[0]["ivas"][1][("id",)] == "1" + assert result[0]["ivas"][2][("id",)] == "2" + assert result[0]["tributos"][0][("id",)] == "1" + assert result[0]["tributos"][1][("id",)] == "2" + assert result[1][("id",)] == "6" + assert len(result[1]["detalles"]) == 0 + + +@pytest.mark.dontusefix +def test_leer_exception(db_mock, cur_mock, mocker): + schema = {} + webservice = "wsfev1" + ids = None + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {'encabezado': 'encabezado'}, + { + 'encabezado': { + 'id': 'id', + 'resultado': 'resultado', + 'webservice': 'webservice', + 'tipo_cbte': 'tipo_cbte', + 'punto_vta': 'punto_vta', + 'cbte_nro': 'cbte_nro' + } + }, + { + 'encabezado': { + 'id': 'id', + 'resultado': 'resultado', + 'webservice': 'webservice', + 'tipo_cbte': 'tipo_cbte', + 'punto_vta': 'punto_vta', + 'cbte_nro': 'cbte_nro' + } + } + ) + ) + mocker.patch("pyafipws.formatos.formato_sql.ejecutar", side_effect=Exception("Database error")) + + with pytest.raises(Exception) as exc_info: + list(leer(db_mock, schema, webservice, ids)) + + assert str(exc_info.value) == "Database error" From a9cf0c60e591b2d2e5509ab07dcbbfd577679e94 Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Thu, 20 Jun 2024 14:24:02 +0530 Subject: [PATCH 6/7] feat : tests added for yuda function Signed-off-by: SONIABHISHEK121 --- formatos/formato_sql.py | 18 ++++++---- tests/test_formato_sql.py | 69 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 80 insertions(+), 7 deletions(-) diff --git a/formatos/formato_sql.py b/formatos/formato_sql.py index 936ffe179..947e23604 100644 --- a/formatos/formato_sql.py +++ b/formatos/formato_sql.py @@ -536,9 +536,9 @@ def leer(db, schema={}, webservice="wsfev1", ids=None, **kwargs): cur.close() -def ayuda(): - print("-- Formato:") - from formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, DATO, PERMISO +def ayuda(file=None): + print("-- Formato:", file=file) # modified to include the file parameter, which redirects the output to the specified file-like object. + from .formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, DATO, PERMISO tipos_registro = [ ("encabezado", ENCABEZADO), @@ -549,10 +549,16 @@ def ayuda(): ("permiso", PERMISO), ("dato", DATO), ] - print("-- Esquema:") - for sql in esquema_sql(tipos_registro): - print(sql) + print("-- Esquema:", file=file) + try: + for sql in esquema_sql(tipos_registro): + print(sql, file=file) + except Exception as e: + print(f"Error al generar esquema SQL: {str(e)}", file=file) +# The file parameter is added to the function signature, +# Which allows specifying the file-like object to which the output should be redirected. +# It defaults to None, which means the output will be printed to the console. if __name__ == "__main__": ayuda() diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index bee58ab62..234bfe966 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -17,9 +17,10 @@ __license__ = "GPL 3.0" import pytest +import io from decimal import Decimal from unittest.mock import MagicMock, patch -from pyafipws.formatos.formato_sql import esquema_sql, leer, configurar, ejecutar, max_id, redondear, escribir, modificar, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL +from pyafipws.formatos.formato_sql import esquema_sql, leer, configurar, ejecutar, max_id, redondear, escribir, modificar, ayuda, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, A, N, I @pytest.mark.dontusefix @@ -1008,3 +1009,69 @@ def test_leer_exception(db_mock, cur_mock, mocker): list(leer(db_mock, schema, webservice, ids)) assert str(exc_info.value) == "Database error" + + + +@pytest.mark.dontusefix +def test_ayuda_output(capsys): + ayuda() + captured = capsys.readouterr() + + # Verify the presence of expected headers + assert "-- Formato:" in captured.out + assert "-- Esquema:" in captured.out + + # Verify the presence of expected CREATE TABLE statements + expected_tables = [ + "encabezado", + "detalle", + "tributo", + "iva", + "cmp_asoc", + "permiso", + "dato", + ] + for table in expected_tables: + assert f"CREATE TABLE {table}" in captured.out + + # Verify the presence of expected field definitions + expected_fields = [ + "tipo_reg INTEGER", + "codigo VARCHAR (30)", + "base_imp NUMERIC (15, 3)", + "iva_id INTEGER", + "cbte_tipo INTEGER", + "id_permiso VARCHAR (16)", + "campo VARCHAR (30)", + ] + for field in expected_fields: + assert field in captured.out + + +@pytest.mark.dontusefix +def test_ayuda_exception_handling(capsys): + # Test handling of KeyboardInterrupt exception + with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=KeyboardInterrupt): + try: + ayuda() + except KeyboardInterrupt: + pass + captured = capsys.readouterr() + assert "-- Formato:" in captured.out + assert "-- Esquema:" in captured.out + + # Test handling of generic exception + with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=Exception("Test Exception")): + ayuda() + captured = capsys.readouterr() + assert "-- Formato:" in captured.out + assert "-- Esquema:" in captured.out + assert "Error al generar esquema SQL: Test Exception" in captured.out + + # Test handling of generic exception + with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=Exception("Test Exception")): + ayuda() + captured = capsys.readouterr() + assert "-- Formato:" in captured.out + assert "-- Esquema:" in captured.out + From 4087a67b5c02b418facd907258afe267b80167ff Mon Sep 17 00:00:00 2001 From: SONIABHISHEK121 Date: Fri, 16 Aug 2024 12:35:43 +0530 Subject: [PATCH 7/7] ran black and flake8 to make sure the test suite is PEP8 compliance Signed-off-by: SONIABHISHEK121 --- tests/test_formato_sql.py | 654 +++++++++++++++++++++++++++++--------- 1 file changed, 495 insertions(+), 159 deletions(-) diff --git a/tests/test_formato_sql.py b/tests/test_formato_sql.py index 234bfe966..5c176c8c1 100644 --- a/tests/test_formato_sql.py +++ b/tests/test_formato_sql.py @@ -17,11 +17,35 @@ __license__ = "GPL 3.0" import pytest -import io from decimal import Decimal from unittest.mock import MagicMock, patch -from pyafipws.formatos.formato_sql import esquema_sql, leer, configurar, ejecutar, max_id, redondear, escribir, modificar, ayuda, CAE_NULL, FECHA_VTO_NULL, RESULTADO_NULL, NULL -from pyafipws.formatos.formato_txt import ENCABEZADO, DETALLE, TRIBUTO, IVA, CMP_ASOC, PERMISO, A, N, I +from pyafipws.formatos.formato_sql import ( + esquema_sql, + leer, + configurar, + ejecutar, + max_id, + redondear, + escribir, + modificar, + ayuda, + CAE_NULL, + FECHA_VTO_NULL, + RESULTADO_NULL, + NULL, +) +from pyafipws.formatos.formato_txt import ( + ENCABEZADO, + DETALLE, + TRIBUTO, + IVA, + CMP_ASOC, + PERMISO, + A, + N, + I, +) + @pytest.mark.dontusefix class TestFormatoSQL: @@ -198,7 +222,8 @@ def test_esquema_sql_permiso(self, auth): assert list(esquema_sql(tipos_registro, conf)) == expected_output def test_esquema_sql_detalle_with_conf(self, auth): - # Prueba para generar el esquema SQL de la tabla "detalle" con configuración personalizada + # Prueba para generar el esquema SQL de la tabla "detalle" + # con configuración personalizada tipos_registro = [("detalle", DETALLE)] conf = {"detalle": {"qty": "cantidad"}} expected_output = [ @@ -229,7 +254,8 @@ def test_esquema_sql_detalle_with_conf(self, auth): assert list(esquema_sql(tipos_registro, conf)) == expected_output def test_esquema_sql_tributo_with_conf(self, auth): - # Prueba para generar el esquema SQL de la tabla "tributo" con configuración personalizada + # Prueba para generar el esquema SQL de la tabla "tributo" + # con configuración personalizada tipos_registro = [("tributo", TRIBUTO)] conf = {"tributo": {"desc": "descripcion"}} expected_output = [ @@ -246,7 +272,8 @@ def test_esquema_sql_tributo_with_conf(self, auth): assert list(esquema_sql(tipos_registro, conf)) == expected_output def test_esquema_sql_iva_with_conf(self, auth): - # Prueba para generar el esquema SQL de la tabla "iva" con configuración personalizada + # Prueba para generar el esquema SQL de la tabla "iva" + # con configuración personalizada tipos_registro = [("iva", IVA)] conf = {"iva": {"importe": "imp_iva"}} expected_output = [ @@ -261,7 +288,8 @@ def test_esquema_sql_iva_with_conf(self, auth): assert list(esquema_sql(tipos_registro, conf)) == expected_output def test_esquema_sql_cmp_asoc_with_conf(self, auth): - # Prueba para generar el esquema SQL de la tabla "cmp_asoc" con configuración personalizada + # Prueba para generar el esquema SQL de la tabla "cmp_asoc" + # con configuración personalizada tipos_registro = [("cmp_asoc", CMP_ASOC)] conf = {"cmp_asoc": {"cbte_nro": "numero"}} expected_output = [ @@ -278,7 +306,8 @@ def test_esquema_sql_cmp_asoc_with_conf(self, auth): assert list(esquema_sql(tipos_registro, conf)) == expected_output def test_esquema_sql_permiso_with_conf(self, auth): - # Prueba para generar el esquema SQL de la tabla "permiso" con configuración personalizada + # Prueba para generar el esquema SQL de la tabla "permiso" + # con configuración personalizada tipos_registro = [("permiso", PERMISO)] conf = {"permiso": {"dst_merc": "destino"}} expected_output = [ @@ -290,7 +319,6 @@ def test_esquema_sql_permiso_with_conf(self, auth): ")\n;" ] assert list(esquema_sql(tipos_registro, conf)) == expected_output - @pytest.mark.dontusefix @@ -401,7 +429,6 @@ def test_configurar_with_partial_schema(self): assert campos == expected_campos assert campos_rev == expected_campos_rev - def test_configurar_with_empty_schema(self): # Test configuring with an empty schema schema = {} @@ -467,15 +494,17 @@ def test_configurar_with_none_schema(self): assert campos_rev == expected_campos_rev - @pytest.mark.dontusefix class TestEjecutar: - + def test_ejecutar_without_params(self, mocker): # Test executing a query without parameters cur = MagicMock() sql = "SELECT * FROM tabla" - mocker.patch('pyafipws.formatos.formato_sql.ejecutar', return_value=None) + mocker.patch( + "pyafipws.formatos.formato_sql.ejecutar", + return_value=None + ) ejecutar(cur, sql) cur.execute.assert_called_once_with(sql) @@ -484,7 +513,10 @@ def test_ejecutar_with_params(self, mocker): cur = MagicMock() sql = "SELECT * FROM tabla WHERE id = ?" params = (1,) - mocker.patch('pyafipws.formatos.formato_sql.ejecutar', return_value=None) + mocker.patch( + "pyafipws.formatos.formato_sql.ejecutar", + return_value=None + ) ejecutar(cur, sql, params) cur.execute.assert_called_once_with(sql, params) @@ -502,7 +534,7 @@ def test_ejecutar_prints_sql_and_params_when_debug_true(self, mocker, capsys): cur = MagicMock() sql = "SELECT * FROM tabla WHERE id = ?" params = (1,) - mocker.patch('pyafipws.formatos.formato_sql.DEBUG', True) + mocker.patch("pyafipws.formatos.formato_sql.DEBUG", True) ejecutar(cur, sql, params) captured = capsys.readouterr() assert captured.out == "SELECT * FROM tabla WHERE id = ? (1,)\n" @@ -511,29 +543,39 @@ def test_ejecutar_does_not_print_when_debug_false(self, mocker, capsys): # Test that ejecutar does not print anything when DEBUG is False cur = MagicMock() sql = "SELECT * FROM tabla" - mocker.patch('pyafipws.formatos.formato_sql.DEBUG', False) + mocker.patch("pyafipws.formatos.formato_sql.DEBUG", False) ejecutar(cur, sql) captured = capsys.readouterr() assert captured.out == "" - + + @pytest.fixture def db_mock(): return MagicMock() + @pytest.fixture def cur_mock(db_mock): cur_mock = MagicMock() db_mock.cursor.return_value = cur_mock return cur_mock + @pytest.mark.dontusefix class TestMaxId: def test_max_id_with_existing_id(self, db_mock, cur_mock, mocker): schema = {} expected_max_id = 42 cur_mock.fetchone.return_value = (expected_max_id,) - configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + configurar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado"}, + {"encabezado": {"id": "id"}}, + {}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") result = max_id(db_mock, schema) @@ -547,8 +589,15 @@ def test_max_id_with_existing_id(self, db_mock, cur_mock, mocker): def test_max_id_with_no_existing_id(self, db_mock, cur_mock, mocker): schema = {} cur_mock.fetchone.return_value = (None,) - configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + configurar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado"}, + {"encabezado": {"id": "id"}}, + {}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") result = max_id(db_mock, schema) @@ -560,11 +609,18 @@ def test_max_id_with_no_existing_id(self, db_mock, cur_mock, mocker): cur_mock.close.assert_called_once() def test_max_id_with_custom_schema(self, db_mock, cur_mock, mocker): - schema = {'encabezado': {'custom_id': 'id'}} + schema = {"encabezado": {"custom_id": "id"}} expected_max_id = 42 cur_mock.fetchone.return_value = (expected_max_id,) - configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'custom_encabezado'}, {'encabezado': {'id': 'custom_id'}}, {})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + configurar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "custom_encabezado"}, + {"encabezado": {"id": "custom_id"}}, + {}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") result = max_id(db_mock, schema) @@ -578,8 +634,15 @@ def test_max_id_with_custom_schema(self, db_mock, cur_mock, mocker): def test_max_id_with_empty_result(self, db_mock, cur_mock, mocker): schema = {} cur_mock.fetchone.return_value = None - configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + configurar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado"}, + {"encabezado": {"id": "id"}}, + {}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") result = max_id(db_mock, schema) @@ -592,20 +655,29 @@ def test_max_id_with_empty_result(self, db_mock, cur_mock, mocker): def test_max_id_with_exception(self, db_mock, cur_mock, mocker): schema = {} - configurar_mock = mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado'}, {'encabezado': {'id': 'id'}}, {})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar', side_effect=Exception('Database error')) + configurar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado"}, + {"encabezado": {"id": "id"}}, + {}, + ), + ) + ejecutar_mock = mocker.patch( + "pyafipws.formatos.formato_sql.ejecutar", + side_effect=Exception("Database error"), + ) with pytest.raises(Exception) as exc_info: max_id(db_mock, schema) - assert str(exc_info.value) == 'Database error' + assert str(exc_info.value) == "Database error" db_mock.cursor.assert_called_once() configurar_mock.assert_called_once_with(schema) ejecutar_mock.assert_called_once() cur_mock.fetchone.assert_not_called() cur_mock.close.assert_called_once() - - + @pytest.mark.dontusefix class TestRedondear: @@ -700,124 +772,288 @@ def test_redondear_with_exception(self, mocker): result = redondear(formato, clave, valor) assert result is None - print.assert_called_once_with("IMPOSIBLE REDONDEAR:", clave, valor, mocker.ANY) - - - - + print.assert_called_once_with( + "IMPOSIBLE REDONDEAR:", + clave, + valor, + mocker.ANY + ) + + @pytest.fixture def db_mock(): return MagicMock() + @pytest.fixture def cur_mock(db_mock): cur_mock = MagicMock() db_mock.cursor.return_value = cur_mock - return cur_mock - - -@pytest.mark.dontusefix -class Test_escribir: + return cur_mock + + +@pytest.mark.dontusefix +class Test_escribir: def test_escribir_without_id(self, db_mock, cur_mock, mocker): - facts = [{"campo1": "valor1", "campo2": "valor2", "detalles": []}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.max_id', return_value=1) - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [{"campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch("pyafipws.formatos.formato_sql.max_id", return_value=1) + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado", "detalle": "detalle"}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 1 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_called_once() def test_escribir_with_id(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado", "detalle": "detalle"}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 1 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_called_once() def test_escribir_with_detalles(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [{"detalle1": "valor1"}, {"detalle2": "valor2"}]}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [ + { + "id": 1, + "campo1": "valor1", + "campo2": "valor2", + "detalles": [{"detalle1": "valor1"}, {"detalle2": "valor2"}], + } + ] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado", "detalle": "detalle"}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 3 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() def test_escribir_with_cbtes_asoc(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "cbtes_asoc": [{"cbte_asoc1": "valor1"}, {"cbte_asoc2": "valor2"}]}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'cmp_asoc': 'cmp_asoc'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'cmp_asoc': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'cmp_asoc': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [ + { + "id": 1, + "campo1": "valor1", + "campo2": "valor2", + "detalles": [], + "cbtes_asoc": [{"cbte_asoc1": "valor1"}, {"cbte_asoc2": "valor2"}], + } + ] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "cmp_asoc": {"id": "id"}, + }, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 3 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() def test_escribir_with_permisos(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "permisos": [{"permiso1": "valor1"}, {"permiso2": "valor2"}]}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'permiso': 'permiso'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'permiso': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'permiso': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [ + { + "id": 1, + "campo1": "valor1", + "campo2": "valor2", + "detalles": [], + "permisos": [{"permiso1": "valor1"}, {"permiso2": "valor2"}], + } + ] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + { + "encabezado": "encabezado", + "detalle": "detalle", + "permiso": "permiso", + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "permiso": {"id": "id"}, + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "permiso": {"id": "id"}, + }, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 3 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() def test_escribir_with_tributos(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "tributos": [{"tributo1": "valor1"}, {"tributo2": "valor2"}]}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'tributo': 'tributo'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'tributo': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'tributo': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [ + { + "id": 1, + "campo1": "valor1", + "campo2": "valor2", + "detalles": [], + "tributos": [{"tributo1": "valor1"}, {"tributo2": "valor2"}], + } + ] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + { + "encabezado": "encabezado", + "detalle": "detalle", + "tributo": "tributo", + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "tributo": {"id": "id"}, + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "tributo": {"id": "id"}, + }, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 3 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() def test_escribir_with_ivas(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": [], "ivas": [{"iva1": "valor1"}, {"iva2": "valor2"}]}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle', 'iva': 'iva'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'iva': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}, 'iva': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') + facts = [ + { + "id": 1, + "campo1": "valor1", + "campo2": "valor2", + "detalles": [], + "ivas": [{"iva1": "valor1"}, {"iva2": "valor2"}], + } + ] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado", "detalle": "detalle", "iva": "iva"}, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "iva": {"id": "id"}, + }, + { + "encabezado": {"id": "id"}, + "detalle": {"id": "id"}, + "iva": {"id": "id"}, + }, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - escribir(facts, db_mock, schema) + escribir(facts, db_mock, schema) - assert ejecutar_mock.call_count == 3 - db_mock.commit.assert_called_once() + assert ejecutar_mock.call_count == 3 + db_mock.commit.assert_called_once() def test_escribir_without_commit(self, db_mock, cur_mock, mocker): - facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] - schema = {} - mocker.patch('pyafipws.formatos.formato_sql.configurar', return_value=({'encabezado': 'encabezado', 'detalle': 'detalle'}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}}, {'encabezado': {'id': 'id'}, 'detalle': {'id': 'id'}})) - ejecutar_mock = mocker.patch('pyafipws.formatos.formato_sql.ejecutar') - - escribir(facts, db_mock, schema, commit=False) - - assert ejecutar_mock.call_count == 1 - db_mock.commit.assert_not_called() - - - - - + facts = [{"id": 1, "campo1": "valor1", "campo2": "valor2", "detalles": []}] + schema = {} + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=( + {"encabezado": "encabezado", "detalle": "detalle"}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + {"encabezado": {"id": "id"}, "detalle": {"id": "id"}}, + ), + ) + ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") + + escribir(facts, db_mock, schema, commit=False) + + assert ejecutar_mock.call_count == 1 + db_mock.commit.assert_not_called() + + @pytest.mark.dontusefix -@pytest.mark.parametrize("cae, fecha_vto, resultado, reproceso, motivo_obs, err_code, err_msg, cbte_nro", [ - ("12345678901234", "20230101", "A", "S", "Observación", "00", "Error mensaje", "1"), - ("NULL", None, None, None, None, None, None, "2"), - ("", None, "", "", "", "", "", "3"), -]) -def test_modificar(db_mock, cur_mock, mocker, cae, fecha_vto, resultado, reproceso, motivo_obs, err_code, err_msg, cbte_nro): +@pytest.mark.parametrize( + "cae, fecha_vto, resultado, reproceso, motivo_obs, err_code, err_msg, cbte_nro", + [ + ( + "12345678901234", + "20230101", + "A", + "S", + "Observación", + "00", + "Error mensaje", + "1", + ), + ("NULL", None, None, None, None, None, None, "2"), + ("", None, "", "", "", "", "", "3"), + ], +) +def test_modificar( + db_mock, + cur_mock, + mocker, + cae, + fecha_vto, + resultado, + reproceso, + motivo_obs, + err_code, + err_msg, + cbte_nro, +): fact = { "id": 1, "cae": cae, @@ -834,7 +1070,10 @@ def test_modificar(db_mock, cur_mock, mocker, cae, fecha_vto, resultado, reproce ids = None conf_db = {"null": True} - mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=({'encabezado': 'encabezado'}, {"encabezado": {"id": "id"}}, {})) + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=({"encabezado": "encabezado"}, {"encabezado": {"id": "id"}}, {}), + ) modificar(fact, db_mock, schema, webservice, ids, conf_db) @@ -851,6 +1090,7 @@ def test_modificar(db_mock, cur_mock, mocker, cae, fecha_vto, resultado, reproce cur_mock.execute.assert_called_once() + @pytest.mark.dontusefix def test_modificar_exception(db_mock, cur_mock, mocker): fact = { @@ -869,91 +1109,178 @@ def test_modificar_exception(db_mock, cur_mock, mocker): ids = None conf_db = {"null": True} - mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=({'encabezado': 'encabezado'}, {"encabezado": {"id": "id"}}, {})) - mocker.patch("pyafipws.formatos.formato_sql.ejecutar", side_effect=Exception("Database error")) + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=({"encabezado": "encabezado"}, {"encabezado": {"id": "id"}}, {}), + ) + mocker.patch( + "pyafipws.formatos.formato_sql.ejecutar", + side_effect=Exception("Database error"), + ) with pytest.raises(Exception) as exc_info: modificar(fact, db_mock, schema, webservice, ids, conf_db) assert str(exc_info.value) == "Database error" - - + + @pytest.fixture def db_mock(): return MagicMock() + @pytest.fixture def cur_mock(db_mock): cur_mock = MagicMock() db_mock.cursor.return_value = cur_mock return cur_mock + @pytest.mark.dontusefix -@pytest.mark.parametrize("kwargs, expected_query", [ - ({}, "SELECT * FROM encabezado WHERE (resultado IS NULL OR resultado='' OR resultado=' ') AND (id IS NOT NULL) AND webservice=? ORDER BY tipo_cbte, punto_vta, cbte_nro"), - ({"campo": "valor"}, "SELECT * FROM encabezado"), -]) +@pytest.mark.parametrize( + "kwargs, expected_query", + [ + ( + {}, + "SELECT * FROM encabezado WHERE (resultado IS NULL OR resultado='' OR resultado=' ') AND (id IS NOT NULL) AND webservice=? ORDER BY tipo_cbte, punto_vta, cbte_nro", + ), + ({"campo": "valor"}, "SELECT * FROM encabezado"), + ], +) def test_leer_query(db_mock, cur_mock, mocker, kwargs, expected_query): schema = {} webservice = "wsfev1" ids = None tablas = {"encabezado": "encabezado"} - campos = {"encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}} - campos_rev = {"encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}} - mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=(tablas, campos, campos_rev)) + campos = { + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", + } + } + campos_rev = { + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", + } + } + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=(tablas, campos, campos_rev), + ) ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") mocker.patch("pyafipws.formatos.formato_sql.redondear") list(leer(db_mock, schema, webservice, ids, **kwargs)) - ejecutar_mock.assert_called_once_with(cur_mock, expected_query, [webservice] if not kwargs and not ids else ids) + ejecutar_mock.assert_called_once_with( + cur_mock, expected_query, [webservice] if not kwargs and not ids else ids + ) + @pytest.mark.dontusefix def test_leer_fetchall(db_mock, cur_mock, mocker): schema = {} webservice = "wsfev1" ids = None - tablas = {"encabezado": "encabezado", "detalle": "detalle", "cmp_asoc": "cmp_asoc", "permiso": "permiso", "iva": "iva", "tributo": "tributo"} + tablas = { + "encabezado": "encabezado", + "detalle": "detalle", + "cmp_asoc": "cmp_asoc", + "permiso": "permiso", + "iva": "iva", + "tributo": "tributo", + } campos = { - "encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}, + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", + }, "detalle": {"id": "id"}, "cmp_asoc": {"id": "id"}, "permiso": {"id": "id"}, "iva": {"id": "id"}, - "tributo": {"id": "id"} + "tributo": {"id": "id"}, } campos_rev = { - "encabezado": {"id": "id", "resultado": "resultado", "webservice": "webservice", "tipo_cbte": "tipo_cbte", "punto_vta": "punto_vta", "cbte_nro": "cbte_nro"}, + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", + }, "detalle": {"id": "id"}, "cmp_asoc": {"id": "id"}, "permiso": {"id": "id"}, "iva": {"id": "id"}, - "tributo": {"id": "id"} + "tributo": {"id": "id"}, } - mocker.patch("pyafipws.formatos.formato_sql.configurar", return_value=(tablas, campos, campos_rev)) + mocker.patch( + "pyafipws.formatos.formato_sql.configurar", + return_value=(tablas, campos, campos_rev), + ) ejecutar_mock = mocker.patch("pyafipws.formatos.formato_sql.ejecutar") - redondear_mock = mocker.patch("pyafipws.formatos.formato_sql.redondear", side_effect=lambda formato, clave, valor: valor) + redondear_mock = mocker.patch( + "pyafipws.formatos.formato_sql.redondear", + side_effect=lambda formato, clave, valor: valor, + ) - encabezado_data = [("1", "Resultado 1", "wsfev1", "1", "2", "3"), ("2", "Resultado 2", "wsfev1", "4", "5", "6")] + encabezado_data = [ + ("1", "Resultado 1", "wsfev1", "1", "2", "3"), + ("2", "Resultado 2", "wsfev1", "4", "5", "6"), + ] detalle_data = [("1", "Detalle 1"), ("1", "Detalle 2"), ("2", "Detalle 3")] cmp_asoc_data = [("1",), ("2",)] permiso_data = [("1",), ("2",)] iva_data = [("1",), ("1",), ("2",)] tributo_data = [("1",), ("2",)] - cur_mock.fetchall.side_effect = [encabezado_data, detalle_data, cmp_asoc_data, permiso_data, iva_data, tributo_data, [], [], [], [], [], []] + cur_mock.fetchall.side_effect = [ + encabezado_data, + detalle_data, + cmp_asoc_data, + permiso_data, + iva_data, + tributo_data, + [], + [], + [], + [], + [], + [], + ] cur_mock.description = [ - (("id",), ("resultado",), ("webservice",), ("tipo_cbte",), ("punto_vta",), ("cbte_nro",)), + ( + ("id",), + ("resultado",), + ("webservice",), + ("tipo_cbte",), + ("punto_vta",), + ("cbte_nro",), + ), + (("id",),), (("id",),), (("id",),), (("id",),), (("id",),), - (("id",),) ] result = list(leer(db_mock, schema, webservice, ids)) - assert len(result) == 2 assert result[0][("id",)] == "3" assert result[0]["detalles"][0][("id",)] == "Detalle 1" @@ -980,30 +1307,33 @@ def test_leer_exception(db_mock, cur_mock, mocker): mocker.patch( "pyafipws.formatos.formato_sql.configurar", return_value=( - {'encabezado': 'encabezado'}, + {"encabezado": "encabezado"}, { - 'encabezado': { - 'id': 'id', - 'resultado': 'resultado', - 'webservice': 'webservice', - 'tipo_cbte': 'tipo_cbte', - 'punto_vta': 'punto_vta', - 'cbte_nro': 'cbte_nro' + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", } }, { - 'encabezado': { - 'id': 'id', - 'resultado': 'resultado', - 'webservice': 'webservice', - 'tipo_cbte': 'tipo_cbte', - 'punto_vta': 'punto_vta', - 'cbte_nro': 'cbte_nro' + "encabezado": { + "id": "id", + "resultado": "resultado", + "webservice": "webservice", + "tipo_cbte": "tipo_cbte", + "punto_vta": "punto_vta", + "cbte_nro": "cbte_nro", } - } - ) + }, + ), + ) + mocker.patch( + "pyafipws.formatos.formato_sql.ejecutar", + side_effect=Exception("Database error"), ) - mocker.patch("pyafipws.formatos.formato_sql.ejecutar", side_effect=Exception("Database error")) with pytest.raises(Exception) as exc_info: list(leer(db_mock, schema, webservice, ids)) @@ -1011,7 +1341,6 @@ def test_leer_exception(db_mock, cur_mock, mocker): assert str(exc_info.value) == "Database error" - @pytest.mark.dontusefix def test_ayuda_output(capsys): ayuda() @@ -1051,7 +1380,9 @@ def test_ayuda_output(capsys): @pytest.mark.dontusefix def test_ayuda_exception_handling(capsys): # Test handling of KeyboardInterrupt exception - with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=KeyboardInterrupt): + with patch( + "pyafipws.formatos.formato_sql.esquema_sql", side_effect=KeyboardInterrupt + ): try: ayuda() except KeyboardInterrupt: @@ -1061,7 +1392,10 @@ def test_ayuda_exception_handling(capsys): assert "-- Esquema:" in captured.out # Test handling of generic exception - with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=Exception("Test Exception")): + with patch( + "pyafipws.formatos.formato_sql.esquema_sql", + side_effect=Exception("Test Exception"), + ): ayuda() captured = capsys.readouterr() assert "-- Formato:" in captured.out @@ -1069,9 +1403,11 @@ def test_ayuda_exception_handling(capsys): assert "Error al generar esquema SQL: Test Exception" in captured.out # Test handling of generic exception - with patch("pyafipws.formatos.formato_sql.esquema_sql", side_effect=Exception("Test Exception")): + with patch( + "pyafipws.formatos.formato_sql.esquema_sql", + side_effect=Exception("Test Exception"), + ): ayuda() captured = capsys.readouterr() assert "-- Formato:" in captured.out assert "-- Esquema:" in captured.out -