datatools-dev/tests/test_audit_fixes.py

"""Regression tests for bugs surfaced by the cross-tool audit.

Each test pins a specific behavioral bug or gap that an audit
identified. Test names match the BUG-N / GAP-N tags in the audit
notes so a future reader can trace why each test exists.
"""

from __future__ import annotations

import json
from pathlib import Path

import numpy as np
import pandas as pd
import pytest

from src.core.analyze import _NULL_LIKE, _detect_mixed_case_email
import src.core.fixes as f
from src.core.config import (
    ColumnStrategyConfig,
    DeduplicationConfig,
    StrategyConfig,
)
from src.core.dedup import (
    Algorithm,
    ColumnMatchStrategy,
    MatchStrategy,
    deduplicate,
)
from src.core.io import detect_header_row
from src.core.text_clean import sentence_case, smart_title_case, strip_bom


# ---------------------------------------------------------------------------
# BUG-1: dedup NaN values must not match as duplicates
# ---------------------------------------------------------------------------

class TestDedupNaNHandling:
    def test_two_nan_emails_do_not_match(self):
        # Both rows have NaN for email; no other matching column. Without
        # the fix, str(NaN) == "nan" would match exactly and the rows
        # would silently merge.
        df = pd.DataFrame({
            "id": [1, 2],
            "email": [np.nan, np.nan],
        })
        strategies = [MatchStrategy(column_strategies=[
            ColumnMatchStrategy(column="email", algorithm=Algorithm.EXACT,
                                threshold=100.0),
        ])]
        result = deduplicate(df, strategies=strategies)
        assert len(result.deduplicated_df) == 2
        assert len(result.match_groups) == 0

    def test_one_nan_one_real_does_not_match(self):
        df = pd.DataFrame({
            "email": [np.nan, "alice@example.com"],
        })
        strategies = [MatchStrategy(column_strategies=[
            ColumnMatchStrategy(column="email", algorithm=Algorithm.EXACT),
        ])]
        result = deduplicate(df, strategies=strategies)
        assert len(result.deduplicated_df) == 2

    def test_none_does_not_match_string_none(self):
        df = pd.DataFrame({
            "name": [None, "None"],
        })
        strategies = [MatchStrategy(column_strategies=[
            ColumnMatchStrategy(column="name", algorithm=Algorithm.EXACT),
        ])]
        result = deduplicate(df, strategies=strategies)
        assert len(result.deduplicated_df) == 2


# ---------------------------------------------------------------------------
# BUG-2: removed_df must preserve column schema even when empty
# ---------------------------------------------------------------------------

class TestDedupRemovedDfSchema:
    def test_empty_removed_df_has_same_columns(self):
        df = pd.DataFrame({
            "name": ["alice", "bob", "carol"],
            "email": ["a@x.com", "b@x.com", "c@x.com"],
        })
        strategies = [MatchStrategy(column_strategies=[
            ColumnMatchStrategy(column="email", algorithm=Algorithm.EXACT),
        ])]
        result = deduplicate(df, strategies=strategies)
        # No duplicates → empty removed_df, but columns must match.
        assert len(result.removed_df) == 0
        assert list(result.removed_df.columns) == list(result.deduplicated_df.columns)


# ---------------------------------------------------------------------------
# GAP-3: missing column reference should raise
# ---------------------------------------------------------------------------

class TestDedupMissingColumn:
    def test_missing_column_raises(self):
        df = pd.DataFrame({"email": ["a@x.com"]})
        strategies = [MatchStrategy(column_strategies=[
            ColumnMatchStrategy(column="e_mail", algorithm=Algorithm.EXACT),
        ])]
        with pytest.raises(ValueError, match="not present in the input"):
            deduplicate(df, strategies=strategies)


# ---------------------------------------------------------------------------
# GAP-4: threshold must be in [0, 100]
# ---------------------------------------------------------------------------

class TestThresholdValidation:
    def test_negative_threshold_rejected(self):
        with pytest.raises(ValueError, match=r"\[0, 100\]"):
            ColumnMatchStrategy(column="x", threshold=-1)

    def test_over_hundred_rejected(self):
        with pytest.raises(ValueError, match=r"\[0, 100\]"):
            ColumnMatchStrategy(column="x", threshold=101)

    def test_zero_and_hundred_allowed(self):
        ColumnMatchStrategy(column="x", threshold=0)
        ColumnMatchStrategy(column="x", threshold=100)

    def test_non_numeric_rejected(self):
        with pytest.raises(TypeError):
            ColumnMatchStrategy(column="x", threshold="high")  # type: ignore[arg-type]


# ---------------------------------------------------------------------------
# BUG-9: replace_null_sentinels must coerce non-string sentinels
# ---------------------------------------------------------------------------

class TestReplaceNullSentinelsTypes:
    def test_int_sentinels_do_not_crash(self):
        df = pd.DataFrame({"x": ["0", "5", ""]})
        out, _ = f.replace_null_sentinels(df, {"sentinels": [0, "5"]})
        assert out.loc[0, "x"] == ""   # "0" matched int 0 stringified
        assert out.loc[1, "x"] == ""   # "5" matched
        assert out.loc[2, "x"] == ""   # already empty

    def test_none_sentinel_skipped(self):
        df = pd.DataFrame({"x": ["a", "b"]})
        # Should not crash on None entry in the sentinel list.
        out, _ = f.replace_null_sentinels(df, {"sentinels": ["a", None]})
        assert out.loc[0, "x"] == ""
        assert out.loc[1, "x"] == "b"


# ---------------------------------------------------------------------------
# BUG-10: malformed regex should raise ValueError, not re.error
# ---------------------------------------------------------------------------

class TestVectorizedRegexErrorHandling:
    def test_malformed_pattern_raises_valueerror(self):
        df = pd.DataFrame({"x": ["abc"]})
        with pytest.raises(ValueError, match="Invalid regex pattern"):
            f._vectorized_regex_sub(df, "[invalid", "")


# ---------------------------------------------------------------------------
# NIT-12: strip_bom strips at most one BOM
# ---------------------------------------------------------------------------

class TestStripBomSingleChar:
    def test_strips_one_leading_bom(self):
        assert strip_bom("hello") == "hello"

    def test_does_not_strip_multiple_consecutive_boms(self):
        # Per docstring: "at most one BOM". Second BOM stays so the
        # caller can see something odd happened.
        assert strip_bom("hello") == "hello"

    def test_no_bom_unchanged(self):
        assert strip_bom("hello") == "hello"

    def test_non_string_passthrough(self):
        assert strip_bom(None) is None  # type: ignore[arg-type]


# ---------------------------------------------------------------------------
# Smart title case — particle behavior at boundaries (regression / docs)
# ---------------------------------------------------------------------------

class TestSmartTitleCaseBoundaries:
    def test_first_word_particle_capitalized(self):
        # "a" at index 0 is a particle but must capitalize as the first
        # word of a title.
        assert smart_title_case("a story") == "A Story"

    def test_last_word_particle_capitalized(self):
        # "to" at the end is the last word; must capitalize.
        assert smart_title_case("things to") == "Things To"

    def test_mid_string_particles_lowercase(self):
        assert smart_title_case("the cat in the hat") == "The Cat in the Hat"


# ---------------------------------------------------------------------------
# NIT-14: sentence_case dead branch removed — regression guard
# ---------------------------------------------------------------------------

class TestSentenceCaseUnchanged:
    def test_basic(self):
        assert sentence_case("hello. world.") == "Hello. World."

    def test_open_paren_does_not_consume_trigger(self):
        # The dead-branch removal didn't change behavior; this is a
        # regression guard that opening punctuation still doesn't
        # capitalize itself but doesn't reset the trigger either.
        assert sentence_case('hello. "world"') == 'Hello. "World"'


# ---------------------------------------------------------------------------
# BUG-18: detect_header_row must not pick all-empty rows
# ---------------------------------------------------------------------------

class TestDetectHeaderRowEmptyRows:
    def test_all_empty_first_row_skipped(self, tmp_path: Path):
        # First row is all-empty — the header is on row 1.
        p = tmp_path / "blank_first.csv"
        p.write_text(",,\nname,email,phone\nalice,a@x.com,555\n")
        assert detect_header_row(p) == 1

    def test_pure_header_at_row_zero(self, tmp_path: Path):
        p = tmp_path / "normal.csv"
        p.write_text("name,email,phone\nalice,a@x.com,555\n")
        assert detect_header_row(p) == 0


# ---------------------------------------------------------------------------
# BUG-20: config.from_dict must accept unknown fields (forward compat)
# ---------------------------------------------------------------------------

class TestConfigForwardCompat:
    def test_extra_field_in_column_config_ignored(self, tmp_path: Path):
        # Simulate a config file written by a future version with an
        # extra ``priority`` field.
        config_dict = {
            "strategies": [{
                "columns": [{
                    "column": "email",
                    "algorithm": "exact",
                    "threshold": 100.0,
                    "normalizer": None,
                    "priority": 5,  # future field — must not crash
                }],
            }],
            "survivor_rule": "first",
            "merge": False,
        }
        loaded = DeduplicationConfig.from_dict(config_dict)
        assert len(loaded.strategies) == 1
        assert loaded.strategies[0].columns[0].column == "email"

    def test_roundtrip_then_reload_with_extra(self, tmp_path: Path):
        cfg = DeduplicationConfig(
            strategies=[StrategyConfig(columns=[
                ColumnStrategyConfig(column="email"),
            ])],
        )
        path = tmp_path / "cfg.json"
        cfg.to_file(path)
        # Manually inject an unknown field to simulate forward-compat.
        data = json.loads(path.read_text())
        data["strategies"][0]["columns"][0]["future_thing"] = "abc"
        path.write_text(json.dumps(data))
        loaded = DeduplicationConfig.from_file(path)
        assert loaded.strategies[0].columns[0].column == "email"


# ---------------------------------------------------------------------------
# BUG-22: mixed-case email detector must not flag all-None columns
# ---------------------------------------------------------------------------

class TestMixedCaseEmailFalsePositive:
    def test_all_none_email_column_no_finding(self):
        df = pd.DataFrame({
            "email": [None, None, None],
        })
        findings = _detect_mixed_case_email(df)
        assert findings == []

    def test_real_mixed_case_still_flagged(self):
        df = pd.DataFrame({
            "email": ["Alice@X.com", "bob@y.com"],
        })
        findings = _detect_mixed_case_email(df)
        assert len(findings) == 1
        assert findings[0].column == "email"


# ---------------------------------------------------------------------------
# NIT-24: <NA> recognized as a null-like sentinel
# ---------------------------------------------------------------------------

class TestNullLikeIncludesPandasNA:
    def test_pd_na_string_repr_recognized(self):
        # str(pd.NA) → "<NA>" — when a DataFrame is loaded with
        # keep_default_na=False, pandas NA values appear as the literal
        # string "<NA>" and the analyzer should flag them.
        assert "<na>" in _NULL_LIKE


# ---------------------------------------------------------------------------
# Production-readiness review (refactor pass)
# ---------------------------------------------------------------------------

class TestProductionReadyValidation:
    def test_chunk_size_must_be_positive(self, tmp_path: Path):
        from src.core.io import read_file
        f = tmp_path / "tiny.csv"
        f.write_text("a,b\n1,2\n")
        with pytest.raises(ValueError, match="chunk_size must be positive"):
            list(read_file(f, chunk_size=0))
        with pytest.raises(ValueError, match="chunk_size must be positive"):
            list(read_file(f, chunk_size=-5))

    def test_write_file_rejects_non_dataframe(self, tmp_path: Path):
        from src.core.io import write_file
        from src.core.errors import InputValidationError
        with pytest.raises(InputValidationError, match="requires a pandas DataFrame"):
            write_file({"a": [1]}, tmp_path / "out.csv")  # type: ignore[arg-type]

    def test_clean_dataframe_rejects_non_dataframe(self):
        from src.core.text_clean import clean_dataframe
        from src.core.errors import InputValidationError
        with pytest.raises(InputValidationError, match="requires a pandas DataFrame"):
            clean_dataframe([{"a": 1}])  # type: ignore[arg-type]

    def test_deduplicate_rejects_non_dataframe(self):
        from src.core.dedup import deduplicate
        from src.core.errors import InputValidationError
        with pytest.raises(InputValidationError, match="requires a pandas DataFrame"):
            deduplicate({"x": [1]})  # type: ignore[arg-type]

    def test_keep_most_recent_requires_date_column(self):
        from src.core.dedup import deduplicate, SurvivorRule
        df = pd.DataFrame({"name": ["a", "b"]})
        with pytest.raises(ValueError, match="date_column"):
            deduplicate(df, survivor_rule=SurvivorRule.KEEP_MOST_RECENT)

    def test_keep_most_recent_with_unknown_date_column(self):
        from src.core.dedup import deduplicate, SurvivorRule
        df = pd.DataFrame({"name": ["a", "b"]})
        with pytest.raises(ValueError, match="not found in input"):
            deduplicate(
                df,
                survivor_rule=SurvivorRule.KEEP_MOST_RECENT,
                date_column="created_at",
            )

    def test_standardize_options_invalid_enum(self):
        from src.core.format_standardize import StandardizeOptions
        with pytest.raises(ValueError, match="Invalid date_order"):
            StandardizeOptions.from_dict({"date_order": "XYZ"})

    def test_standardize_options_invalid_field_type(self):
        from src.core.format_standardize import StandardizeOptions
        with pytest.raises(ValueError, match="Invalid field type"):
            StandardizeOptions.from_dict({"column_types": {"x": "made_up"}})

    def test_month_locale_validation(self):
        from src.core.format_standardize import standardize_date
        with pytest.raises(ValueError, match="Unknown month locale"):
            standardize_date("15 januar 2024", month_locales=["DE"])  # uppercase typo

    def test_config_from_file_missing(self, tmp_path: Path):
        from src.core.config import DeduplicationConfig
        with pytest.raises(OSError, match="Could not read"):
            DeduplicationConfig.from_file(tmp_path / "missing.json")

    def test_config_from_file_bad_json(self, tmp_path: Path):
        from src.core.config import DeduplicationConfig
        path = tmp_path / "bad.json"
        path.write_text("{not: valid json")
        with pytest.raises(ValueError, match="Invalid JSON"):
            DeduplicationConfig.from_file(path)