94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""Tests for the column mapping functionality."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pandas as pd
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
# Add meta_files to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "meta_files"))
|
|
from generate_config import apply_column_mapping, get_canonical_schemas
|
|
|
|
|
|
def test_apply_column_mapping():
|
|
"""Test that column mapping correctly renames columns."""
|
|
# Create test DataFrame with instrument-specific column names
|
|
test_df = pd.DataFrame(
|
|
{
|
|
"Old_Time_Sec": [1.0, 2.0, 3.0],
|
|
"Old_Particle_Flags": [0, 1, 0],
|
|
"Other_Column": ["a", "b", "c"],
|
|
}
|
|
)
|
|
|
|
# Create mapping from canonical to file column names
|
|
test_mapping = {
|
|
"Time (sec)": "Old_Time_Sec",
|
|
"Particle Flags": "Old_Particle_Flags",
|
|
"Nonexistent Column": "YOUR_COLUMN_NAME_FOR_NONEXISTENT",
|
|
"Missing Column": None,
|
|
}
|
|
|
|
result = apply_column_mapping(test_df, test_mapping)
|
|
|
|
# Check that columns were renamed correctly
|
|
assert "Time (sec)" in result.columns
|
|
assert "Particle Flags" in result.columns
|
|
assert "Other_Column" in result.columns # Unmapped columns should remain
|
|
|
|
# Check that placeholder and None mappings were ignored
|
|
assert "YOUR_COLUMN_NAME_FOR_NONEXISTENT" not in result.columns
|
|
assert "Nonexistent Column" not in result.columns
|
|
|
|
# Check that data was preserved
|
|
assert result["Time (sec)"].tolist() == [1.0, 2.0, 3.0]
|
|
assert result["Particle Flags"].tolist() == [0, 1, 0]
|
|
|
|
|
|
def test_get_canonical_schemas():
|
|
"""Test that canonical schemas contain expected columns."""
|
|
schemas = get_canonical_schemas()
|
|
|
|
# Check that both PbP and HK canonical schemas exist
|
|
assert "pbp_canonical" in schemas
|
|
assert "hk_canonical" in schemas
|
|
|
|
# Check for key columns that should be present
|
|
pbp_canonical = schemas["pbp_canonical"]
|
|
assert "Time (sec)" in pbp_canonical
|
|
assert "Particle Flags" in pbp_canonical
|
|
assert "Incand Mass (fg)" in pbp_canonical
|
|
|
|
hk_canonical = schemas["hk_canonical"]
|
|
assert "Time Stamp" in hk_canonical
|
|
assert "Time (sec)" in hk_canonical
|
|
assert "Sample Flow Controller Read (sccm)" in hk_canonical
|
|
|
|
|
|
def test_column_mapping_empty_dataframe():
|
|
"""Test column mapping with empty DataFrame."""
|
|
empty_df = pd.DataFrame()
|
|
mapping = {"Time (sec)": "Old_Time_Sec"}
|
|
|
|
result = apply_column_mapping(empty_df, mapping)
|
|
assert result.empty
|
|
|
|
|
|
def test_column_mapping_no_mapping():
|
|
"""Test column mapping when no mapping is provided."""
|
|
test_df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]})
|
|
|
|
result = apply_column_mapping(test_df, {})
|
|
|
|
# Should return original DataFrame unchanged
|
|
pd.testing.assert_frame_equal(result, test_df)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_apply_column_mapping()
|
|
test_get_canonical_schemas()
|
|
test_column_mapping_empty_dataframe()
|
|
test_column_mapping_no_mapping()
|
|
print("All tests passed!")
|