"""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",)