mirror of
https://github.com/wassname/geopandas.git
synced 2026-09-20 12:50:47 +08:00
Adds tests for the read_postgis methods. A tests/util.py is introduced containing some of the common code for downloading the NYC boroughs file, managing the existence of psycopg2, and connecting to a database.
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
import unittest
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
|
|
import numpy as np
|
|
from shapely.geometry import Point, Polygon
|
|
|
|
|
|
from geopandas import GeoDataFrame, read_file
|
|
import tests.util
|
|
|
|
|
|
class TestDataFrame(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
N = 10
|
|
|
|
nybb_filename = tests.util.download_nybb()
|
|
|
|
self.df = read_file('/nybb_13a/nybb.shp', vfs='zip://' + nybb_filename)
|
|
self.tempdir = tempfile.mkdtemp()
|
|
self.boros = np.array(['Staten Island', 'Queens', 'Brooklyn',
|
|
'Manhattan', 'Bronx'])
|
|
self.crs = {'init': 'epsg:4326'}
|
|
self.df2 = GeoDataFrame([
|
|
{'geometry' : Point(x, y), 'value1': x + y, 'value2': x * y}
|
|
for x, y in zip(range(N), range(N))], crs=self.crs)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tempdir)
|
|
|
|
def test_df_init(self):
|
|
self.assertTrue(type(self.df2) is GeoDataFrame)
|
|
self.assertTrue(self.df2.crs == self.crs)
|
|
|
|
def test_to_json(self):
|
|
text = self.df.to_json()
|
|
data = json.loads(text)
|
|
self.assertTrue(data['type'] == 'FeatureCollection')
|
|
self.assertTrue(len(data['features']) == 5)
|
|
|
|
def test_copy(self):
|
|
df2 = self.df.copy()
|
|
self.assertTrue(type(df2) is GeoDataFrame)
|
|
self.assertEqual(self.df.crs, df2.crs)
|
|
|
|
def test_to_file(self):
|
|
""" Test to_file and from_file """
|
|
tempfilename = os.path.join(self.tempdir, 'boros.shp')
|
|
self.df.to_file(tempfilename)
|
|
# Read layer back in?
|
|
df = GeoDataFrame.from_file(tempfilename)
|
|
self.assertTrue('geometry' in df)
|
|
self.assertTrue(len(df) == 5)
|
|
self.assertTrue(np.alltrue(df['BoroName'].values == self.boros))
|
|
|
|
def test_mixed_types_to_file(self):
|
|
""" Test that mixed geometry types raise error when writing to file """
|
|
tempfilename = os.path.join(self.tempdir, 'test.shp')
|
|
s = GeoDataFrame({'geometry' : [Point(0, 0),
|
|
Polygon([(0, 0), (1, 0), (1, 1)])]})
|
|
with self.assertRaises(ValueError):
|
|
s.to_file(tempfilename)
|
|
|
|
def test_bool_index(self):
|
|
# Find boros with 'B' in their name
|
|
df = self.df[self.df['BoroName'].str.contains('B')]
|
|
self.assertTrue(len(df) == 2)
|
|
boros = df['BoroName'].values
|
|
self.assertTrue('Brooklyn' in boros)
|
|
self.assertTrue('Bronx' in boros)
|
|
self.assertTrue(type(df) is GeoDataFrame)
|
|
|
|
def test_transform(self):
|
|
df2 = self.df2.copy()
|
|
df2.crs = {'init': 'epsg:26918', 'no_defs': True}
|
|
lonlat = df2.to_crs(epsg=4326)
|
|
utm = lonlat.to_crs(epsg=26918)
|
|
self.assertTrue(all(df2['geometry'].almost_equals(utm['geometry'], decimal=2)))
|
|
|
|
def test_from_postgis_default(self):
|
|
con = tests.util.connect('test_geopandas')
|
|
if con is None or not tests.util.create_db(self.df):
|
|
raise unittest.case.SkipTest()
|
|
|
|
try:
|
|
sql = "SELECT * FROM nybb;"
|
|
df = GeoDataFrame.from_postgis(sql, con)
|
|
finally:
|
|
con.close()
|
|
|
|
tests.util.validate_boro_df(self, df)
|
|
|
|
def test_from_postgis_custom_geom_col(self):
|
|
con = tests.util.connect('test_geopandas')
|
|
if con is None or not tests.util.create_db(self.df):
|
|
raise unittest.case.SkipTest()
|
|
|
|
try:
|
|
sql = """SELECT
|
|
borocode, boroname, shape_leng, shape_area,
|
|
geom AS __geometry__
|
|
FROM nybb;"""
|
|
df = GeoDataFrame.from_postgis(sql, con, geom_col='__geometry__')
|
|
finally:
|
|
con.close()
|
|
|
|
tests.util.validate_boro_df(self, df)
|