Refining example, add utilities, and fix xdist test error

feathr_project/feathr/client.py

@@ -415,7 +415,6 @@ def get_offline_features(self,
                              output_path: Union[str, Sink],
                              execution_configurations: Union[SparkExecutionConfiguration ,Dict[str,str]] = {},
                              config_file_name:str = "feature_join_conf/feature_join.conf",
-                             udf_files = None,
                              verbose: bool = False
                              ):
         """
@@ -609,7 +608,7 @@ def _valid_materialize_keys(self, features: List[str], allow_empty_key=False):
                         self.logger.error(f"Inconsistent feature keys. Current keys are {str(keys)}")
                         return False
         return True
-    
+
     def materialize_features(self, settings: MaterializationSettings, execution_configurations: Union[SparkExecutionConfiguration ,Dict[str,str]] = {}, verbose: bool = False, allow_materialize_non_agg_feature: bool = False):
         """Materialize feature data
 
@@ -621,7 +620,7 @@ def materialize_features(self, settings: MaterializationSettings, execution_conf
         feature_list = settings.feature_names
         if len(feature_list) > 0 and not self._valid_materialize_keys(feature_list):
             raise RuntimeError(f"Invalid materialization features: {feature_list}, since they have different keys. Currently Feathr only supports materializing features of the same keys.")
-        
+
         if not allow_materialize_non_agg_feature:
             # Check if there are non-aggregation features in the list
             for fn in feature_list:

feathr_project/feathr/udf/_preprocessing_pyudf_manager.py

@@ -176,6 +176,7 @@ def prepare_pyspark_udf_files(feature_names: List[str], local_workspace_dir):
         for feature_name in feature_names:
             if feature_name in features_with_preprocessing:
                 has_py_udf_preprocessing = True
+                break
 
         if has_py_udf_preprocessing:
             pyspark_driver_path = os.path.join(local_workspace_dir, FEATHR_PYSPARK_DRIVER_FILE_NAME)

feathr_project/test/unit/utils/test_config.py

@@ -6,26 +6,38 @@
 from feathr.utils.config import FEATHR_CONFIG_TEMPLATE, generate_config
 
 
-@pytest.mark.parametrize(
-    "output_filepath", [None, NamedTemporaryFile().name],
-)
-def test__generate_config(output_filepath: str):
-
-    config = FEATHR_CONFIG_TEMPLATE.format(
+@pytest.fixture(scope="session")
+def feathr_config_str() -> str:
+    return FEATHR_CONFIG_TEMPLATE.format(
         resource_prefix="test_prefix",
         project_name="test_project",
         spark_cluster="local",
     )
 
+
+@pytest.mark.parametrize(
+    "output_filepath", [None, "config.yml"],
+)
+def test__generate_config(
+    output_filepath: str,
+    feathr_config_str: str,
+    tmp_path: Path,
+):
+    # Use tmp_path so that the test files get cleaned up after the tests
+    if output_filepath:
+        output_filepath = str(tmp_path / output_filepath)
+
     config_filepath = generate_config(
         resource_prefix="test_prefix",
         project_name="test_project",
         spark_cluster="local",
         output_filepath=output_filepath,
     )
 
+    # Assert if the config file was generated in the specified output path.
     if output_filepath:
         assert output_filepath == config_filepath
 
+    # Assert the generated config string is correct.
     with open(config_filepath, "r") as f:
-        assert config == f.read()
+        assert feathr_config_str == f.read()