hangshuo652
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import sys, os
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
|
from comparator.aligner import align_records
|
|
|
|
|
|
def test_align_by_key():
|
|
c = [{"CUST-ID": "C001", "AMT": 100}, {"CUST-ID": "C002", "AMT": 200}]
|
|
s = [{"CUST-ID": "C002", "AMT": 200}, {"CUST-ID": "C001", "AMT": 100}]
|
|
result = align_records(c, s, key_field="CUST-ID")
|
|
assert len(result) == 2
|
|
assert all(st == "MATCHED" for _, _, st in result)
|
|
|
|
|
|
def test_missing_in_spark():
|
|
c = [{"CUST-ID": "C001"}, {"CUST-ID": "C002"}]
|
|
s = [{"CUST-ID": "C001"}]
|
|
result = align_records(c, s, key_field="CUST-ID")
|
|
assert "MISSING_IN_SPARK" in [st for _, _, st in result]
|
|
|
|
|
|
def test_extra_in_spark():
|
|
c = [{"CUST-ID": "C001"}]
|
|
s = [{"CUST-ID": "C001"}, {"CUST-ID": "C002"}]
|
|
result = align_records(c, s, key_field="CUST-ID")
|
|
assert "EXTRA_IN_SPARK" in [st for _, _, st in result]
|
|
|
|
|
|
def test_empty_inputs():
|
|
assert align_records([], [], "key") == []
|
|
|
|
|
|
def test_duplicate_keys():
|
|
c = [{"ID": "K1", "V": 1}, {"ID": "K1", "V": 2}]
|
|
s = [{"ID": "K1", "V": 1}, {"ID": "K1", "V": 2}]
|
|
assert len(align_records(c, s, key_field="ID")) == 2
|
|
|
|
|
|
def test_none_key():
|
|
assert len(align_records([{"ID": None, "V": 1}], [{"ID": None, "V": 1}], "ID")) == 1
|