Skip to content

test_prep_sap_general_material_data

Tests for prep_sap_general_material_data function.

test_prep_sap_general_material_data_column_renaming(spark)

Test prep_sap_general_material_data with column renaming.

Source code in code\tests\test_process_order\test_prep_sap_general_material_data.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def test_prep_sap_general_material_data_column_renaming(spark: SparkSession):
    """Test prep_sap_general_material_data with column renaming."""
    schema = StructType(
        [
            StructField("MATNR", StringType(), True),
            StructField("CUSTOM_GLOBAL", StringType(), True),
            StructField("NTGEW", DoubleType(), True),
            StructField("MTART", StringType(), True),
            StructField("BISMT", StringType(), True),
            StructField("LVORM", StringType(), True),
        ]
    )

    data = [("MAT1", "GMAT1", 10.5, "TYPE1", None, None)]

    df = spark.createDataFrame(data, schema)
    result = prep_sap_general_material_data(df, "CUSTOM_GLOBAL")

    assert "GLOBAL_MATERIAL_NUMBER" in result.columns
    assert result.first()["GLOBAL_MATERIAL_NUMBER"] == "GMAT1"

test_prep_sap_general_material_data_filtering(spark)

Test prep_sap_general_material_data with filtering.

Source code in code\tests\test_process_order\test_prep_sap_general_material_data.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def test_prep_sap_general_material_data_filtering(spark: SparkSession):
    """Test prep_sap_general_material_data with filtering."""
    schema = StructType(
        [
            StructField("MATNR", StringType(), True),
            StructField("GLOBAL_MAT", StringType(), True),
            StructField("NTGEW", DoubleType(), True),
            StructField("MTART", StringType(), True),
            StructField("BISMT", StringType(), True),
            StructField("LVORM", StringType(), True),
        ]
    )

    data = [
        ("MAT1", "GMAT1", 10.5, "TYPE1", None, None),  # Should be included
        ("MAT2", "GMAT2", 20.0, "TYPE2", "ARCHIVE", None),  # Should be excluded
        ("MAT3", "GMAT3", 30.0, "TYPE3", "DUPLICATE", ""),  # Should be excluded
        ("MAT4", "GMAT4", 40.0, "TYPE4", None, "X"),  # Should be excluded
    ]

    df = spark.createDataFrame(data, schema)
    result = prep_sap_general_material_data(df, "GLOBAL_MAT")

    assert result.count() == 1
    assert result.first()["MATNR"] == "MAT1"

test_prep_sap_general_material_data_invalid_inputs(spark)

Test prep_sap_general_material_data with invalid inputs.

Source code in code\tests\test_process_order\test_prep_sap_general_material_data.py 
80
81
82
83
84
85
86
87
88
89
def test_prep_sap_general_material_data_invalid_inputs(spark: SparkSession):
    """Test prep_sap_general_material_data with invalid inputs."""
    with pytest.raises(TypeError, match="Invalid input DataFrame"):
        prep_sap_general_material_data(None, "GLOBAL_MAT")

    with pytest.raises(TypeError, match="Invalid input DataFrame"):
        prep_sap_general_material_data("not a dataframe", "GLOBAL_MAT")

    with pytest.raises(TypeError, match="Invalid input column name"):
        prep_sap_general_material_data(spark.createDataFrame([], StructType([])), None)

test_prep_sap_general_material_data_valid_input(spark)

Test prep_sap_general_material_data with valid input data.

Source code in code\tests\test_process_order\test_prep_sap_general_material_data.py 
 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_sap_general_material_data_valid_input(spark: SparkSession):
    """Test prep_sap_general_material_data with valid input data."""
    schema = StructType(
        [
            StructField("MATNR", StringType(), True),
            StructField("GLOBAL_MAT", StringType(), True),
            StructField("NTGEW", DoubleType(), True),
            StructField("MTART", StringType(), True),
            StructField("BISMT", StringType(), True),
            StructField("LVORM", StringType(), True),
        ]
    )

    data = [("MAT1", "GMAT1", 10.5, "TYPE1", None, None), ("MAT2", "GMAT2", 20.0, "TYPE2", "VALID", "")]

    df = spark.createDataFrame(data, schema)
    result = prep_sap_general_material_data(df, "GLOBAL_MAT")

    assert result.count() == 2
    assert set(result.columns) == {"MATNR", "GLOBAL_MATERIAL_NUMBER", "NTGEW", "MTART"}
    first_row = result.first()
    assert first_row["MATNR"] == "MAT1"
    assert first_row["GLOBAL_MATERIAL_NUMBER"] == "GMAT1"