datatools-dev/tests/test_fixes_unit.py

"""Isolated unit tests for individual fix functions in src.core.fixes.

The integration tests at tests/test_normalize.py exercise these
functions through the full analyze→fix pipeline. These tests pin each
function's behavior in isolation so a regression surfaces close to the
broken function rather than at the pipeline output.
"""

from __future__ import annotations

import pandas as pd
import pytest

from src.core.fixes import (
    clean_headers,
    normalize_line_endings,
    repair_mojibake,
    strip_nbsp,
    strip_zero_width,
    trim_whitespace,
)


# ---------------------------------------------------------------------------
# trim_whitespace
# ---------------------------------------------------------------------------

class TestTrimWhitespace:
    def test_strips_leading_trailing(self):
        df = pd.DataFrame({"x": ["  hello  ", " world "]})
        out, changed = trim_whitespace(df)
        assert list(out["x"]) == ["hello", "world"]
        assert changed == 2

    def test_collapses_internal_runs(self):
        df = pd.DataFrame({"x": ["a   b   c"]})
        out, _ = trim_whitespace(df)
        assert out.loc[0, "x"] == "a b c"

    def test_preserves_internal_in_structured(self):
        # Phone-shaped strings keep internal spacing (often semantic).
        df = pd.DataFrame({"x": ["(555) 123-4567"]})
        out, changed = trim_whitespace(df)
        assert out.loc[0, "x"] == "(555) 123-4567"
        assert changed == 0

    def test_empty_df(self):
        df = pd.DataFrame({"x": []})
        out, changed = trim_whitespace(df)
        assert len(out) == 0
        assert changed == 0

    def test_no_string_columns(self):
        df = pd.DataFrame({"n": [1, 2, 3]})
        out, changed = trim_whitespace(df)
        assert changed == 0
        assert list(out["n"]) == [1, 2, 3]

    def test_nan_preserved(self):
        df = pd.DataFrame({"x": ["  ok  ", None]})
        out, _ = trim_whitespace(df)
        assert out.loc[0, "x"] == "ok"
        # NaN/None passes through (becomes empty string after strip OR stays)
        assert out.loc[1, "x"] is None or out.loc[1, "x"] == ""

    def test_idempotent(self):
        df = pd.DataFrame({"x": ["  hello  world  "]})
        out1, _ = trim_whitespace(df)
        out2, changed2 = trim_whitespace(out1)
        assert changed2 == 0
        assert list(out2["x"]) == list(out1["x"])


# ---------------------------------------------------------------------------
# strip_nbsp
# ---------------------------------------------------------------------------

class TestStripNbsp:
    def test_replaces_nbsp_with_ascii_space(self):
        df = pd.DataFrame({"x": ["a b"]})
        out, changed = strip_nbsp(df)
        assert out.loc[0, "x"] == "a b"
        assert changed == 1

    def test_no_change_when_clean(self):
        df = pd.DataFrame({"x": ["a b c"]})
        out, changed = strip_nbsp(df)
        assert changed == 0

    def test_other_unicode_spaces(self):
        # Em space (U+2003), thin space (U+2009)
        df = pd.DataFrame({"x": ["a b c"]})
        out, _ = strip_nbsp(df)
        assert out.loc[0, "x"] == "a b c"

    def test_idempotent(self):
        df = pd.DataFrame({"x": ["a  b"]})
        out1, _ = strip_nbsp(df)
        out2, changed2 = strip_nbsp(out1)
        assert changed2 == 0


# ---------------------------------------------------------------------------
# strip_zero_width
# ---------------------------------------------------------------------------

class TestStripZeroWidth:
    def test_removes_zero_width_space(self):
        df = pd.DataFrame({"x": ["a​b"]})
        out, changed = strip_zero_width(df)
        assert out.loc[0, "x"] == "ab"
        assert changed == 1

    def test_removes_zero_width_joiner(self):
        df = pd.DataFrame({"x": ["a‍b"]})
        out, _ = strip_zero_width(df)
        assert out.loc[0, "x"] == "ab"

    def test_clean_passthrough(self):
        df = pd.DataFrame({"x": ["clean"]})
        out, changed = strip_zero_width(df)
        assert changed == 0

    def test_idempotent(self):
        df = pd.DataFrame({"x": ["a​b‌c"]})
        out1, _ = strip_zero_width(df)
        out2, changed2 = strip_zero_width(out1)
        assert changed2 == 0


# ---------------------------------------------------------------------------
# normalize_line_endings
# ---------------------------------------------------------------------------

class TestNormalizeLineEndings:
    def test_crlf_to_lf(self):
        df = pd.DataFrame({"x": ["line1\r\nline2"]})
        out, changed = normalize_line_endings(df)
        assert out.loc[0, "x"] == "line1\nline2"
        assert changed == 1

    def test_bare_cr_to_lf(self):
        df = pd.DataFrame({"x": ["line1\rline2"]})
        out, _ = normalize_line_endings(df)
        assert out.loc[0, "x"] == "line1\nline2"

    def test_already_lf_unchanged(self):
        df = pd.DataFrame({"x": ["line1\nline2"]})
        out, changed = normalize_line_endings(df)
        assert changed == 0

    def test_idempotent(self):
        df = pd.DataFrame({"x": ["a\r\nb\rc"]})
        out1, _ = normalize_line_endings(df)
        out2, changed2 = normalize_line_endings(out1)
        assert changed2 == 0


# ---------------------------------------------------------------------------
# clean_headers
# ---------------------------------------------------------------------------

class TestCleanHeaders:
    def test_strips_bom_from_header(self):
        df = pd.DataFrame({"name": [1], "email": [2]})
        out, changed = clean_headers(df)
        assert "name" in out.columns
        assert "name" not in out.columns
        assert changed >= 1

    def test_strips_nbsp_from_header(self):
        df = pd.DataFrame({"first name": [1]})
        out, _ = clean_headers(df)
        assert "first name" in out.columns

    def test_strips_trailing_whitespace_from_header(self):
        df = pd.DataFrame({"Email ": [1]})
        out, _ = clean_headers(df)
        assert "Email" in out.columns
        assert "Email " not in out.columns

    def test_non_string_label_preserved(self):
        df = pd.DataFrame({0: [1], 1: [2]})
        out, changed = clean_headers(df)
        assert list(out.columns) == [0, 1]
        assert changed == 0

    def test_clean_headers_idempotent(self):
        df = pd.DataFrame({"name": [1]})
        out1, _ = clean_headers(df)
        out2, changed2 = clean_headers(out1)
        assert changed2 == 0
        assert list(out2.columns) == list(out1.columns)


# ---------------------------------------------------------------------------
# repair_mojibake
# ---------------------------------------------------------------------------

_HAS_FTFY = True
try:
    import ftfy  # noqa: F401
except ImportError:
    _HAS_FTFY = False


@pytest.mark.skipif(not _HAS_FTFY, reason="ftfy library not installed — fix is a no-op")
class TestRepairMojibake:
    def test_classic_cafe_repair(self):
        df = pd.DataFrame({"x": ["café"]})  # café miscoded
        out, changed = repair_mojibake(df)
        assert out.loc[0, "x"] == "café"
        assert changed == 1

    def test_clean_text_unchanged(self):
        df = pd.DataFrame({"x": ["café"]})
        out, changed = repair_mojibake(df)
        assert changed == 0

    def test_no_string_columns(self):
        df = pd.DataFrame({"n": [1, 2]})
        out, changed = repair_mojibake(df)
        assert changed == 0

    def test_idempotent(self):
        df = pd.DataFrame({"x": ["café"]})
        out1, _ = repair_mojibake(df)
        out2, changed2 = repair_mojibake(out1)
        assert changed2 == 0


class TestRepairMojibakeNoFtfy:
    @pytest.mark.skipif(_HAS_FTFY, reason="ftfy installed — exercises the no-op path")
    def test_returns_input_unchanged_without_ftfy(self):
        df = pd.DataFrame({"x": ["café"]})
        out, changed = repair_mojibake(df)
        assert changed == 0
        assert out.loc[0, "x"] == "café"