"""Tests for synthetic fixtures only; no customer data, model, or API is used.""" import io import json import subprocess import sys import unittest from pathlib import Path from validate_orders import ValidationError, validate_orders EXAMPLE_DIR = Path(__file__).resolve().parent HEADER = "record_id,status,amount_cny\n" def verify(body): return validate_orders(io.StringIO(HEADER + body)) class OrderVerificationTests(unittest.TestCase): def test_synthetic_fixture(self): with (EXAMPLE_DIR / "orders.csv").open(newline="", encoding="utf-8") as source: result = validate_orders(source) self.assertEqual(result["input_rows"], 8) self.assertEqual(result["distinct_valid_record_ids"], 7) self.assertEqual(result["exact_duplicate_rows"], 1) self.assertEqual(result["confirmed_paid_records"], 3) self.assertEqual(result["confirmed_excluded_records"], 2) self.assertEqual(result["unresolved_records"], 2) self.assertEqual(result["confirmed_paid_total"], "1400.50") self.assertTrue(result["incomplete"]) self.assertEqual( [(item["record_id"], item["reason"], item["lines"]) for item in result["anomalies"]], [("A05", "missing_amount", [7]), ("A07", "invalid_amount", [9])], ) def test_empty_file_is_not_valid_schema(self): with self.assertRaises(ValidationError): validate_orders(io.StringIO("")) def test_header_only_has_no_unknown_records(self): result = verify("") self.assertEqual(result["input_rows"], 0) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertFalse(result["incomplete"]) def test_strict_schema(self): for header in ( "record_id,status\n", "record_id,status,amount_cny,extra\n", "status,record_id,amount_cny\n", "record_id,status,status\n", ): with self.subTest(header=header), self.assertRaises(ValidationError): validate_orders(io.StringIO(header)) def test_wrong_number_of_row_fields(self): for row in ("A01,paid\n", "A01,paid,10.00,extra\n"): with self.subTest(row=row), self.assertRaises(ValidationError): verify(row) def test_malformed_csv_is_rejected(self): with self.assertRaises(ValidationError): verify('A01,paid,"10.00\n') def test_distinct_ids_with_same_amount_both_count(self): result = verify("A01,paid,80.50\nA02,paid,80.50\n") self.assertEqual(result["confirmed_paid_total"], "161.00") self.assertEqual(result["confirmed_paid_records"], 2) self.assertEqual(result["exact_duplicate_rows"], 0) def test_exact_duplicates_count_once_and_keep_lines(self): result = verify("A01,paid,10.00\nA01,paid,10.00\nA01,paid,10.00\n") self.assertEqual(result["confirmed_paid_total"], "10.00") self.assertEqual(result["exact_duplicate_rows"], 2) self.assertEqual(result["included"][0]["lines"], [2, 3, 4]) self.assertEqual(result["duplicates"][0]["first_line"], 2) def test_conflicting_amounts_quarantine_every_variant(self): for body in ( "A01,paid,10.00\nA01,paid,20.00\n", "A01,paid,20.00\nA01,paid,10.00\n", ): with self.subTest(body=body): result = verify(body) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertEqual(result["confirmed_paid_records"], 0) self.assertEqual(result["unresolved_records"], 1) self.assertEqual(result["unresolved_rows_after_deduplication"], 2) self.assertEqual(result["anomalies"][0]["lines"], [2, 3]) self.assertEqual(result["anomalies"][0]["reason"], "conflicting_record_id") def test_conflicting_status_is_not_resolved_by_row_order(self): result = verify("A01,paid,10.00\nA01,refunded,10.00\n") self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertEqual(result["confirmed_excluded_records"], 0) self.assertTrue(result["incomplete"]) def test_duplicate_inside_conflict_is_reported_but_never_included(self): result = verify("A01,paid,10.00\nA01,paid,10.00\nA01,paid,20.00\n") self.assertEqual(result["exact_duplicate_rows"], 1) self.assertEqual(result["unresolved_rows_after_deduplication"], 2) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertEqual(result["anomalies"][0]["lines"], [2, 3, 4]) def test_formatting_difference_is_a_conflict_not_silent_normalization(self): result = verify('A01,paid,1200.00\nA01,paid,"1,200.00"\n') self.assertEqual(result["anomalies"][0]["reason"], "conflicting_record_id") def test_decimal_addition_does_not_use_binary_float(self): result = verify("A01,paid,0.10\nA02,paid,0.20\n") self.assertEqual(result["confirmed_paid_total"], "0.30") self.assertFalse(result["incomplete"]) def test_valid_thousands_and_decimal_formats(self): for raw, expected in ( ("0", "0.00"), ("0.5", "0.50"), ("1234.56", "1234.56"), ("1,200.00", "1200.00"), ("123,456,789.01", "123456789.01"), ): with self.subTest(raw=raw): result = verify('A01,paid,"{}"\n'.format(raw)) self.assertEqual(result["confirmed_paid_total"], expected) self.assertFalse(result["incomplete"]) def test_invalid_amounts_are_not_guessed_or_rounded(self): for raw in ( "", "12O.00", "12,00", "1234,567.00", "0,000.00", "1,200,00", ".50", "1.", "1.001", "00.50", "01.00", "+1.00", " 1.00 ", "NaN", "Infinity", "-Infinity", "1e2", "1_000.00", "$1.00", ): with self.subTest(raw=raw): result = verify('A01,paid,"{}"\n'.format(raw)) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertTrue(result["incomplete"]) self.assertEqual(result["unresolved_records"], 1) def test_negative_paid_amount_is_explicitly_rejected(self): result = verify("A01,paid,-1.00\n") self.assertEqual(result["anomalies"][0]["reason"], "negative_paid_amount") def test_unknown_status_and_partial_refund_need_an_explicit_rule(self): for status in ("", "Paid", "pending", "partially_refunded"): with self.subTest(status=status): result = verify("A01,{},10.00\n".format(status)) self.assertEqual(result["anomalies"][0]["reason"], "unknown_status") self.assertEqual(result["confirmed_paid_total"], "0.00") def test_fully_refunded_and_unpaid_are_confirmed_exclusions(self): result = verify("A01,refunded,10.00\nA02,unpaid,20.00\n") self.assertEqual(result["confirmed_excluded_records"], 2) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertFalse(result["incomplete"]) def test_missing_or_invalid_ids_are_not_merged(self): for record_id in ("", " A01", "A01 ", "A 01"): with self.subTest(record_id=record_id): result = verify("{},paid,10.00\n{},paid,10.00\n".format( record_id, record_id)) self.assertEqual(result["confirmed_paid_total"], "0.00") self.assertEqual(result["unresolved_records"], 2) self.assertEqual(result["exact_duplicate_rows"], 0) def test_very_large_exact_amount_is_not_rounded_by_decimal_context(self): raw = "123456789012345678901234567890.12" result = verify("A01,paid,{}\nA02,paid,0.01\n".format(raw)) self.assertEqual(result["confirmed_paid_total"], "123456789012345678901234567890.13") def test_cli_prints_json_for_the_synthetic_file(self): completed = subprocess.run( [sys.executable, str(EXAMPLE_DIR / "validate_orders.py"), str(EXAMPLE_DIR / "orders.csv")], text=True, capture_output=True, check=True, ) self.assertEqual(json.loads(completed.stdout)["confirmed_paid_total"], "1400.50") self.assertEqual(completed.stderr, "") if __name__ == "__main__": unittest.main()