[SPARK-46076][PYTHON][TESTS] Remove unittest deprecated alias usage for Python 3.12

### What changes were proposed in this pull request?

This PR aims to remove `unittest` alias usage for Python 3.12.

Currently, it fails like the following.
- https://github.com/apache/spark/actions/runs/6971394284/job/18971420822
```
======================================================================
ERROR [0.554s]: test_find_spark_home (pyspark.tests.test_util.UtilTests.test_find_spark_home)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/__w/spark/spark/python/pyspark/tests/test_util.py", line 83, in test_find_spark_home
    self.assertEquals(origin, _find_spark_home())
    ^^^^^^^^^^^^^^^^^
AttributeError: 'UtilTests' object has no attribute 'assertEquals'. Did you mean: 'assertEqual'?
```

### Why are the changes needed?

Python 3.12 removes the following deprecated aliases.

- https://docs.python.org/3/whatsnew/3.12.html#id3

<img width="802" alt="Screenshot 2023-11-23 at 12 52 33 PM" src="https://github.com/apache/spark/assets/9700541/0158c1a4-fcfc-4a02-85c5-7fcbd6c6a034">

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

No.

### How was this patch tested?

Pass the CIs with Python 3.9.

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

No.

Closes #43986 from dongjoon-hyun/SPARK-46076.

Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>

Author: dongjoon-hyun

python/pyspark/ml/tests/test_linalg.py

@@ -362,7 +362,7 @@ def test_unwrap_udt(self):
             Row(v2=unwrapped_vec(1, None, None, [1.0, 2.0, 3.0])),
             Row(v2=unwrapped_vec(0, 3, [1, 2], [1.0, 5.5])),
         ]
-        self.assertEquals(results, expected)
+        self.assertEqual(results, expected)
 
 
 class MatrixUDTTests(MLlibTestCase):

python/pyspark/pandas/tests/indexes/test_base.py

@@ -62,11 +62,11 @@ def test_index_basic(self):
             self.assert_eq(psdf.index.dtype, pdf.index.dtype)
 
         self.assert_eq(ps.Index([])._summary(), "Index: 0 entries")
-        with self.assertRaisesRegexp(ValueError, "The truth value of a Index is ambiguous."):
+        with self.assertRaisesRegex(ValueError, "The truth value of a Index is ambiguous."):
             bool(ps.Index([1]))
-        with self.assertRaisesRegexp(TypeError, "Index.name must be a hashable type"):
+        with self.assertRaisesRegex(TypeError, "Index.name must be a hashable type"):
             ps.Index([1, 2, 3], name=[(1, 2, 3)])
-        with self.assertRaisesRegexp(TypeError, "Index.name must be a hashable type"):
+        with self.assertRaisesRegex(TypeError, "Index.name must be a hashable type"):
             ps.Index([1.0, 2.0, 3.0], name=[(1, 2, 3)])
 
     def test_index_from_series(self):

python/pyspark/pandas/tests/indexes/test_category.py

@@ -68,9 +68,9 @@ def test_categorical_index(self):
         self.assert_eq(psidx.codes, pd.Index(pidx.codes))
         self.assert_eq(psidx.ordered, pidx.ordered)
 
-        with self.assertRaisesRegexp(TypeError, "Index.name must be a hashable type"):
+        with self.assertRaisesRegex(TypeError, "Index.name must be a hashable type"):
             ps.CategoricalIndex([1, 2, 3], name=[(1, 2, 3)])
-        with self.assertRaisesRegexp(
+        with self.assertRaisesRegex(
             TypeError, "Cannot perform 'all' with this index type: CategoricalIndex"
         ):
             ps.CategoricalIndex([1, 2, 3]).all()

python/pyspark/pandas/tests/indexes/test_datetime.py

@@ -64,9 +64,9 @@ def _disallow_nanoseconds(self, f):
         self.assertRaises(ValueError, lambda: f(freq="N"))
 
     def test_datetime_index(self):
-        with self.assertRaisesRegexp(TypeError, "Index.name must be a hashable type"):
+        with self.assertRaisesRegex(TypeError, "Index.name must be a hashable type"):
             ps.DatetimeIndex(["2004-01-01", "2002-12-31", "2000-04-01"], name=[(1, 2)])
-        with self.assertRaisesRegexp(
+        with self.assertRaisesRegex(
             TypeError, "Cannot perform 'all' with this index type: DatetimeIndex"
         ):
             ps.DatetimeIndex(["2004-01-01", "2002-12-31", "2000-04-01"]).all()

python/pyspark/pandas/tests/indexes/test_timedelta.py

@@ -90,9 +90,9 @@ def test_timedelta_index(self):
         )
 
         # ps.TimedeltaIndex(ps.Index([1, 2, 3]))
-        with self.assertRaisesRegexp(TypeError, "Index.name must be a hashable type"):
+        with self.assertRaisesRegex(TypeError, "Index.name must be a hashable type"):
             ps.TimedeltaIndex([timedelta(1), timedelta(microseconds=2)], name=[(1, 2)])
-        with self.assertRaisesRegexp(
+        with self.assertRaisesRegex(
             TypeError, "Cannot perform 'all' with this index type: TimedeltaIndex"
         ):
             psidx.all()

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

@@ -1824,7 +1824,7 @@ def test_observe(self):
 
         self.assert_eq(cdf, df)
 
-        self.assertEquals(cobservation.get, observation.get)
+        self.assertEqual(cobservation.get, observation.get)
 
         observed_metrics = cdf.attrs["observed_metrics"]
         self.assert_eq(len(observed_metrics), 1)
@@ -3449,11 +3449,11 @@ def test_can_create_multiple_sessions_to_different_remotes(self):
         self.assertIsNotNone(self.spark._client)
         # Creates a new remote session.
         other = PySparkSession.builder.remote("sc://other.remote:114/").create()
-        self.assertNotEquals(self.spark, other)
+        self.assertNotEqual(self.spark, other)
 
         # Gets currently active session.
         same = PySparkSession.builder.remote("sc://other.remote.host:114/").getOrCreate()
-        self.assertEquals(other, same)
+        self.assertEqual(other, same)
         same.release_session_on_close = False  # avoid sending release to dummy connection
         same.stop()
 

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

@@ -379,7 +379,7 @@ def test_simple_binary_expressions(self):
         self.assertEqual(len(pdf.index), 4)
 
         res = pd.DataFrame(data={"id": [0, 30, 60, 90]})
-        self.assert_(pdf.equals(res), f"{pdf.to_string()} != {res.to_string()}")
+        self.assertTrue(pdf.equals(res), f"{pdf.to_string()} != {res.to_string()}")
 
     def test_literal_with_acceptable_type(self):
         for value, dataType in [

python/pyspark/sql/tests/pandas/test_pandas_map.py

@@ -394,7 +394,7 @@ def func(iterator):
 
             for offheap in ["true", "false"]:
                 with self.sql_conf({"spark.sql.columnVector.offheap.enabled": offheap}):
-                    self.assertEquals(
+                    self.assertEqual(
                         self.spark.read.parquet(path).mapInPandas(func, "id long").head(), Row(0)
                     )
         finally:

python/pyspark/sql/tests/pandas/test_pandas_udf_scalar.py

@@ -181,7 +181,7 @@ def test_input_nested_structs(self):
 
         mirror = pandas_udf(lambda s: s, df.dtypes[0][1])
 
-        self.assertEquals(
+        self.assertEqual(
             df.select(mirror(df.struct).alias("res")).first(),
             Row(
                 res=Row(
@@ -194,13 +194,13 @@ def test_input_nested_maps(self):
         df = self.df_with_nested_maps
 
         str_repr = pandas_udf(lambda s: s.astype(str), StringType())
-        self.assertEquals(
+        self.assertEqual(
             df.select(str_repr(df.attributes).alias("res")).first(),
             Row(res="{'personal': {'name': 'John', 'city': 'New York'}}"),
         )
 
         extract_name = pandas_udf(lambda s: s.apply(lambda x: x["personal"]["name"]), StringType())
-        self.assertEquals(
+        self.assertEqual(
             df.select(extract_name(df.attributes).alias("res")).first(),
             Row(res="John"),
         )
@@ -209,7 +209,7 @@ def test_input_nested_arrays(self):
         df = self.df_with_nested_arrays
 
         str_repr = pandas_udf(lambda s: s.astype(str), StringType())
-        self.assertEquals(
+        self.assertEqual(
             df.select(str_repr(df.nested_array).alias("res")).first(),
             Row(res="[array([1, 2, 3], dtype=int32) array([4, 5], dtype=int32)]"),
         )
@@ -1450,9 +1450,7 @@ def udf(x):
 
             for offheap in ["true", "false"]:
                 with self.sql_conf({"spark.sql.columnVector.offheap.enabled": offheap}):
-                    self.assertEquals(
-                        self.spark.read.parquet(path).select(udf("id")).head(), Row(0)
-                    )
+                    self.assertEqual(self.spark.read.parquet(path).select(udf("id")).head(), Row(0))
         finally:
             shutil.rmtree(path)
 

python/pyspark/sql/tests/streaming/test_streaming.py

@@ -36,7 +36,7 @@ def test_streaming_query_functions_basic(self):
             .start()
         )
         try:
-            self.assertEquals(query.name, "test_streaming_query_functions_basic")
+            self.assertEqual(query.name, "test_streaming_query_functions_basic")
             self.assertTrue(isinstance(query.id, str))
             self.assertTrue(isinstance(query.runId, str))
             self.assertTrue(query.isActive)