doczyai-pipelines/fieldExtraction/tests/test_crosswalk_utils.py

import json

import pandas as pd
import pytest
from src.crosswalk.crosswalk_builder import CrosswalkBuilder
from src.utils.crosswalk_utils import apply_crosswalk


@pytest.fixture
def sample_excel_file(tmp_path):
    data = {"from_col": ["A", "B", "C"], "to_col": ["one", "two", "three"]}
    df = pd.DataFrame(data)
    file_path = tmp_path / "sample.xlsx"
    df.to_excel(file_path, index=False)
    return file_path


@pytest.fixture
def sample_json_file(tmp_path):
    data = {
        "metadata": {"from_col": "from_col", "to_col": "to_col"},
        "mapping": {"A": "1", "B": "2", "C": "3"},
    }
    file_path = tmp_path / "sample.json"
    with open(file_path, "w") as f:
        json.dump(data, f)
    return file_path


def test_from_excel(sample_excel_file):
    builder = CrosswalkBuilder()
    builder.from_excel(sample_excel_file, "from_col", "to_col")
    assert builder.mapping == {"A": "one", "B": "two", "C": "three"}
    assert builder.metadata == {"from_col": "from_col", "to_col": "to_col"}


def test_from_excel_missing_column(sample_excel_file):
    builder = CrosswalkBuilder()
    with pytest.raises(ValueError, match="Column 'nonexistent' not found"):
        builder.from_excel(sample_excel_file, "nonexistent", "to_col")


def test_from_dict():
    mapping = {"A": "1", "B": "2", "C": "3"}
    builder = CrosswalkBuilder()
    builder.from_dict(mapping)
    assert builder.mapping == mapping
    assert builder.metadata == {"from_col": None, "to_col": None}


def test_from_json(sample_json_file):
    builder = CrosswalkBuilder()
    builder.from_json(sample_json_file)
    assert builder.mapping == {"A": "1", "B": "2", "C": "3"}
    assert builder.metadata == {"from_col": "from_col", "to_col": "to_col"}


def test_json_roundtrip(tmp_path):
    # Create and save a crosswalk
    original = CrosswalkBuilder()
    original.from_dict({"A": "1"}, "source", "target")
    json_path = tmp_path / "test.json"
    original.save(json_path)

    # Load it back
    loaded = CrosswalkBuilder().from_json(json_path)

    # Check everything matches
    assert loaded.mapping == original.mapping
    assert loaded.metadata == original.metadata


def test_save(tmp_path):
    mapping = {"A": "1", "B": "2", "C": "3"}
    builder = CrosswalkBuilder()
    builder.from_dict(mapping, "from_col", "to_col")
    output_path = tmp_path / "output.json"
    builder.save(output_path)

    with open(output_path, "r") as f:
        data = json.load(f)

    assert data["mapping"] == mapping
    assert data["metadata"] == {"from_col": "from_col", "to_col": "to_col"}


def test_to_csv(tmp_path):
    # Create a builder with some data
    builder = CrosswalkBuilder()
    builder.from_dict(
        mapping={"A": "one", "B": "two", "C": "three"},
        from_col="Source",
        to_col="Target",
    )

    # Export to CSV
    csv_path = tmp_path / "test.csv"
    builder.to_csv(csv_path)

    # Read back and verify
    df = pd.read_csv(csv_path)

    # Check column names match metadata
    assert list(df.columns) == ["Source", "Target"]

    # Check data matches original mapping
    result_dict = dict(zip(df["Source"], df["Target"]))
    assert result_dict == {"A": "one", "B": "two", "C": "three"}


def test_to_csv_no_metadata(tmp_path):
    # Create a builder without metadata
    builder = CrosswalkBuilder()
    builder.from_dict({"A": "one", "B": "two"})

    # Export to CSV
    csv_path = tmp_path / "test.csv"
    builder.to_csv(csv_path)

    # Read back and verify
    df = pd.read_csv(csv_path)

    # Check default column names
    assert list(df.columns) == ["source", "target"]

    # Check data
    result_dict = dict(zip(df["source"], df["target"]))
    assert result_dict == {"A": "one", "B": "two"}


def test_apply_crosswalk():
    mapping = {"A": "1", "B": "2", "C": "3"}
    assert apply_crosswalk("A", mapping) == "1"
    assert apply_crosswalk("D", mapping) == "N/A"
    assert apply_crosswalk("D", mapping, default="0") == "N/A"


def test_create_reverse_mapping():
    builder = CrosswalkBuilder()
    builder.from_dict({"A": "1", "B": "1", "C": "2", "D": "2", "E": "3"})

    reversed_mapping = builder.create_reverse_mapping()
    assert reversed_mapping == {"1": "A|B", "2": "C|D", "3": "E"}


# Defensive programming tests
def test_apply_crosswalk_edge_cases():
    """Test defensive programming features in apply_crosswalk"""
    mapping = {"A": "1", "B": "2", "C": "3"}

    # Test None and empty values
    assert apply_crosswalk(None, mapping, default="default") == "default"
    assert apply_crosswalk("", mapping, default="default") == "default"
    assert (
        apply_crosswalk("   ", mapping, default="default") == "default"
    )  # Empty after strip, WITH default
    assert apply_crosswalk("   ", mapping) == "N/A"  # Empty after strip, no default

    # Test non-string inputs (should return original as string)
    assert apply_crosswalk(123, mapping) == "N/A"  # No default = return original
    assert apply_crosswalk(123, mapping, default="default") == "N/A"  # With default

    # Test list-like string input
    assert apply_crosswalk("['A']", mapping) == "1"  # String representation of list

    # Test comma-separated values
    result = apply_crosswalk("A,B", mapping)
    assert result in ["1|2", "2|1"]  # Order may vary due to set

    # Test pipe-separated values
    result = apply_crosswalk("A|B", mapping)
    assert result in ["1|2", "2|1"]

    # Test list string format
    result = apply_crosswalk("['A', 'B']", mapping)
    assert result in ["1|2", "2|1"]


def test_apply_crosswalk_nested_structures():
    """Test handling of nested data structures"""
    mapping = {"A": "1", "B": "2", "C": "3"}

    # Test nested lists
    assert apply_crosswalk("[['A'], ['B']]", mapping) in ["1|2", "2|1"]

    # Test mixed nested structures
    assert apply_crosswalk("['A', ['B', 'C']]", mapping) in [
        "1|2|3",
        "1|3|2",
        "2|1|3",
        "2|3|1",
        "3|1|2",
        "3|2|1",
    ]


def test_apply_crosswalk_reverse_mapping():
    """Test when value is already in mapping.values()"""
    mapping = {"A": "1", "B": "2", "C": "3"}

    # Value already mapped should be preserved
    assert apply_crosswalk("1", mapping) == "1"
    assert apply_crosswalk("2", mapping) == "2"


def test_apply_crosswalk_empty_results():
    """Test when mapping results in empty values"""
    mapping = {"A": "", "B": "2", "C": ""}  # Some empty mappings

    # Should return non-empty results only
    assert apply_crosswalk("A,B", mapping) == "2"
    assert apply_crosswalk("A,C", mapping, default="default") == "N/A"


def test_flatten_to_strings():
    """Test the flatten_to_strings helper function"""
    from src.utils.string_utils import flatten_to_strings

    # Test simple cases
    assert flatten_to_strings("A") == ["A"]
    assert flatten_to_strings(["A", "B"]) == ["A", "B"]
    assert flatten_to_strings(("A", "B")) == ["A", "B"]

    # Test nested structures
    assert flatten_to_strings([["A"], ["B"]]) == ["A", "B"]
    assert flatten_to_strings([["A", "B"], "C"]) == ["A", "B", "C"]

    # Test mixed types
    assert flatten_to_strings([1, ["2", 3]]) == ["1", "2", "3"]

    # Test with whitespace
    assert flatten_to_strings([" A ", [" B "]]) == ["A", "B"]