diff --git a/formatos/formato_csv.py b/formatos/formato_csv.py index b6a1ef809..f1d50a12f 100644 --- a/formatos/formato_csv.py +++ b/formatos/formato_csv.py @@ -24,43 +24,39 @@ import csv from decimal import Decimal import os +from openpyxl import load_workbook - -def leer(fn="entrada.csv", delimiter=";"): - "Analiza un archivo CSV y devuelve un diccionario (aplanado)" +def leer(fn="entrada.csv", delimiter=";", header=True): + "Analiza un archivo CSV y devuelve una lista de listas con los datos" ext = os.path.splitext(fn)[1].lower() items = [] + if ext == ".csv": - csvfile = open(fn, "rb") - # deducir dialecto y delimitador - try: - dialect = csv.Sniffer().sniff(csvfile.read(256), delimiters=[";", ","]) - except csv.Error: - dialect = csv.excel - dialect.delimiter = delimiter - csvfile.seek(0) - csv_reader = csv.reader(csvfile, dialect) - for row in csv_reader: - r = [] - for c in row: - if isinstance(c, basestring): - c = c.strip() - r.append(c) - items.append(r) - elif ext == ".xlsx": - # extraigo los datos de la planilla Excel - from openpyxl import load_workbook + with open(fn, "r", encoding="utf-8") as csvfile: + try: + dialect = csv.Sniffer().sniff(csvfile.read(256), delimiters=[delimiter, ","]) + except csv.Error: + dialect = csv.excel + dialect.delimiter = delimiter + csvfile.seek(0) + csv_reader = csv.reader(csvfile, dialect) + if header: + next(csv_reader, None) # Skip the header row + for row in csv_reader: + items.append([c.strip() if isinstance(c, str) else c for c in row]) + elif ext == ".xlsx": wb = load_workbook(filename=fn) - ws1 = wb.get_active_sheet() - for row in ws1.rows: - fila = [] - for cell in row: - fila.append(cell.value) - items.append(fila) + ws1 = wb.active + rows = ws1.iter_rows(values_only=True) + if header: + next(rows, None) # Skip the header row + for row in rows: + items.append([cell for cell in row]) + return items # TODO: return desaplanar(items) - + def aplanar(regs): "Convierte una estructura python en planilla CSV (PyRece)" @@ -87,64 +83,63 @@ def aplanar(regs): if reg.get("cbte_nro"): fila["cbt_numero"] = reg["cbte_nro"] - for i, det in enumerate(reg["detalles"]): - li = i + 1 + for i, det in enumerate(reg.get("detalles", []), start=1): fila.update( { - "codigo%s" % li: det.get("codigo", ""), - "descripcion%s" % li: det.get("ds", ""), - "umed%s" % li: det.get("umed"), - "cantidad%s" % li: det.get("qty"), - "precio%s" % li: det.get("precio"), - "importe%s" % li: det.get("importe"), - "iva_id%s" % li: det.get("iva_id"), - "imp_iva%s" % li: det.get("imp_iva"), - "bonif%s" % li: det.get("bonif"), - "numero_despacho%s" % li: det.get("despacho"), - "dato_a%s" % li: det.get("dato_a"), - "dato_b%s" % li: det.get("dato_b"), - "dato_c%s" % li: det.get("dato_c"), - "dato_d%s" % li: det.get("dato_d"), - "dato_e%s" % li: det.get("dato_e"), + "codigo%s" % i: det.get("codigo", ""), + "descripcion%s" % i: det.get("ds", ""), + "umed%s" % i: det.get("umed"), + "cantidad%s" % i: det.get("qty"), + "precio%s" % i: det.get("precio"), + "importe%s" % i: det.get("importe"), + "iva_id%s" % i: det.get("iva_id"), + "imp_iva%s" % i: det.get("imp_iva"), + "bonif%s" % i: det.get("bonif"), + "numero_despacho%s" % i: det.get("despacho"), + "dato_a%s" % i: det.get("dato_a"), + "dato_b%s" % i: det.get("dato_b"), + "dato_c%s" % i: det.get("dato_c"), + "dato_d%s" % i: det.get("dato_d"), + "dato_e%s" % i: det.get("dato_e"), } ) - for i, iva in enumerate(reg["ivas"]): - li = i + 1 + + for i, iva in enumerate(reg.get("ivas", []), start=1): fila.update( { - "iva_id_%s" % li: iva["iva_id"], - "iva_base_imp_%s" % li: iva["base_imp"], - "iva_importe_%s" % li: iva["importe"], + "iva_id_%s" % i: iva.get("iva_id"), + "iva_base_imp_%s" % i: iva.get("base_imp"), + "iva_importe_%s" % i: iva.get("importe"), } ) - for i, tributo in enumerate(reg["tributos"]): - li = i + 1 + + for i, tributo in enumerate(reg.get("tributos", []), start=1): fila.update( { - "tributo_id_%s" % li: tributo["tributo_id"], - "tributo_base_imp_%s" % li: tributo["base_imp"], - "tributo_desc_%s" % li: tributo["desc"], - "tributo_alic_%s" % li: tributo["alic"], - "tributo_importe_%s" % li: tributo["importe"], + "tributo_id_%s" % i: tributo.get("tributo_id"), + "tributo_base_imp_%s" % i: tributo.get("base_imp"), + "tributo_desc_%s" % i: tributo.get("desc"), + "tributo_alic_%s" % i: tributo.get("alic"), + "tributo_importe_%s" % i: tributo.get("importe"), } ) - for i, opcional in enumerate(reg.get("opcionales", [])): - li = i + 1 + + for i, opcional in enumerate(reg.get("opcionales", []), start=1): fila.update( { - "opcional_id_%s" % li: opcional["opcional_id"], - "opcional_valor_%s" % li: opcional["valor"], + "opcional_id_%s" % i: opcional.get("opcional_id"), + "opcional_valor_%s" % i: opcional.get("valor"), } ) - for i, cbte_asoc in enumerate(reg.get("cbtes_asoc", [])): - li = i + 1 + + for i, cbte_asoc in enumerate(reg.get("cbtes_asoc", []), start=1): fila.update( { - "cbte_asoc_tipo_%s" % li: cbte_asoc["cbte_tipo"], - "cbte_asoc_pto_vta_%s" % li: cbte_asoc["cbte_punto_vta"], - "cbte_asoc_nro_%s" % li: cbte_asoc["cbte_nro"], - "cbte_asoc_cuit_%s" % li: cbte_asoc["cbte_cuit"], - "cbte_asoc_fecha_%s" % li: cbte_asoc["cbte_fecha"], + "cbte_asoc_tipo_%s" % i: cbte_asoc.get("cbte_tipo"), + "cbte_asoc_pto_vta_%s" % i: cbte_asoc.get("cbte_punto_vta"), + "cbte_asoc_nro_%s" % i: cbte_asoc.get("cbte_nro"), + "cbte_asoc_cuit_%s" % i: cbte_asoc.get("cbte_cuit"), + "cbte_asoc_fecha_%s" % i: cbte_asoc.get("cbte_fecha"), } ) @@ -188,12 +183,14 @@ def aplanar(regs): "numero_remito", "obs_generales", "obs_comerciales", + "forma_pago", + "pdf", ] # filtro y ordeno las columnas l = [k for f in filas for k in list(f.keys())] s = set(l) - set(cols) - cols = cols + list(s) + cols.extend(sorted(s)) ret = [cols] for fila in filas: @@ -202,6 +199,8 @@ def aplanar(regs): return ret + + def desaplanar(filas): "Dado una planilla, conviertir en estructura python" @@ -339,22 +338,16 @@ def escribir(filas, fn="salida.csv", delimiter=";"): "Dado una lista de comprobantes (diccionarios), aplana y escribe" ext = os.path.splitext(fn)[1].lower() if ext == ".csv": - f = open(fn, "wb") - csv_writer = csv.writer(f, dialect="excel", delimiter=";") - # TODO: filas = aplanar(regs) - for fila in filas: - # convertir a ISO-8859-1 (evita error de encoding de csv writer): - fila = [ - celda.encode("latin1") if isinstance(celda, str) else celda - for celda in fila - ] - csv_writer.writerow(fila) - f.close() + with open(fn, "w", newline="") as f: + csv_writer = csv.writer(f, dialect="excel", delimiter=delimiter) + # TODO: filas = aplanar(regs) + for fila in filas: + csv_writer.writerow(fila) elif ext == ".xlsx": from openpyxl import Workbook wb = Workbook() - ws1 = wb.get_active_sheet() + ws1 = wb.active for fila in filas: ws1.append(fila) wb.save(filename=fn) @@ -375,4 +368,4 @@ def escribir(filas, fn="salida.csv", delimiter=";"): for fila1, fila2 in zip(filas1, filas2): for celda1, celda2 in zip(fila1, fila2): if celda1 != celda2: - print(celda1, celda2) + print(celda1, celda2) \ No newline at end of file diff --git a/requirements-dev.txt b/requirements-dev.txt index 8ca42cd32..7729ac8d6 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -11,4 +11,5 @@ pytest-freezegun==0.4.2; python_version > '3' pytest-mock==2.0.0; python_version <= '2.7' pytest-mock==3.10.0; python_version > '2.7' pywin32==304; sys_platform == "win32" and python_version > '3' -py2exe==0.11.1.1; sys_platform == "win32" and python_version > '3' \ No newline at end of file +py2exe==0.11.1.1; sys_platform == "win32" and python_version > '3' +openpyxl==3.1.2; python_version > '3' \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 59cb59a04..ea6b63188 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,4 @@ tabulate==0.8.5 certifi>=2020.4.5.1 qrcode==6.1 future==0.18.3 +openpyxl==3.1.2; python_version > '3' diff --git a/tests/conftest.py b/tests/conftest.py index bb1283ba5..2431c86b0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,6 +11,7 @@ # for more details. import os +import json import pytest from pyafipws.wsaa import WSAA diff --git a/tests/test_formato_csv.py b/tests/test_formato_csv.py new file mode 100644 index 000000000..72d3f2ef9 --- /dev/null +++ b/tests/test_formato_csv.py @@ -0,0 +1,1499 @@ +#!/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_csv""" + +__author__ = "Mariano Reingart " +__copyright__ = "Copyright (C) 2010-2019 Mariano Reingart" +__license__ = "GPL 3.0" + +import os +import sys +import csv +import pytest +import tempfile +from pyafipws.formatos.formato_csv import leer, aplanar, desaplanar, escribir +from openpyxl import Workbook, load_workbook +from io import StringIO +from unittest.mock import patch + +# Add the 'formatos' directory to the Python module search path +formatos_dir = os.path.abspath( + os.path.join(os.path.dirname(__file__), "..", "formatos") +) +sys.path.insert(0, formatos_dir) + + +@pytest.mark.dontusefix +class TestLeerFunction: + def test_leer_csv_file(self): + """ + Test that the leer function can read a valid CSV file correctly. + """ + expected_data = [ + [ + "1", + "6", + "4004", + "526", + "20170826", + "80", + "30500010912", + "PES", + "1.000000", + "889.82", + "186.86", + "8.89", + "0.00", + "0.00", + "1085.57", + "1", + "", + "", + "", + "61233038185853", + "20110619", + "A", + "", + "S", + "", + "", + "", + "", + "", + "mariano@sistemasagiles.com.ar", + "21601192", + "6443", + "Exento", + "82016336", + "8001", + "", + "", + "P1675G", + "COD2", + "8.89", + "1076.68", + "0", + "1.0", + "0", + "0", + "5", + "1.00", + "20205766", + "110170P", + "Impuesto municipal matanza", + "889.82", + "186.86", + "7", + "0", + "PRUEBA ART", + "SEGUNDO ART", + "", + "", + "Cliente XXX", + "", + "99", + "", + "889.82", + "Patricia 1 - Cdad de Buenos Aires - 1405 - Capital Federal - " + "Argentina", + "1076.68", + "0", + "30 Dias", + "0.00", + "1", + "17", + "1", + "1801", + "30500010912", + "1802", + "BNA", + ] + ] + result = leer("datos/facturas.csv", delimiter=";") + assert result == expected_data + + def test_leer_csv_custom_delimiter(self): + """ + Test that the leer function can read a CSV file with + a custom delimiter. + """ + sample_csv_data = ( + "Column1|Column2|Column3\n" + "Value1|Value2|Value3\n" + "Value4|Value5|Value6" + ) + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv", delimiter="|") + assert result == expected_data + + def test_leer_csv_with_header(self): + """ + Test that the leer function can read a CSV file + with a header row correctly. + """ + sample_csv_data = ( + "Column1,Column2,Column3\n" + "Value1,Value2,Value3\n" + "Value4,Value5,Value6" + ) + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv") + assert result == expected_data + + def test_leer_csv_without_header(self): + """ + Test that the leer function can read a CSV file + without a header row correctly. + """ + sample_csv_data = ( + "Value1,Value2,Value3\n" + "Value4,Value5,Value6" + ) + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv", header=False) + assert result == expected_data + + def test_leer_csv_with_whitespace(self): + """ + Test that the leer function can handle leading/trailing whitespace + in CSV values correctly. + """ + sample_csv_data = ( + "Column1,Column2,Column3\n" + "Value1 , Value2 , Value3 \n" + "Value4 , Value5 , Value6 " + ) + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv") + assert result == expected_data + + def test_leer_csv_with_non_string_values(self): + """ + Test that the leer function can handle non-string + values in a CSV file correctly. + """ + sample_csv_data = ( + "Column1,Column2,Column3\n" + "1,2.5,True\n" + "4,5.7,False" + ) + expected_data = [ + ["1", "2.5", "True"], + ["4", "5.7", "False"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv") + assert result == expected_data + + def test_leer_csv_different_dialect(self): + """ + Test that the leer function can handle a CSV file + with a different dialect correctly. + """ + sample_csv_data = ( + "Column1;Column2;Column3\n" + "Value1;Value2;Value3\n" + "Value4;Value5;Value6" + ) + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv", delimiter=";") + assert result == expected_data + + def test_leer_csv_empty_file(self): + """ + Test that the leer function handles an empty CSV file correctly. + """ + sample_csv_data = "" + expected_data = [] + + with patch("builtins.open", return_value=StringIO(sample_csv_data)): + result = leer("data/sample.csv") + assert result == expected_data + + def test_leer_xlsx_file_with_header(self): + """ + Test that the leer function can read an Excel file + with a header row correctly. + """ + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + # Create a temporary Excel file for testing + with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as temp_file: + workbook = Workbook() + worksheet = workbook.active + worksheet["A1"] = "Column1" + worksheet["B1"] = "Column2" + worksheet["C1"] = "Column3" + worksheet["A2"] = "Value1" + worksheet["B2"] = "Value2" + worksheet["C2"] = "Value3" + worksheet["A3"] = "Value4" + worksheet["B3"] = "Value5" + worksheet["C3"] = "Value6" + workbook.save(temp_file.name) + + result = leer(temp_file.name) + assert result == expected_data + + # Clean up the temporary file + os.unlink(temp_file.name) + + def test_leer_xlsx_file_without_header(self): + """ + Test that the leer function can read an Excel file + without a header row correctly. + """ + expected_data = [ + ["Value1", "Value2", "Value3"], + ["Value4", "Value5", "Value6"] + ] + + # Create a temporary Excel file for testing + with tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False) as temp_file: + workbook = Workbook() + worksheet = workbook.active + worksheet["A1"] = "Value1" + worksheet["B1"] = "Value2" + worksheet["C1"] = "Value3" + worksheet["A2"] = "Value4" + worksheet["B2"] = "Value5" + worksheet["C2"] = "Value6" + workbook.save(temp_file.name) + + result = leer(temp_file.name, header=False) + assert result == expected_data + + # Clean up the temporary file + os.unlink(temp_file.name) + + +@pytest.mark.dontusefix +class TestAplanarFunction: + def test_aplanar_single_record(self): + """ + Test that the aplanar function correctly flattens a single record. + """ + reg = { + "id": 1, + "tipo_cbte": 1, + "punto_vta": 1, + "cbte_nro": 1, + "fecha_cbte": "2023-06-08", + "tipo_doc": 80, + "nro_doc": "20123456789", + "imp_total": 126, + "detalles": [ + { + "codigo": "P1", + "ds": "Producto 1", + "qty": 2, + "umed": 7, + "precio": 50, + "importe": 100, + } + ], + "ivas": [ + { + "iva_id": 5, + "base_imp": 100, + "importe": 21, + } + ], + "tributos": [ + { + "tributo_id": 1, + "base_imp": 100, + "desc": "Tributo 1", + "alic": 5, + "importe": 5, + } + ], + "forma_pago": "Contado", + } + + expected_data = [ + [ + "id", + "tipo_cbte", + "punto_vta", + "cbt_numero", + "fecha_cbte", + "tipo_doc", + "nro_doc", + "moneda_id", + "moneda_ctz", + "imp_neto", + "imp_iva", + "imp_trib", + "imp_op_ex", + "imp_tot_conc", + "imp_total", + "concepto", + "fecha_venc_pago", + "fecha_serv_desde", + "fecha_serv_hasta", + "cae", + "fecha_vto", + "resultado", + "motivo", + "reproceso", + "nombre", + "domicilio", + "localidad", + "telefono", + "categoria", + "email", + "numero_cliente", + "numero_orden_compra", + "condicion_frente_iva", + "numero_cotizacion", + "numero_remito", + "obs_generales", + "obs_comerciales", + "forma_pago", + "pdf", + "bonif1", + "cantidad1", + "cbte_nro", + "codigo1", + "cuit", + "dato_a1", + "dato_b1", + "dato_c1", + "dato_d1", + "dato_e1", + "descripcion1", + "detalles", + "domicilio_cliente", + "id_impositivo", + "idioma", + "idioma_cbte", + "imp_iva1", + "importe1", + "incoterms", + "incoterms_ds", + "iva_base_imp_1", + "iva_id1", + "iva_id_1", + "iva_importe_1", + "ivas", + "localidad_cliente", + "nombre_cliente", + "numero_despacho1", + "pais_dst_cmp", + "permiso_existente", + "precio1", + "provincia_cliente", + "telefono_cliente", + "tipo_expo", + "tributo_alic_1", + "tributo_base_imp_1", + "tributo_desc_1", + "tributo_id_1", + "tributo_importe_1", + "tributos", + "umed1", + ], + [ + 1, + 1, + 1, + 1, + "2023-06-08", + 80, + "20123456789", + None, + None, + None, + None, + None, + None, + None, + 126, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + "Contado", + "", + None, + 2, + 1, + "P1", + None, + None, + None, + None, + None, + None, + "Producto 1", + [ + { + "codigo": "P1", + "ds": "Producto 1", + "qty": 2, + "umed": 7, + "precio": 50, + "importe": 100, + } + ], + None, + None, + None, + None, + None, + 100, + None, + None, + 100, + None, + 5, + 21, + [{"iva_id": 5, "base_imp": 100, "importe": 21}], + None, + None, + None, + None, + None, + 50, + None, + None, + None, + 5, + 100, + "Tributo 1", + 1, + 5, + [ + { + "tributo_id": 1, + "base_imp": 100, + "desc": "Tributo 1", + "alic": 5, + "importe": 5, + } + ], + 7, + ], + ] + + result = aplanar([reg]) + assert result == expected_data + + def test_aplanar_multiple_records(self): + """ + Test that the aplanar function correctly flattens multiple records. + """ + regs = [ + { + "id": 1, + "tipo_cbte": 1, + "punto_vta": 1, + "cbte_nro": 1, + "fecha_cbte": "2023-06-08", + "tipo_doc": 80, + "nro_doc": "20123456789", + "imp_total": 126, + "detalles": [ + { + "codigo": "P1", + "ds": "Producto 1", + "qty": 2, + "umed": 7, + "precio": 50, + "importe": 100, + } + ], + "ivas": [ + { + "iva_id": 5, + "base_imp": 100, + "importe": 21, + } + ], + "tributos": [ + { + "tributo_id": 1, + "base_imp": 100, + "desc": "Tributo 1", + "alic": 5, + "importe": 5, + } + ], + "forma_pago": "Contado", + }, + { + "id": 2, + "tipo_cbte": 6, + "punto_vta": 1, + "cbte_nro": 2, + "fecha_cbte": "2023-06-09", + "tipo_doc": 80, + "nro_doc": "20987654321", + "imp_total": 200, + "detalles": [ + { + "codigo": "P2", + "ds": "Producto 2", + "qty": 1, + "umed": 7, + "precio": 200, + "importe": 200, + } + ], + "ivas": [ + { + "iva_id": 5, + "base_imp": 200, + "importe": 42, + } + ], + "tributos": [], + "forma_pago": "Tarjeta de Crédito", + }, + ] + + expected_data = [ + [ + "id", + "tipo_cbte", + "punto_vta", + "cbt_numero", + "fecha_cbte", + "tipo_doc", + "nro_doc", + "moneda_id", + "moneda_ctz", + "imp_neto", + "imp_iva", + "imp_trib", + "imp_op_ex", + "imp_tot_conc", + "imp_total", + "concepto", + "fecha_venc_pago", + "fecha_serv_desde", + "fecha_serv_hasta", + "cae", + "fecha_vto", + "resultado", + "motivo", + "reproceso", + "nombre", + "domicilio", + "localidad", + "telefono", + "categoria", + "email", + "numero_cliente", + "numero_orden_compra", + "condicion_frente_iva", + "numero_cotizacion", + "numero_remito", + "obs_generales", + "obs_comerciales", + "forma_pago", + "pdf", + "bonif1", + "cantidad1", + "cbte_nro", + "codigo1", + "cuit", + "dato_a1", + "dato_b1", + "dato_c1", + "dato_d1", + "dato_e1", + "descripcion1", + "detalles", + "domicilio_cliente", + "id_impositivo", + "idioma", + "idioma_cbte", + "imp_iva1", + "importe1", + "incoterms", + "incoterms_ds", + "iva_base_imp_1", + "iva_id1", + "iva_id_1", + "iva_importe_1", + "ivas", + "localidad_cliente", + "nombre_cliente", + "numero_despacho1", + "pais_dst_cmp", + "permiso_existente", + "precio1", + "provincia_cliente", + "telefono_cliente", + "tipo_expo", + "tributo_alic_1", + "tributo_base_imp_1", + "tributo_desc_1", + "tributo_id_1", + "tributo_importe_1", + "tributos", + "umed1", + ], + [ + 1, + 1, + 1, + 1, + "2023-06-08", + 80, + "20123456789", + None, + None, + None, + None, + None, + None, + None, + 126, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + "Contado", + "", + None, + 2, + 1, + "P1", + None, + None, + None, + None, + None, + None, + "Producto 1", + [ + { + "codigo": "P1", + "ds": "Producto 1", + "qty": 2, + "umed": 7, + "precio": 50, + "importe": 100, + } + ], + None, + None, + None, + None, + None, + 100, + None, + None, + 100, + None, + 5, + 21, + [{"iva_id": 5, "base_imp": 100, "importe": 21}], + None, + None, + None, + None, + None, + 50, + None, + None, + None, + 5, + 100, + "Tributo 1", + 1, + 5, + [ + { + "tributo_id": 1, + "base_imp": 100, + "desc": "Tributo 1", + "alic": 5, + "importe": 5, + } + ], + 7, + ], + [ + 2, + 6, + 1, + 2, + "2023-06-09", + 80, + "20987654321", + None, + None, + None, + None, + None, + None, + None, + 200, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + None, + "Tarjeta de Crédito", + "", + None, + 1, + 2, + "P2", + None, + None, + None, + None, + None, + None, + "Producto 2", + [ + { + "codigo": "P2", + "ds": "Producto 2", + "qty": 1, + "umed": 7, + "precio": 200, + "importe": 200, + } + ], + None, + None, + None, + None, + None, + 200, + None, + None, + 200, + None, + 5, + 42, + [{"iva_id": 5, "base_imp": 200, "importe": 42}], + None, + None, + None, + None, + None, + 200, + None, + None, + None, + None, + None, + None, + None, + None, + [], + 7, + ], + ] + + result = aplanar(regs) + assert result == expected_data + + +@pytest.mark.dontusefix +class TestDesplanarFunction: + def test_desaplanar_single_record(self): + """ + Test that the desaplanar function correctly converts + a single record from flattened format to structured format. + """ + filas = [ + [ + "id", + "tipo_cbte", + "punto_vta", + "cbt_numero", + "fecha_cbte", + "tipo_doc", + "nro_doc", + "moneda_id", + "moneda_ctz", + "imp_neto", + "imp_iva", + "imp_trib", + "imp_op_ex", + "imp_tot_conc", + "imp_total", + "concepto", + "fecha_venc_pago", + "fecha_serv_desde", + "fecha_serv_hasta", + "cae", + "fecha_vto", + "resultado", + "motivo", + "reproceso", + "nombre", + "domicilio", + "localidad", + "telefono", + "categoria", + "email", + "numero_cliente", + "numero_orden_compra", + "condicion_frente_iva", + "numero_cotizacion", + "numero_remito", + "obs_generales", + "obs_comerciales", + "forma_pago", + "pdf", + "codigo1", + "descripcion1", + "cantidad1", + "umed1", + "precio1", + "importe1", + "bonif1", + "iva_id1", + "imp_iva1", + "tributo_id_1", + "tributo_desc_1", + "tributo_base_imp_1", + "tributo_alic_1", + "tributo_importe_1", + "iva_id_1", + "iva_base_imp_1", + "iva_importe_1", + ], + [ + 1, + 1, + 1, + 1, + "2023-06-08", + 80, + "20123456789", + "PES", + 1, + 100, + 21, + 5, + 0, + 0, + 126, + 1, + "2023-06-08", + "2023-06-08", + "2023-06-08", + "1234567890", + "2023-06-18", + "A", + "", + "", + "John Doe", + "123 Main St", + "City", + "1234567890", + "A", + "john@example.com", + "ABC123", + "OC123", + "Responsable Inscripto", + "COT123", + "REM123", + "Observaciones generales", + "Observaciones comerciales", + "Contado", + "", + "P1", + "Producto 1", + 2, + 7, + 50, + 100, + 0, + 5, + 21, + 1, + "Tributo 1", + 100, + 5, + 5, + 5, + 100, + 21, + ], + ] + + expected_data = [ + { + "cbte_nro": 1, + "tipo_cbte": 1, + "punto_vta": 1, + "cbt_numero": 1, + "fecha_cbte": "2023-06-08", + "concepto": 1, + "moneda_id": "PES", + "moneda_ctz": 1, + "tipo_doc": 80, + "nro_doc": "20123456789", + "email": "john@example.com", + "numero_cliente": "ABC123", + "numero_orden_compra": "OC123", + "condicion_frente_iva": "Responsable Inscripto", + "numero_cotizacion": "COT123", + "numero_remito": "REM123", + "imp_total": 126, + "imp_tot_conc": 0, + "imp_neto": 100, + "imp_iva": 21, + "imp_trib": 5, + "imp_op_ex": 0, + "fecha_serv_desde": "2023-06-08", + "fecha_serv_hasta": "2023-06-08", + "fecha_venc_pago": "2023-06-08", + "obs_generales": "Observaciones generales", + "obs_comerciales": "Observaciones comerciales", + "resultado": "A", + "cae": "1234567890", + "fecha_vto": "2023-06-18", + "reproceso": "", + "motivo": "", + "id": 1, + "detalles": [ + { + "codigo": "P1", + "ds": "Producto 1", + "umed": 7, + "qty": 2, + "precio": 50, + "importe": 100, + "iva_id": 5, + "imp_iva": 21, + "bonif": None, + "despacho": False, + "dato_a": False, + "dato_b": False, + "dato_c": False, + "dato_d": False, + "dato_e": False, + } + ], + "tributos": [ + { + "tributo_id": 1, + "desc": "Tributo 1", + "base_imp": 100, + "alic": 5, + "importe": 5, + } + ], + "ivas": [{"iva_id": 5, "base_imp": 100, "importe": 21}], + "permisos": [], + "opcionales": [], + "cbtes_asoc": [], + "forma_pago": "Contado", + "datos": [ + {"campo": "nombre", "valor": "John Doe", "pagina": ""}, + {"campo": "domicilio", "valor": "123 Main St", "pagina": ""}, + {"campo": "localidad", "valor": "City", "pagina": ""}, + {"campo": "telefono", "valor": "1234567890", "pagina": ""}, + {"campo": "categoria", "valor": "A", "pagina": ""}, + {"campo": "pdf", "valor": "", "pagina": ""}, + ], + } + ] + + result = desaplanar(filas) + assert result == expected_data + + def test_desaplanar_multiple_records(self): + """ + Test that the desaplanar function correctly converts + multiple recordsfrom flattened format to structured format. + """ + filas = [ + [ + "id", + "tipo_cbte", + "punto_vta", + "cbt_numero", + "fecha_cbte", + "tipo_doc", + "nro_doc", + "moneda_id", + "moneda_ctz", + "imp_neto", + "imp_iva", + "imp_trib", + "imp_op_ex", + "imp_tot_conc", + "imp_total", + "concepto", + "fecha_venc_pago", + "fecha_serv_desde", + "fecha_serv_hasta", + "cae", + "fecha_vto", + "resultado", + "motivo", + "reproceso", + "nombre", + "domicilio", + "localidad", + "telefono", + "categoria", + "email", + "numero_cliente", + "numero_orden_compra", + "condicion_frente_iva", + "numero_cotizacion", + "numero_remito", + "obs_generales", + "obs_comerciales", + "forma_pago", + "pdf", + "codigo1", + "descripcion1", + "cantidad1", + "umed1", + "precio1", + "importe1", + "bonif1", + "iva_id1", + "imp_iva1", + "tributo_id_1", + "tributo_desc_1", + "tributo_base_imp_1", + "tributo_alic_1", + "tributo_importe_1", + "iva_id_1", + "iva_base_imp_1", + "iva_importe_1", + ], + [ + 1, + 1, + 1, + 1, + "2023-06-08", + 80, + "20123456789", + "PES", + 1, + 100, + 21, + 5, + 0, + 0, + 126, + 1, + "2023-06-08", + "2023-06-08", + "2023-06-08", + "1234567890", + "2023-06-18", + "A", + "", + "", + "John Doe", + "123 Main St", + "City", + "1234567890", + "A", + "john@example.com", + "ABC123", + "OC123", + "Responsable Inscripto", + "COT123", + "REM123", + "Observaciones generales", + "Observaciones comerciales", + "Contado", + "", + "P1", + "Producto 1", + 2, + 7, + 50, + 100, + 0, + 5, + 21, + 1, + "Tributo 1", + 100, + 5, + 5, + 5, + 100, + 21, + ], + [ + 2, + 1, + 1, + 2, + "2023-06-09", + 80, + "20987654321", + "PES", + 1, + 200, + 42, + 10, + 0, + 0, + 252, + 1, + "2023-06-09", + "2023-06-09", + "2023-06-09", + "0987654321", + "2023-06-19", + "A", + "", + "", + "Jane Smith", + "456 Elm St", + "Town", + "0987654321", + "B", + "jane@example.com", + "XYZ789", + "OC456", + "Responsable Inscripto", + "COT456", + "REM456", + "Observaciones generales", + "Observaciones comerciales", + "Tarjeta de Crédito", + "", + "P2", + "Producto 2", + 1, + 7, + 200, + 200, + 0, + 5, + 42, + 2, + "Tributo 2", + 200, + 5, + 10, + 5, + 200, + 42, + ], + ] + + expected_data = [ + { + "cbte_nro": 1, + "tipo_cbte": 1, + "punto_vta": 1, + "cbt_numero": 1, + "fecha_cbte": "2023-06-08", + "concepto": 1, + "moneda_id": "PES", + "moneda_ctz": 1, + "tipo_doc": 80, + "nro_doc": "20123456789", + "email": "john@example.com", + "numero_cliente": "ABC123", + "numero_orden_compra": "OC123", + "condicion_frente_iva": "Responsable Inscripto", + "numero_cotizacion": "COT123", + "numero_remito": "REM123", + "imp_total": 126, + "imp_tot_conc": 0, + "imp_neto": 100, + "imp_iva": 21, + "imp_trib": 5, + "imp_op_ex": 0, + "fecha_serv_desde": "2023-06-08", + "fecha_serv_hasta": "2023-06-08", + "fecha_venc_pago": "2023-06-08", + "obs_generales": "Observaciones generales", + "obs_comerciales": "Observaciones comerciales", + "resultado": "A", + "cae": "1234567890", + "fecha_vto": "2023-06-18", + "reproceso": "", + "motivo": "", + "id": 1, + "detalles": [ + { + "codigo": "P1", + "ds": "Producto 1", + "umed": 7, + "qty": 2, + "precio": 50, + "importe": 100, + "iva_id": 5, + "imp_iva": 21, + "bonif": None, + "despacho": False, + "dato_a": False, + "dato_b": False, + "dato_c": False, + "dato_d": False, + "dato_e": False, + } + ], + "tributos": [ + { + "tributo_id": 1, + "desc": "Tributo 1", + "base_imp": 100, + "alic": 5, + "importe": 5, + } + ], + "ivas": [{"iva_id": 5, "base_imp": 100, "importe": 21}], + "permisos": [], + "opcionales": [], + "cbtes_asoc": [], + "forma_pago": "Contado", + "datos": [ + {"campo": "nombre", "valor": "John Doe", "pagina": ""}, + {"campo": "domicilio", "valor": "123 Main St", "pagina": ""}, + {"campo": "localidad", "valor": "City", "pagina": ""}, + {"campo": "telefono", "valor": "1234567890", "pagina": ""}, + {"campo": "categoria", "valor": "A", "pagina": ""}, + {"campo": "pdf", "valor": "", "pagina": ""}, + ], + }, + { + "cbte_nro": 2, + "tipo_cbte": 1, + "punto_vta": 1, + "cbt_numero": 2, + "fecha_cbte": "2023-06-09", + "concepto": 1, + "moneda_id": "PES", + "moneda_ctz": 1, + "tipo_doc": 80, + "nro_doc": "20987654321", + "email": "jane@example.com", + "numero_cliente": "XYZ789", + "numero_orden_compra": "OC456", + "condicion_frente_iva": "Responsable Inscripto", + "numero_cotizacion": "COT456", + "numero_remito": "REM456", + "imp_total": 252, + "imp_tot_conc": 0, + "imp_neto": 200, + "imp_iva": 42, + "imp_trib": 10, + "imp_op_ex": 0, + "fecha_serv_desde": "2023-06-09", + "fecha_serv_hasta": "2023-06-09", + "fecha_venc_pago": "2023-06-09", + "obs_generales": "Observaciones generales", + "obs_comerciales": "Observaciones comerciales", + "resultado": "A", + "cae": "0987654321", + "fecha_vto": "2023-06-19", + "reproceso": "", + "motivo": "", + "id": 2, + "detalles": [ + { + "codigo": "P2", + "ds": "Producto 2", + "umed": 7, + "qty": 1, + "precio": 200, + "importe": 200, + "iva_id": 5, + "imp_iva": 42, + "bonif": None, + "despacho": False, + "dato_a": False, + "dato_b": False, + "dato_c": False, + "dato_d": False, + "dato_e": False, + } + ], + "tributos": [ + { + "tributo_id": 2, + "desc": "Tributo 2", + "base_imp": 200, + "alic": 5, + "importe": 10, + } + ], + "ivas": [{"iva_id": 5, "base_imp": 200, "importe": 42}], + "permisos": [], + "opcionales": [], + "cbtes_asoc": [], + "forma_pago": "Tarjeta de Crédito", + "datos": [ + {"campo": "nombre", "valor": "Jane Smith", "pagina": ""}, + {"campo": "domicilio", "valor": "456 Elm St", "pagina": ""}, + {"campo": "localidad", "valor": "Town", "pagina": ""}, + {"campo": "telefono", "valor": "0987654321", "pagina": ""}, + {"campo": "categoria", "valor": "B", "pagina": ""}, + {"campo": "pdf", "valor": "", "pagina": ""}, + ], + }, + ] + + result = desaplanar(filas) + assert result == expected_data + + +@pytest.mark.dontusefix +class TestEscribirFunction: + def test_escribir_facturas_csv(self): + """ + Test that the escribir function can write data + from facturas.csv correctly. + """ + # Read the contents of facturas.csv + with open("datos/facturas.csv", "r", newline="") as file: + csv_reader = csv.reader(file, delimiter=";") + filas = [row for row in csv_reader] + + # Write the data to a new file using the escribir function + filename = "test_facturas_output.csv" + escribir(filas, filename) + + # Check if the file was created + assert os.path.isfile(filename) + + # Read the contents of the output file and compare + # with the original data + with open(filename, "r", newline="") as file: + csv_reader = csv.reader(file, delimiter=";") + result = [row for row in csv_reader] + assert result == filas + + # Clean up the test file + os.remove(filename) + + def test_escribir_facturas_xlsx(self): + """ + Test that the escribir function can write data from + facturas.csv to an XLSX file correctly. + """ + # Read the contents of facturas.csv + with open("datos/facturas.csv", "r", newline="") as file: + csv_reader = csv.reader(file, delimiter=";") + filas = [row for row in csv_reader] + + # Write the data to an XLSX file using the escribir function + filename = "test_facturas_output.xlsx" + escribir(filas, filename) + + # Check if the file was created + assert os.path.isfile(filename) + + # Read the contents of the XLSX file and compare with the original data + workbook = load_workbook(filename) + sheet = workbook.active + result = [] + for row in sheet.iter_rows(values_only=True): + # Convert None values to empty strings + row = ["" if cell is None else str(cell) for cell in row] + result.append(row) + assert result == filas + + # Clean up the test file + os.remove(filename) +