def verify_online_record()

in pyspark-sdk/integration_test/BatchIngestionTest.py [0:0]

• 10 lines of code
• 3 McCabe index (conditional complexity)

def verify_online_record(ingested_row: Row, record_dict: dict):
    ingested_row_dict = ingested_row.asDict()
    for key in ingested_row_dict.keys():
        ingested_value = ingested_row_dict.get(key, None)
        filterd_record_list = list(filter(lambda feature_value: feature_value["FeatureName"] == key, record_dict))
        if ingested_value is not None:
            filterd_record = filterd_record_list[0]
            tc.assertEqual(str(ingested_row_dict[key]), filterd_record["ValueAsString"])
        else:
            tc.assertEqual(len(filterd_record_list), 0)