[SPARK-53921][GEO][PYTHON][FOLLOWUP] Fix invalid algorithm error for Geography JSON parsing

### What changes were proposed in this pull request?
This PR follows up on #52627, and fixes an issue with `GeographyType` JSON parsing in PySpark. Also, this PR adds appropriate tests for JSON parsing, both for `GeographyType` and `GeometryType`.

### Why are the changes needed?
Fixing a wrong error class thrown in the JSON parsing method for `GeographyType`.

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

### How was this patch tested?
Added new test cases to:
- `pyspark.sql.tests.test_geographytype`
- `pyspark.sql.tests.test_geometrytype`

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

Closes #52772 from uros-db/geo-python-types-tests.

Authored-by: Uros Bojanic <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>

Author: uros-db

python/pyspark/sql/tests/test_geographytype.py

@@ -78,6 +78,47 @@ def test_geographytype_different_srid_values(self):
             geography_type_2 = GeographyType("ANY")
             self.assertNotEqual(geography_type_1.srid, geography_type_2.srid)
 
+    def test_geographytype_from_invalid_crs(self):
+        """Test that GeographyType construction fails when an invalid CRS is specified."""
+
+        for invalid_crs in ["srid", "any", "ogccrs84", "ogc:crs84", "ogc:CRS84", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeographyType._from_crs(invalid_crs, "SPHERICAL")
+            crs_header = "[ST_INVALID_CRS_VALUE] Invalid or unsupported CRS"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{crs_header} (coordinate reference system) value: '{invalid_crs}'.",
+            )
+
+    # The tests below verify GEOGRAPHY type JSON parsing based on CRS and algorithm.
+
+    def test_geographytype_from_invalid_algorithm(self):
+        """Test that GeographyType construction fails when an invalid CRS is specified."""
+
+        for invalid_alg in ["alg", "algorithm", "KARNEY", "spherical", "SPHEROID", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeographyType._from_crs("OGC:CRS84", invalid_alg)
+            alg_header = "[ST_INVALID_ALGORITHM_VALUE] Invalid or unsupported"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{alg_header} edge interpolation algorithm value: '{invalid_alg}'.",
+            )
+
+    def test_geographytype_from_valid_crs_and_algorithm(self):
+        """Test that GeographyType construction passes when valid CRS & ALG are specified."""
+
+        supported_crs = {
+            "OGC:CRS84": 4326,
+        }
+        for valid_crs, srid in supported_crs.items():
+            for valid_alg in ["SPHERICAL"]:
+                geography_type = GeographyType._from_crs(valid_crs, valid_alg)
+                self.assertEqual(geography_type.srid, srid)
+                self.assertEqual(geography_type.typeName(), "geography")
+                self.assertEqual(geography_type.simpleString(), f"geography({srid})")
+                self.assertEqual(geography_type.jsonValue(), f"geography({valid_crs}, {valid_alg})")
+                self.assertEqual(repr(geography_type), f"GeographyType({srid})")
+
 
 class GeographyTypeTest(GeographyTypeTestMixin, ReusedSQLTestCase):
     pass

python/pyspark/sql/tests/test_geometrytype.py

@@ -78,6 +78,34 @@ def test_geometrytype_different_srid_values(self):
             geometry_type_2 = GeometryType("ANY")
             self.assertNotEqual(geometry_type_1.srid, geometry_type_2.srid)
 
+    # The tests below verify GEOMETRY type JSON parsing based on the CRS value.
+
+    def test_geometrytype_from_invalid_crs(self):
+        """Test that GeometryType construction fails when an invalid CRS is specified."""
+
+        for invalid_crs in ["srid", "any", "ogccrs84", "ogc:crs84", "ogc:CRS84", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeometryType._from_crs(invalid_crs)
+            crs_header = "[ST_INVALID_CRS_VALUE] Invalid or unsupported CRS"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{crs_header} (coordinate reference system) value: '{invalid_crs}'.",
+            )
+
+    def test_geometrytype_from_valid_crs(self):
+        """Test that GeometryType construction passes when a valid CRS is specified."""
+
+        supported_crs = {
+            "OGC:CRS84": 4326,
+        }
+        for valid_crs, srid in supported_crs.items():
+            geometry_type = GeometryType._from_crs(valid_crs)
+            self.assertEqual(geometry_type.srid, srid)
+            self.assertEqual(geometry_type.typeName(), "geometry")
+            self.assertEqual(geometry_type.simpleString(), f"geometry({srid})")
+            self.assertEqual(geometry_type.jsonValue(), f"geometry({valid_crs})")
+            self.assertEqual(repr(geometry_type), f"GeometryType({srid})")
+
 
 class GeometryTypeTest(GeometryTypeTestMixin, ReusedSQLTestCase):
     pass

python/pyspark/sql/types.py

@@ -567,7 +567,7 @@ def _from_crs(cls, crs: str, alg: str) -> "GeographyType":
         # Algorithm value must be validated, although only SPHERICAL is supported currently.
         if alg != cls.DEFAULT_ALG:
             raise IllegalArgumentException(
-                errorClass="INVALID_ALGORITHM_VALUE",
+                errorClass="ST_INVALID_ALGORITHM_VALUE",
                 messageParameters={
                     "alg": str(alg),
                 },