ambers 0.4.2

Pure Rust reader for SPSS .sav and .zsav files
Documentation 
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325

"""Tests for ambers.validate().

Covers: unlabeled values (error), duplicate labels (warning),
column filtering, return types, shared helper consistency.

Run with:
    pytest tests/test_validate.py -v
"""

import polars as pl
import pytest
from polars.testing import assert_frame_equal

import ambers as am


def _meta_with_labels(labels: dict) -> am.SpssMetadata:
    return am.SpssMetadata(variable_value_labels=labels)


# ---------------------------------------------------------------------------
# Unlabeled values (error)
# ---------------------------------------------------------------------------

class TestUnlabeledValues:
    def test_all_labeled_no_errors(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0, 3.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B", 3: "C"}})
        report = am.validate(df, meta)
        assert report.is_valid
        assert len(report.errors) == 0

    def test_unlabeled_values_detected(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0, 3.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})
        report = am.validate(df, meta)
        assert not report.is_valid
        assert len(report.errors) == 1
        err = report.errors[0]
        assert err.column == "Q1"
        assert err.check == "unlabeled_values"
        assert 3.0 in err.details["unlabeled_values"]
        assert 99.0 in err.details["unlabeled_values"]
        assert err.details["unique_in_data"] == 4
        assert err.details["labeled_in_data"] == 2

    def test_column_without_value_labels_skipped(self):
        """Continuous numeric columns (no labels) are not checked."""
        df = pl.DataFrame({"age": [25.0, 30.0, 35.0]})
        meta = am.SpssMetadata()  # no value labels at all
        report = am.validate(df, meta)
        assert report.is_valid

    def test_string_column_skipped(self):
        """String columns are not checked even if they have value labels."""
        df = pl.DataFrame({"city": ["NYC", "LA", "CHI"]})
        meta = _meta_with_labels({"city": {"NYC": "New York", "LA": "Los Angeles"}})
        report = am.validate(df, meta)
        assert report.is_valid  # string column skipped

    def test_all_null_no_error(self):
        df = pl.DataFrame({"Q1": pl.Series([None, None, None], dtype=pl.Float64)})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})
        report = am.validate(df, meta)
        assert report.is_valid

    def test_multiple_columns_different_issues(self):
        df = pl.DataFrame({
            "Q1": [1.0, 2.0, 99.0],
            "Q2": [1.0, 2.0, 3.0],
            "Q3": [1.0, 2.0, 98.0],
        })
        meta = _meta_with_labels({
            "Q1": {1: "A", 2: "B"},       # 99 unlabeled
            "Q2": {1: "X", 2: "Y", 3: "Z"},  # all labeled
            "Q3": {1: "P", 2: "Q"},       # 98 unlabeled
        })
        report = am.validate(df, meta)
        assert not report.is_valid
        assert len(report.errors) == 2
        error_cols = {e.column for e in report.errors}
        assert error_cols == {"Q1", "Q3"}

    def test_labels_in_meta_not_in_data_is_fine(self):
        """Labels for values not present in data are NOT flagged."""
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B", 3: "C", 4: "D", 5: "E"}})
        report = am.validate(df, meta)
        assert report.is_valid


# ---------------------------------------------------------------------------
# Duplicate labels (warning)
# ---------------------------------------------------------------------------

class TestDuplicateLabels:
    def test_no_duplicates(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})
        report = am.validate(df, meta)
        assert len(report.warnings) == 0

    def test_duplicate_detected(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "Male", 2: "Female", 9: "Male"}})
        report = am.validate(df, meta)
        assert report.is_valid  # warnings don't affect is_valid
        assert len(report.warnings) == 1
        warn = report.warnings[0]
        assert warn.column == "Q1"
        assert warn.check == "duplicate_labels"
        assert "Male" in warn.details["duplicates"]

    def test_multiple_duplicates(self):
        df = pl.DataFrame({"Q1": [1.0]})
        meta = _meta_with_labels({
            "Q1": {1: "Yes", 2: "No", 3: "Yes", 4: "No"},
        })
        report = am.validate(df, meta)
        assert len(report.warnings) == 1
        dupes = report.warnings[0].details["duplicates"]
        assert "Yes" in dupes
        assert "No" in dupes


# ---------------------------------------------------------------------------
# Filtering (columns + exclude)
# ---------------------------------------------------------------------------

class TestFiltering:
    def test_columns_specific(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0], "Q2": [1.0, 98.0]})
        meta = _meta_with_labels({
            "Q1": {1: "A"},
            "Q2": {1: "X"},
        })
        report = am.validate(df, meta, columns=["Q1"])
        assert len(report.errors) == 1
        assert report.errors[0].column == "Q1"

    def test_exclude(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0], "Q2": [1.0, 98.0]})
        meta = _meta_with_labels({
            "Q1": {1: "A"},
            "Q2": {1: "X"},
        })
        report = am.validate(df, meta, exclude=["Q1"])
        assert len(report.errors) == 1
        assert report.errors[0].column == "Q2"

    def test_columns_and_exclude_combined(self):
        """columns + exclude can be combined: filter first, then exclude."""
        df = pl.DataFrame({
            "Q1": [1.0, 99.0],
            "Q2": [1.0, 98.0],
            "Q3": [1.0, 97.0],
        })
        meta = _meta_with_labels({
            "Q1": {1: "A"},
            "Q2": {1: "X"},
            "Q3": {1: "P"},
        })
        report = am.validate(df, meta, columns=["Q1", "Q2", "Q3"], exclude=["Q2"])
        error_cols = {e.column for e in report.errors}
        assert "Q2" not in error_cols
        assert "Q1" in error_cols
        assert "Q3" in error_cols


# ---------------------------------------------------------------------------
# Return type handling
# ---------------------------------------------------------------------------

class TestReturnTypes:
    def test_dataframe_input(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A"}})
        report = am.validate(df, meta)
        assert isinstance(report, am.ValidationReport)

    def test_lazyframe_input(self):
        lf = pl.DataFrame({"Q1": [1.0, 99.0]}).lazy()
        meta = _meta_with_labels({"Q1": {1: "A"}})
        report = am.validate(lf, meta)
        assert isinstance(report, am.ValidationReport)
        assert not report.is_valid

    def test_is_valid_true_no_errors(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})
        report = am.validate(df, meta)
        assert report.is_valid

    def test_is_valid_true_warnings_only(self):
        """Warnings don't affect is_valid."""
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B", 9: "A"}})
        report = am.validate(df, meta)
        assert report.is_valid  # has warnings but no errors
        assert len(report.warnings) == 1

    def test_is_valid_false_with_errors(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A"}})
        report = am.validate(df, meta)
        assert not report.is_valid

    def test_raise_if_invalid_raises_on_errors(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A"}})
        report = am.validate(df, meta)
        with pytest.raises(ValueError, match="validate"):
            report.raise_if_invalid()

    def test_raise_if_invalid_silent_on_warnings(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B", 9: "A"}})
        report = am.validate(df, meta)
        report.raise_if_invalid()  # should not raise

    def test_to_frame(self):
        df = pl.DataFrame({"Q1": [1.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 9: "A"}})
        report = am.validate(df, meta)
        frame = report.to_frame()
        assert isinstance(frame, pl.DataFrame)
        assert set(frame.columns) == {"severity", "column", "check", "message"}
        assert frame.height == 2  # 1 error + 1 warning

    def test_empty_report(self):
        df = pl.DataFrame({"Q1": [1.0, 2.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})
        report = am.validate(df, meta)
        assert report.is_valid
        assert len(report.issues) == 0
        assert report.to_frame().height == 0

    def test_empty_df(self):
        df = pl.DataFrame({"Q1": pl.Series([], dtype=pl.Float64)})
        meta = _meta_with_labels({"Q1": {1: "A"}})
        report = am.validate(df, meta)
        assert report.is_valid

    def test_invalid_df_type_raises(self):
        with pytest.raises(TypeError):
            am.validate([1, 2, 3], am.SpssMetadata())


# ---------------------------------------------------------------------------
# Shared helper consistency
# ---------------------------------------------------------------------------

# ---------------------------------------------------------------------------
# Repr truncation
# ---------------------------------------------------------------------------

class TestReprTruncation:
    def test_many_issues_truncated(self):
        """Repr should show max 10 errors, then '... and N more'."""
        df = pl.DataFrame({f"Q{i}": [1.0, 2.0, 99.0] for i in range(30)})
        meta = _meta_with_labels({
            f"Q{i}": {1: "Yes", 2: "No"} for i in range(30)
        })
        report = am.validate(df, meta)
        repr_str = repr(report)
        # Should show exactly 10 error column names
        assert repr_str.count("[x] Q") == 10
        # Should mention remaining
        assert "and 20 more errors" in repr_str
        # Should tell user how to see all
        assert "report.to_frame()" in repr_str or "report.errors" in repr_str

    def test_long_duplicate_message_truncated(self):
        """Repr should truncate messages with many duplicate labels."""
        labels = {i: f"Label{i % 3}" for i in range(30)}  # many dupes
        df = pl.DataFrame({"Q1": [1.0]})
        meta = _meta_with_labels({"Q1": labels})
        report = am.validate(df, meta)
        repr_str = repr(report)
        # No line should exceed box width
        for line in repr_str.split("\n"):
            assert len(line) <= 80, f"Line too long ({len(line)}): {line}"

    def test_box_width_capped(self):
        """Box should never exceed 80 characters wide."""
        # Single column with a very long label
        labels = {i: f"Very Long Label Name For Code {i}" for i in range(20)}
        df = pl.DataFrame({"Q1": list(range(20))})
        meta = _meta_with_labels({"Q1": labels})
        report = am.validate(df, meta)
        for line in repr(report).split("\n"):
            assert len(line) <= 80, f"Line too long ({len(line)}): {line}"

    def test_to_frame_has_full_messages(self):
        """to_frame() should contain full untruncated messages."""
        labels = {}
        for i in range(30):
            labels[float(i)] = f"Label{i % 3}"
        df = pl.DataFrame({"Q1": [float(i) for i in range(30)]})
        meta = _meta_with_labels({"Q1": labels})
        report = am.validate(df, meta)
        frame = report.to_frame()
        # to_frame message should NOT be truncated
        # The full message should contain all unlabeled value details
        assert frame.height > 0


# ---------------------------------------------------------------------------
# Shared helper consistency
# ---------------------------------------------------------------------------

class TestSharedHelpers:
    def test_validate_and_apply_labels_flag_same_values(self):
        """validate() and apply_labels() should catch the same unlabeled values."""
        df = pl.DataFrame({"Q1": [1.0, 2.0, 99.0]})
        meta = _meta_with_labels({"Q1": {1: "A", 2: "B"}})

        # validate should find 99 as unlabeled
        report = am.validate(df, meta)
        assert not report.is_valid
        assert 99.0 in report.errors[0].details["unlabeled_values"]

        # apply_labels should raise on the same issue
        with pytest.raises(ValueError):
            am.apply_labels(df, meta, output="enum")