[SPARK-46044][PYTHON][TESTS] Improve test coverage of udf.py

### What changes were proposed in this pull request?
Improve test coverage of udf.py

### Why are the changes needed?
Subtasks of [SPARK-46041](https://issues.apache.org/jira/browse/SPARK-46041) to improve test coverage

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Test changes only.

### Was this patch authored or co-authored using generative AI tooling?
No

Closes apache#43947 from xinrong-meng/test_udf.

Authored-by: Xinrong Meng <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

python/pyspark/sql/tests/connect/test_parity_udf.py

@@ -86,6 +86,12 @@ def test_non_existed_udf_with_sql_context(self):
     def test_udf_registration_returns_udf_on_sql_context(self):
         super().test_udf_registration_returns_udf_on_sql_context()
 
+    def test_err_udf_registration(self):
+        self.check_err_udf_registration()
+
+    def test_err_udf_init(self):
+        self.check_err_udf_init()
+
 
 if __name__ == "__main__":
     import unittest

python/pyspark/sql/tests/test_udf.py

@@ -377,12 +377,17 @@ def test_udf_with_order_by_and_limit(self):
     def test_udf_registration_returns_udf(self):
         df = self.spark.range(10)
         add_three = self.spark.udf.register("add_three", lambda x: x + 3, IntegerType())
-
         self.assertListEqual(
             df.selectExpr("add_three(id) AS plus_three").collect(),
             df.select(add_three("id").alias("plus_three")).collect(),
         )
 
+        add_three_str = self.spark.udf.register("add_three_str", lambda x: x + 3)
+        self.assertListEqual(
+            df.selectExpr("add_three_str(id) AS plus_three").collect(),
+            df.select(add_three_str("id").alias("plus_three")).collect(),
+        )
+
     def test_udf_registration_returns_udf_on_sql_context(self):
         df = self.spark.range(10)
 
@@ -425,6 +430,20 @@ def test_register_java_udaf(self):
         ).first()
         self.assertEqual(row.asDict(), Row(name="b", avg=102.0).asDict())
 
+    def test_err_udf_registration(self):
+        with QuietTest(self.sc):
+            self.check_err_udf_registration()
+
+    def check_err_udf_registration(self):
+        with self.assertRaises(PySparkTypeError) as pe:
+            self.spark.udf.register("f", UserDefinedFunction("x", StringType()), "int")
+
+        self.check_error(
+            exception=pe.exception,
+            error_class="NOT_CALLABLE",
+            message_parameters={"arg_name": "func", "arg_type": "str"},
+        )
+
     def test_non_existed_udf(self):
         spark = self.spark
         self.assertRaisesRegex(
@@ -1027,6 +1046,38 @@ def test_python_udf_segfault(self):
 
                 self.spark.range(1).select(udf(lambda x: ctypes.string_at(0))("id")).collect()
 
+    def test_err_udf_init(self):
+        with QuietTest(self.sc):
+            self.check_err_udf_init()
+
+    def check_err_udf_init(self):
+        with self.assertRaises(PySparkTypeError) as pe:
+            UserDefinedFunction("x", StringType())
+
+        self.check_error(
+            exception=pe.exception,
+            error_class="NOT_CALLABLE",
+            message_parameters={"arg_name": "func", "arg_type": "str"},
+        )
+
+        with self.assertRaises(PySparkTypeError) as pe:
+            UserDefinedFunction(lambda x: x, 1)
+
+        self.check_error(
+            exception=pe.exception,
+            error_class="NOT_DATATYPE_OR_STR",
+            message_parameters={"arg_name": "returnType", "arg_type": "int"},
+        )
+
+        with self.assertRaises(PySparkTypeError) as pe:
+            UserDefinedFunction(lambda x: x, StringType(), evalType="SQL_BATCHED_UDF")
+
+        self.check_error(
+            exception=pe.exception,
+            error_class="NOT_INT",
+            message_parameters={"arg_name": "evalType", "arg_type": "str"},
+        )
+
 
 class UDFTests(BaseUDFTestsMixin, ReusedSQLTestCase):
     @classmethod