You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
224 lines
10 KiB
224 lines
10 KiB
"""All examples in this module are invented synthetic financial records.""" |
|
|
|
from decimal import localcontext, Inexact |
|
from pathlib import Path |
|
|
|
import pytest |
|
|
|
from ingestion.parsers import ParseError, parse_document, partial_context_from_creation_line |
|
from ingestion.normalization import normalize |
|
|
|
|
|
HEADER = 'Data,Dependencia,Historico,Documento,Tipo,Valor\n' |
|
|
|
|
|
def bank(rows, encoding='utf-8'): |
|
return parse_document((HEADER + rows).encode(encoding), 'BANK_CSV_V1') |
|
|
|
|
|
def card_line(day='15.02.2026', body='LOJA INVENTADA', debit='12,34', credit='0,00', partial=False): |
|
return day + body.ljust(43 if partial else 37) + 'BR' + f' {debit} {credit}' |
|
|
|
|
|
def alternate_full_card_line(day='15.02.2026', body='LOJA EXTERIOR', country='ESTADOS', debit='12,34', credit='0,00'): |
|
return day + body.ljust(33) + country.ljust(7) + f' {debit} {credit}' |
|
|
|
|
|
def card(text, partial=False, context=None): |
|
return normalize(parse_document(text.encode(), 'OUROCARD_PARTIAL_V1' if partial else 'OUROCARD_FULL_V1', context)) |
|
|
|
|
|
def test_csv_quotes_signs_balances_encoding_and_physical_locators(): |
|
result = bank('01/02/2026,,"Café, inventado",,,-12.34\n02/02/2026,,S A L D O,,,22.00\n\n03/02/2026,,RECEITA,,,23.00\n', 'latin-1') |
|
rows = normalize(result) |
|
assert result.encoding == 'latin-1' |
|
assert sum(result.skipped.values()) == 2 |
|
assert [r.to_dict()['locator'] for r in rows] == ['line:2', 'line:5'] |
|
assert rows[0].to_dict() == dict(locator='line:2', section='1', posting_date='2026-02-01', occurrence_date='2026-02-01', description='Café, inventado', amount='-12.34', kind='EXPENSE', installment_number=None, installment_total=None, errors=[]) |
|
assert rows[1].kind == 'INCOME' |
|
|
|
|
|
@pytest.mark.parametrize('data,profile,code', [(b'\x00','BANK_CSV_V1','NUL_BYTE'), (b'x','OFX','UNSUPPORTED_PROFILE'), (b'a,b,c\n','BANK_CSV_V1','INVALID_HEADER'), (b'x' * (10*1024*1024+1),'BANK_CSV_V1','INPUT_TOO_LARGE')], ids=['nul','unsupported','header','size']) |
|
def test_file_errors(data, profile, code): |
|
with pytest.raises(ParseError) as exc: |
|
parse_document(data, profile) |
|
assert exc.value.code == code |
|
|
|
|
|
def test_bad_csv_rows_are_reviewable_without_raw_diagnostics(): |
|
rows = normalize(bank('31/02/2026,,SECRET,,,NaN\nnot,a,row\n')) |
|
assert set(rows[0].errors) == {'INVALID_DATE', 'INVALID_AMOUNT'} |
|
assert rows[0].description == 'SECRET' |
|
assert rows[1].errors == ('INVALID_ROW',) |
|
assert rows[1].description == 'row' |
|
assert 'SECRET' not in str(rows[0].errors) |
|
|
|
|
|
def test_full_card_sections_installments_credit_and_payment(): |
|
rows = card('1 - SYNTHETIC OWNER\n' + card_line(body='LOJA PARC 2/3') + '\n2 - SECOND INVENTED OWNER\n' + card_line(debit='-8,11') + '\n15.02.2026PGTO DEBITO CONTA 200211 33,00 0,00\nSaldo Anterior 20,00\nunknown text') |
|
assert rows[0].section == '1' |
|
assert rows[0].installment_number == 2 and rows[0].installment_total == 3 |
|
assert rows[1].section == '2' and rows[1].amount == '8.11' and rows[1].kind == 'REFUND' |
|
assert rows[2].amount == '33.00' and rows[2].kind == 'TRANSFER' |
|
assert rows[3].errors == ('UNRECOGNIZED_LINE',) |
|
assert rows[3].description == 'unknown text' |
|
|
|
|
|
def test_partial_card_skips_metadata_and_parses_payment_with_variable_identifiers(): |
|
parsed = parse_document(( |
|
'SYNTHETIC INVOICE HEADER\n' |
|
'15/01 PGTO DEBITO CONTA 0001 123456789 123 123 -100,00 0,00\n' |
|
).encode(), 'OUROCARD_PARTIAL_V1', {'start_date': '2025-02-01', 'end_date': '2026-01-31'}) |
|
|
|
rows = normalize(parsed) |
|
|
|
assert parsed.skipped['metadata'] == 1 |
|
assert len(rows) == 1 |
|
assert rows[0].errors == () |
|
assert rows[0].kind == 'TRANSFER' |
|
assert rows[0].amount == '100.00' |
|
|
|
@pytest.mark.parametrize('country', ['ESTADOS', 'Brasil']) |
|
def test_full_card_accepts_extended_country_layout(country): |
|
row = card(alternate_full_card_line(country=country))[0] |
|
assert row.errors == () |
|
assert row.description == 'LOJA EXTERIOR' |
|
assert row.amount == '-12.34' |
|
|
|
|
|
def test_card_ambiguity_bad_shapes_and_installments_block(): |
|
assert 'AMBIGUOUS_AMOUNT' in card(card_line(debit='2,00', credit='1,00'))[0].errors |
|
assert 'INVALID_ROW' in card('15.02.2026 MALFORMED')[0].errors |
|
assert 'INVALID_ROW' in card('15/01 MALFORMED', True)[0].errors |
|
assert 'INVALID_INSTALLMENT' in card(card_line(body='LOJA PARC 3/2'))[0].errors |
|
assert 'INVALID_AMOUNT' in card(card_line(debit='1.23,00'))[0].errors |
|
|
|
|
|
def test_partial_dates_require_explicit_unique_context_and_retain_snapshot(): |
|
text = card_line(day='31/12', partial=True) + '\n' + card_line(day='01/01', partial=True) |
|
context = {'start_date':'2025-12-01','end_date':'2026-01-31'} |
|
parsed = parse_document(text.encode(), 'OUROCARD_PARTIAL_V1', context) |
|
context['start_date'] = '2000-01-01' |
|
assert [r.posting_date for r in normalize(parsed)] == ['2025-12-31','2026-01-01'] |
|
assert 'MISSING_DATE_CONTEXT' in card(text, True)[0].errors |
|
assert 'AMBIGUOUS_DATE' in card(card_line(day='01/01',partial=True), True, {'start_date':'2025-01-01','end_date':'2026-01-01'})[0].errors |
|
assert 'INVALID_DATE_CONTEXT' in card(text, True, {'start_date':'2024-01-01','end_date':'2026-01-01'})[0].errors |
|
|
|
|
|
def test_decimal_context_does_not_change_amounts(): |
|
with localcontext() as ctx: |
|
ctx.prec = 2 |
|
ctx.traps[Inexact] = True |
|
assert card(card_line(debit='12.345,67'))[0].amount == '-12345.67' |
|
assert card('15.02.2026PGTO DEBITO CONTA 200211 -12.345,67 0,00')[0].amount == '12345.67' |
|
|
|
|
|
def test_csv_multiline_locator_and_bom(): |
|
parsed = parse_document(('\ufeff' + HEADER + '01/02/2026,,"two\nlines",,,1.00\n02/02/2026,,next,,,2.00\n').encode(), 'BANK_CSV_V1') |
|
assert parsed.encoding == 'utf-8-sig' |
|
assert [r.locator for r in normalize(parsed)] == ['line:2','line:4'] |
|
|
|
|
|
def test_candidate_row_limit(): |
|
with pytest.raises(ParseError, match='TOO_MANY_ITEMS'): |
|
bank('01/02/2026,,INVENTED,,,1.00\n' * 20001) |
|
|
|
|
|
def test_invalid_quoting_is_file_error_and_header_positions_are_fixed(): |
|
with pytest.raises(ParseError, match='INVALID_CSV'): |
|
bank('01/02/2026,,"unterminated,,,1.00') |
|
with pytest.raises(ParseError, match='INVALID_HEADER'): |
|
parse_document(b'Valor,Unused,Historico,Unused,Unused,Data\n', 'BANK_CSV_V1') |
|
|
|
|
|
def test_partial_leap_date_and_invalid_ranges(): |
|
text = card_line(day='29/02', partial=True) |
|
assert card(text, True, {'start_date':'2024-01-01','end_date':'2024-12-31'})[0].posting_date == '2024-02-29' |
|
assert card(text, True, {'start_date':'2025-01-01','end_date':'2025-12-31'})[0].errors == ('INVALID_DATE',) |
|
assert card(text, True, {'start_date':'2026-01-01','end_date':'2025-12-31'})[0].errors == ('INVALID_DATE_CONTEXT',) |
|
|
|
|
|
def test_nonzero_payment_secondary_column_is_ambiguous(): |
|
assert card('15.02.2026PGTO DEBITO CONTA 200211 33,00 2,00')[0].errors == ('AMBIGUOUS_AMOUNT',) |
|
|
|
|
|
def test_partial_context_uses_the_validated_line_21_creation_date(): |
|
data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode() |
|
|
|
assert partial_context_from_creation_line(data) == { |
|
'start_date': '2025-02-01', |
|
'end_date': '2026-01-31', |
|
} |
|
|
|
|
|
def test_partial_context_accepts_fixed_width_line_21_creation_date(): |
|
data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + [ |
|
'15/01/2026 SYNTHETIC-STATEMENT 01:02:03' |
|
]) + '\n').encode() |
|
|
|
assert partial_context_from_creation_line(data) == { |
|
'start_date': '2025-02-01', |
|
'end_date': '2026-01-31', |
|
} |
|
|
|
|
|
@pytest.mark.parametrize('data,code', [ |
|
(b'SYNTHETIC\n' * 20, 'MISSING_CREATION_DATE'), |
|
(('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/26']) + '\n').encode(), 'INVALID_CREATION_DATE'), |
|
], ids=['absent-line-21', 'malformed-line-21']) |
|
def test_partial_context_rejects_missing_or_malformed_line_21_creation_date(data, code): |
|
with pytest.raises(ParseError, match=code): |
|
partial_context_from_creation_line(data) |
|
|
|
|
|
def test_partial_context_resolves_december_rows_to_the_previous_year_after_january_creation(): |
|
data = ('\n'.join(['SYNTHETIC HEADER'] * 20 + ['Data de emiss\u00e3o: 15/01/2026']) + '\n').encode() |
|
context = partial_context_from_creation_line(data) |
|
|
|
rows = card(card_line(day='31/12', partial=True), True, context) |
|
|
|
assert rows[0].posting_date == '2025-12-31' |
|
|
|
|
|
@pytest.mark.parametrize('creation', ['20/01/0001', '15/11/0001']) |
|
def test_partial_context_rejects_derived_interval_year_underflow(creation): |
|
data = ('\n' * 20 + f'Data de emissão: {creation}\n').encode() |
|
with pytest.raises(ParseError, match='^INVALID_CREATION_DATE$'): |
|
partial_context_from_creation_line(data) |
|
|
|
|
|
def test_partial_context_accepts_year_one_when_derived_bounds_are_representable(): |
|
data = ('\n' * 20 + 'Data de emissão: 15/12/0001\n').encode() |
|
assert partial_context_from_creation_line(data) == { |
|
'start_date': '0001-01-01', 'end_date': '0001-12-31', |
|
} |
|
|
|
|
|
@pytest.mark.parametrize('filename,profile,amounts', [ |
|
('bank.csv', 'BANK_CSV_V1', ['-12.34','100.00']), |
|
('full.txt', 'OUROCARD_FULL_V1', ['-12.34','8.11','33.00']), |
|
('partial.txt', 'OUROCARD_PARTIAL_V1', ['-12.34','-12.34']), |
|
]) |
|
def test_synthetic_fixture_profiles(filename, profile, amounts): |
|
data = (Path(__file__).parent / 'fixtures' / 'synthetic' / 'ingestion' / filename).read_bytes() |
|
rows = normalize(parse_document(data, profile, {'start_date':'2025-12-01','end_date':'2026-01-31'})) |
|
assert [row.amount for row in rows] == amounts |
|
assert all(not row.errors for row in rows) |
|
|
|
|
|
@pytest.mark.parametrize("line,position,profile,skipped", [ |
|
("Data de emissão: 20/01/2026", 21, "OUROCARD_PARTIAL_V1", True), |
|
("Data de emissão: 31/02/2026", 21, "OUROCARD_PARTIAL_V1", False), |
|
("Data de emissão: 20/01/2026 trailing text", 21, "OUROCARD_PARTIAL_V1", False), |
|
("Data de emissão: 20/01/2026", 20, "OUROCARD_PARTIAL_V1", False), |
|
("Data de emissão: 20/01/2026", 21, "OUROCARD_FULL_V1", False), |
|
]) |
|
def test_only_valid_line_21_partial_creation_metadata_is_excluded(line, position, profile, skipped): |
|
data = ("\n" * (position - 1) + line + "\n").encode() |
|
parsed = parse_document(data, profile) |
|
if skipped: |
|
assert parsed.items == () |
|
assert parsed.skipped["metadata"] == 1 |
|
else: |
|
assert len(parsed.items) == 1 |
|
assert parsed.items[0].errors == ("UNRECOGNIZED_LINE",)
|
|
|