VenkataKrishna Reddy Avula
a5f54a2a48
Feature/daip2- 17 - Aarete derived field values and crosswalk mapping * fix pipeline error * fix pipeline error * fix pipeline error * fix pipeline error * fix pipeline error * Update fieldSet unit test * Update test_from_values * Fix circular import * Added NLTK * Add NLTK to path * Update gitignore * included txt files for nltk_data * move nltk_data to src * Fix last upload count * Last upload count bugfix * Fixed B processing * Remove client-specific postprocessing * split consolidate_output * aarete derived AC fields mapping * chnages reverted * AC Croasswalk Mapping * Merge branch 'main' into feature/daip2-17 * Update poetry.lock * Add init for crosswalk * Complete paths Approved-by: Alex Galarce
144 lines
4.1 KiB
Python
144 lines
4.1 KiB
Python
import json
|
|
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from crosswalk.crosswalk_utils import CrosswalkBuilder, 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"}
|