Katon Minhas
c4e519894b
Refactor/one time io * Delete client work * Streamline imports * Fix list class * Update tests * Clear code funcs test * Fix unit tests * Single read code funcs * Single-read model * Single-load embeddings * Successful E2E test * update unit tests * Update importsg * Reload poetry.lock * remove old exhibit header function * Remove aarete_derived generic function * remove align and format tables * Remove strings to dict * references * Clear code * Move preprocessing_funcs * remove keywords * refactor postprocessing_funcs * Pass unit test - remove qa_qc directory * Black and isort * remove print Approved-by: Alex Galarce
234 lines
7.1 KiB
Python
234 lines
7.1 KiB
Python
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) == "D"
|
|
assert apply_crosswalk("D", mapping, default="0") == "0"
|
|
|
|
|
|
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) == "" # Empty after strip, no default
|
|
|
|
# Test non-string inputs (should return original as string)
|
|
assert apply_crosswalk(123, mapping) == "123" # No default = return original
|
|
assert apply_crosswalk(123, mapping, default="default") == "default" # 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") == "default"
|
|
|
|
|
|
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"]
|