csv-fix-kit/test_clean_csv.py at main · alvinhastra-web/csv-fix-kit · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
import csv
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

from clean_csv import clean_csv


class CleanupTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.source = self.root / "input.csv"
        self.output = self.root / "result"

    def write(self, text):
        self.source.write_text(text, encoding="utf-8", newline="")

    def read_output(self):
        with (self.output / "cleaned.csv").open(newline="", encoding="utf-8") as handle:
            return list(csv.reader(handle))

    def test_explicit_cleanup_retains_conflicts_and_leading_zero_ids(self):
        self.write("id,name,email\n00123, Alice ,alice@example.com\n00123,Alice,alice@example.com\n00123,Alice,other@example.com\n, Bob ,bob@example.com\n")
        original = self.source.read_bytes()
        report = clean_csv(self.source, self.output, trim=("name",), keys=("id",), drop_duplicates=True)
        self.assertEqual(self.read_output(), [
            ["id", "name", "email"], ["00123", "Alice", "alice@example.com"],
            ["00123", "Alice", "other@example.com"], ["", "Bob", "bob@example.com"],
        ])
        self.assertEqual(report["summary"], {
            "input_records": 4, "output_records": 3, "trimmed_records": 2,
            "exact_duplicates": 1, "removed_records": 1,
            "conflicting_key_groups": 1, "records_missing_key": 1,
        })
        self.assertEqual(report["conflicts"], [{"key": {"id": "00123"}, "records": [2, 3, 4], "differing_columns": ["email"]}])
        self.assertEqual(report["duplicates"][0]["duplicate_of_record"], 2)
        self.assertEqual(self.source.read_bytes(), original)
        self.assertEqual(json.loads((self.output / "report.json").read_text()), report)

    def test_defaults_preserve_whitespace_duplicates_dates_and_formulas_as_text(self):
        self.write('id,date,note\n001,01/02/26, =1+2 \n001,01/02/26, =1+2 \n')
        report = clean_csv(self.source, self.output)
        self.assertEqual(self.read_output()[1:], [["001", "01/02/26", " =1+2 "]] * 2)
        self.assertEqual(report["summary"]["removed_records"], 0)
        self.assertEqual(report["summary"]["exact_duplicates"], 1)

    def test_quoted_multiline_unicode_and_record_locations(self):
        self.write('\ufeffid,note\n001," café, tea\nand cake "\n001,"café, tea\nand cake"\n')
        report = clean_csv(self.source, self.output, trim=("note",), drop_duplicates=True)
        self.assertEqual(self.read_output(), [["id", "note"], ["001", "café, tea\nand cake"]])
        self.assertEqual(report["trimmed"][0], {"record": 2, "start_line": 2, "end_line": 3, "columns": ["note"]})
        self.assertEqual(report["duplicates"][0]["start_line"], 4)

    def test_invalid_inputs_do_not_create_partial_results(self):
        cases = ["", "a,a\n1,2\n", "a,\n1,2\n", "a,b\n1\n", 'a,b\n1,"unfinished\n', "a,b\n\n"]
        for content in cases:
            with self.subTest(content=content):
                self.write(content)
                with self.assertRaises((ValueError, csv.Error)):
                    clean_csv(self.source, self.output)
                self.assertFalse(self.output.exists())

    def test_unknown_column_and_existing_output_are_rejected(self):
        self.write("a\n1\n")
        with self.assertRaisesRegex(ValueError, "Unknown columns"):
            clean_csv(self.source, self.output, trim=("missing",))
        self.output.mkdir()
        (self.output / "keep.txt").write_text("keep me")
        with self.assertRaisesRegex(ValueError, "already exists"):
            clean_csv(self.source, self.output)
        self.assertEqual((self.output / "keep.txt").read_text(), "keep me")

    def test_composite_keys_distinguish_records(self):
        self.write("region,id,value\nA,01,x\nB,01,y\nA,01,z\n")
        report = clean_csv(self.source, self.output, keys=("region", "id"))
        self.assertEqual(report["conflicts"], [{"key": {"region": "A", "id": "01"}, "records": [2, 4], "differing_columns": ["value"]}])
        self.assertEqual(report["summary"]["output_records"], 3)

    def test_header_only_input_is_valid(self):
        self.write("a,b\n")
        report = clean_csv(self.source, self.output)
        self.assertEqual(self.read_output(), [["a", "b"]])
        self.assertEqual(report["summary"]["input_records"], 0)

    def test_cli_end_to_end_with_semicolon_delimiter(self):
        self.write("id;name\n007; Ada \n007;Ada\n")
        result = subprocess.run([
            sys.executable, str(Path(__file__).with_name("clean_csv.py")),
            str(self.source), str(self.output), "--delimiter", ";", "--trim", "name",
            "--drop-exact-duplicates", "--key", "id",
        ], capture_output=True, text=True, timeout=10)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["removed_records"], 1)
        with (self.output / "cleaned.csv").open(newline="") as handle:
            self.assertEqual(list(csv.reader(handle, delimiter=";")), [["id", "name"], ["007", "Ada"]])


if __name__ == "__main__":
    unittest.main()