@@ -1195,3 +1195,69 @@ def test_to_metadata(self, Lake):
11951195
11961196 # Check that the spatial index was not duplicated
11971197 assert len (new_Lake .indexes ) == 1
1198+
1199+
1200+ class TestAsBinaryWKT :
1201+ def test_create_insert (self , conn , dialect_name ):
1202+ class GeometryWkt (Geometry ):
1203+ """Geometry type that uses WKT strings."""
1204+
1205+ from_text = "ST_GeomFromEWKT"
1206+ as_binary = "ST_AsText"
1207+ ElementType = WKTElement
1208+
1209+ dialects_with_srid = ["geopackage" , "mysql" , "mariadb" ]
1210+
1211+ # Define the table
1212+ cols = [
1213+ Column ("id" , Integer , primary_key = True ),
1214+ ]
1215+ cols .append (Column ("geom_with_srid" , GeometryWkt (geometry_type = "LINESTRING" , srid = 4326 )))
1216+ if dialect_name not in dialects_with_srid :
1217+ cols .append (Column ("geom" , GeometryWkt (geometry_type = "LINESTRING" )))
1218+ t = Table ("use_wkt" , MetaData (), * cols )
1219+
1220+ # Create the table
1221+ t .drop (bind = conn , checkfirst = True )
1222+ t .create (bind = conn )
1223+
1224+ # Test element insertion
1225+ inserted_values = [
1226+ {"geom_with_srid" : v }
1227+ for v in [
1228+ "SRID=4326;LINESTRING(0 0,1 1)" ,
1229+ WKTElement ("LINESTRING(0 0,2 2)" , srid = 4326 ),
1230+ WKTElement ("SRID=4326;LINESTRING(0 0,3 3)" , extended = True ),
1231+ from_shape (LineString ([[0 , 0 ], [4 , 4 ]]), srid = 4326 ),
1232+ ]
1233+ ]
1234+ if dialect_name not in dialects_with_srid :
1235+ for i , v in zip (
1236+ inserted_values ,
1237+ [
1238+ "LINESTRING(0 0,1 1)" ,
1239+ WKTElement ("LINESTRING(0 0,2 2)" ),
1240+ WKTElement ("SRID=-1;LINESTRING(0 0,3 3)" , extended = True ),
1241+ from_shape (LineString ([[0 , 0 ], [4 , 4 ]])),
1242+ ],
1243+ ):
1244+ i ["geom" ] = v
1245+
1246+ conn .execute (t .insert (), inserted_values )
1247+
1248+ results = conn .execute (t .select ())
1249+ rows = results .fetchall ()
1250+
1251+ for row_num , row in enumerate (rows ):
1252+ for num , element in enumerate (row [1 :]):
1253+ assert isinstance (element , WKTElement )
1254+ wkt = conn .execute (element .ST_AsText ()).scalar ()
1255+ assert format_wkt (wkt ) == f"LINESTRING(0 0,{ row_num + 1 } { row_num + 1 } )"
1256+ srid = conn .execute (element .ST_SRID ()).scalar ()
1257+ if num == 1 :
1258+ assert srid == 0 if dialect_name != "sqlite" else - 1
1259+ else :
1260+ assert srid == 4326
1261+
1262+ # Drop the table
1263+ t .drop (bind = conn )
0 commit comments