Skip to content

test_prep_plant_data_for_material

test_prep_plant_data_for_material_additional_fields(spark)

Test additional fields handling.

Parameters:

Name Type Description Default
spark SparkSession

Spark session object.

 required
Source code in code\tests\test_local_material\test_prep_plant_data_for_material.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def test_prep_plant_data_for_material_additional_fields(spark: SparkSession):
    """Test additional fields handling.

    Args:
        spark (SparkSession): Spark session object.
    """
    data = [
        ("SYS1", "10000001", "PLANT1", None, "CAT1"),
        ("SYS1", "10000002", "PLANT2", None, "CAT2"),
    ]
    schema = """
        SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING,
        LVORM STRING, CATEGORY STRING
    """
    input_df = spark.createDataFrame(data, schema=schema)

    expected_data = [
        ("SYS1", "10000001", "PLANT1", "CAT1"),
        ("SYS1", "10000002", "PLANT2", "CAT2"),
    ]
    expected_schema = """
        SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING,
        CATEGORY STRING
    """
    expected_df = spark.createDataFrame(expected_data, schema=expected_schema)

    result_df = prep_plant_data_for_material(
        input_df,
        check_deletion_flag_is_null=True,
        drop_duplicate_records=False,
        additional_fields=["CATEGORY"],
    )

    assertDataFrameEqual(result_df, expected_df)

test_prep_plant_data_for_material_deletion_flag(spark)

Test deletion flag filtering.

Parameters:

Name Type Description Default
spark SparkSession

Spark session object.

 required
Source code in code\tests\test_local_material\test_prep_plant_data_for_material.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def test_prep_plant_data_for_material_deletion_flag(spark: SparkSession):
    """Test deletion flag filtering.

    Args:
        spark (SparkSession): Spark session object.
    """
    data = [
        ("SYS1", "10000001", "PLANT1", None),
        ("SYS1", "10000002", "PLANT2", "X"),
    ]
    schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING, LVORM STRING"
    input_df = spark.createDataFrame(data, schema=schema)

    expected_data = [("SYS1", "10000001", "PLANT1")]
    expected_schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING"
    expected_df = spark.createDataFrame(expected_data, schema=expected_schema)

    result_df = prep_plant_data_for_material(
        input_df, check_deletion_flag_is_null=True, drop_duplicate_records=False
    )

    assertDataFrameEqual(result_df, expected_df)

test_prep_plant_data_for_material_duplicates(spark)

Test duplicate record handling.

Parameters:

Name Type Description Default
spark SparkSession

Spark session object.

 required
Source code in code\tests\test_local_material\test_prep_plant_data_for_material.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def test_prep_plant_data_for_material_duplicates(spark: SparkSession):
    """Test duplicate record handling.

    Args:
        spark (SparkSession): Spark session object.
    """
    data = [
        ("SYS1", "10000001", "PLANT1", None),
        ("SYS1", "10000001", "PLANT1", None),
    ]
    schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING, LVORM STRING"
    input_df = spark.createDataFrame(data, schema=schema)

    expected_data = [("SYS1", "10000001", "PLANT1")]
    expected_schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING"
    expected_df = spark.createDataFrame(expected_data, schema=expected_schema)

    result_df = prep_plant_data_for_material(
        input_df, check_deletion_flag_is_null=True, drop_duplicate_records=True
    )

    assertDataFrameEqual(result_df, expected_df)

test_prep_plant_data_for_material_invalid_input(spark)

Test with invalid input types.

Parameters:

Name Type Description Default
spark SparkSession

Spark session object.

 required
Source code in code\tests\test_local_material\test_prep_plant_data_for_material.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def test_prep_plant_data_for_material_invalid_input(spark: SparkSession):
    """Test with invalid input types.

    Args:
        spark (SparkSession): Spark session object.
    """
    with pytest.raises(TypeError, match="df must be a DataFrame"):
        prep_plant_data_for_material(
            None, check_deletion_flag_is_null=True, drop_duplicate_records=False
        )

    with pytest.raises(
        TypeError, match="check_deletion_flag_is_null must be a boolean"
    ):
        prep_plant_data_for_material(
            spark.createDataFrame([], "SOURCE_SYSTEM_ERP STRING"),
            check_deletion_flag_is_null="True",
            drop_duplicate_records=False,
        )

test_prep_plant_data_for_material_valid_input(spark)

Test with valid input and default parameters.

Parameters:

Name Type Description Default
spark SparkSession

Spark session object.

 required
Source code in code\tests\test_local_material\test_prep_plant_data_for_material.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def test_prep_plant_data_for_material_valid_input(spark: SparkSession):
    """Test with valid input and default parameters.

    Args:
        spark (SparkSession): Spark session object.
    """
    data = [
        ("SYS1", "10000001", "PLANT1", None),
        ("SYS1", "10000002", "PLANT2", None),
    ]
    schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING, LVORM STRING"
    input_df = spark.createDataFrame(data, schema=schema)

    expected_data = [
        ("SYS1", "10000001", "PLANT1"),
        ("SYS1", "10000002", "PLANT2"),
    ]
    expected_schema = "SOURCE_SYSTEM_ERP STRING, MATNR STRING, WERKS STRING"
    expected_df = spark.createDataFrame(expected_data, schema=expected_schema)

    result_df = prep_plant_data_for_material(
        input_df, check_deletion_flag_is_null=True, drop_duplicate_records=False
    )

    assertDataFrameEqual(result_df, expected_df)