WIP-pyshapelib-bramz/test/test_postgis_db.py

# Copyright (C) 2003, 2004, 2005, 2006 by Intevation GmbH
# Authors:
# Bernhard Herzog <[email protected]>
#
# This program is free software under the GPL (>=v2)
# Read the file COPYING coming with the software for details.

"""Test for Thuban.Model.postgisdb"""

import os
import unittest


try:
    import psycopg
except ImportError:
    # No psycopg available. Nothing to be done here because the
    # postgis.py support module determines this too and the tests will
    # be skipped completely.
    pass

import postgissupport

import support
support.initthuban()

from Thuban.Model.postgisdb import ConnectionError, PostGISConnection, \
     PostGISTable, PostGISShapeStore
from Thuban.Model.table import FIELDTYPE_INT, FIELDTYPE_STRING, \
     FIELDTYPE_DOUBLE
from Thuban.Model.data import SHAPETYPE_POINT, SHAPETYPE_POLYGON, \
     SHAPETYPE_ARC, RAW_WKT


class NonConnection(PostGISConnection):

    """Special connection class that doesn't actually connect"""

    def connect(self):
        pass


class TestPostGISSimple(unittest.TestCase):

    """Some simple PostGISConnection tests that don't need a real connection"""

    def test_brief_description(self):
        """Test PostGISConnection.BriefDescription()"""
        self.assertEquals(NonConnection("somedb").BriefDescription(),
                          "postgis://@:/somedb")
        self.assertEquals(NonConnection("db", host="here",
                                        port="123").BriefDescription(),
                          "postgis://@here:123/db")
        self.assertEquals(NonConnection("db", user="me",
                                        port="123").BriefDescription(),
                          "postgis://me@:123/db")

    def test_matches_parameters(self):
        """Test PostGISConnection.MatchesParameters()"""
        def do_test(testfn, conn, host = "", port="", dbname = "", user = ""):
            testfn(conn.MatchesParameters({"host": host, "port": port,
                                           "dbname": dbname, "user": user}))

        do_test(self.assert_, NonConnection("somedb"),
                dbname="somedb")
        do_test(self.assert_, NonConnection("somedb", host="here"),
                dbname="somedb", host="here")
        do_test(self.assert_, NonConnection("db", user="me", port="123"),
                dbname="db", port="123", user="me")

        do_test(self.failIf, NonConnection("somedb"),
                dbname="someotherdb")
        do_test(self.failIf, NonConnection("somedb", host="here"),
                dbname="somedb", host="here", port="765")
        do_test(self.failIf, NonConnection("db", user="me", port="123"),
                dbname="db", port="123", user="you")


class TestPostGISConnection(unittest.TestCase):

    def setUp(self):
        """Start the server and create a database.

        The database name will be stored in self.dbname, the server
        object in self.server and the db object in self.db.
        """
        postgissupport.skip_if_no_postgis()
        self.server = postgissupport.get_test_server()
        self.dbname = ".".join(self.id().split(".")[-2:])[-31:]
        self.db = self.server.new_postgis_db(self.dbname)

    def test_gis_tables_empty(self):
        """Test PostGISConnection.GISTables() on empty DB"""
        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))

        # An empty database doesn't have any GIS tables
        self.assertEquals(db.GeometryTables(), [])

    def test_gis_tables_with_views_and_tables(self):
        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))

        conn = psycopg.connect("dbname=%s " % self.dbname
                               + self.server.connection_string("admin"))
        cursor = conn.cursor()
        # First a normal table, i.e. one without a geometry column
        # we need an explicit WITH OIDS for postgresql >= 8.0
        cursor.execute("CREATE TABLE normal (A INT, B VARCHAR) WITH OIDS;"
                       "GRANT SELECT ON normal TO PUBLIC;")

        # Then a table with a geometry column
        # we need an explicit WITH OIDS for postgresql >= 8.0
        cursor.execute("CREATE TABLE geo (A INT) WITH OIDS;"
                       "GRANT SELECT ON geo TO PUBLIC;")
        cursor.execute("SELECT AddGeometryColumn(%(dbname)s, 'geo',"
                       " 'the_geom', -1, 'POINT', 2);",
                       {"dbname": self.dbname})

        # And create a view containing a geometry column
        cursor.execute("CREATE VIEW my_view AS"
                       " (SELECT * FROM geo WHERE a > 100);"
                       "GRANT SELECT ON my_view TO PUBLIC;")
        conn.commit()

        self.assertEquals(db.GeometryTables(), ["geo", "my_view"])
        self.assertEquals(db.table_columns("geo"),
                          [("oid", FIELDTYPE_INT), ("a", FIELDTYPE_INT),
                           ("the_geom", "geometry")])
        self.assertEquals(db.table_columns("my_view"),
                          [("a", FIELDTYPE_INT), ("the_geom", "geometry")])
        self.assertEquals(db.table_columns("normal"),
                          [("oid", FIELDTYPE_INT), ("a", FIELDTYPE_INT),
                           ("b", FIELDTYPE_STRING)])


class TestPostgisDBExceptions(unittest.TestCase):

    def setUp(self):
        """Start the postgis server and switch on authentication"""
        postgissupport.skip_if_no_postgis()
        self.server = postgissupport.get_test_server()
        self.postgisdb = self.server.get_default_static_data_db()
        self.server.require_authentication(True)

    def tearDown(self):
        """Extend the inherited method to switch off postgresql authentication
        """
        self.server.require_authentication(False)

    def test_no_password(self):
        """Test PostGISConnection with omitted but required password"""
        connection_params = self.server.connection_params("user")
        # remove the password deliberately
        del connection_params["password"]

        self.assertRaises(ConnectionError,
                          PostGISConnection, dbname = self.postgisdb.dbname,
                          **connection_params)


class TestPostGISSpecialCases(unittest.TestCase):

    """Tests for special cases of PostGIS table usage"""

    def setUp(self):
        """Start the server and create a database.

        The database name will be stored in self.dbname, the server
        object in self.server and the db object in self.db.
        """
        postgissupport.skip_if_no_postgis()
        self.server = postgissupport.get_test_server()
        self.dbname = ".".join(self.id().split(".")[-2:])[-31:]
        self.db = self.server.new_postgis_db(self.dbname)

    def test_unsupported_types(self):
        """test PostGISTable on a table with unsupported data types"""
        stmt = """CREATE TABLE foo (
                        gid integer,
                        ignored bigint,
                        length float);
                  GRANT SELECT ON foo TO PUBLIC;
                  """
        self.server.execute_sql(self.dbname, "admin", stmt)

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        table = PostGISTable(db, "foo")

        # The bigint column will be ignored because it's not mapped to a
        # known integer type, so there are only two colunns
        self.assertEquals(table.NumColumns(), 2)
        self.assertEquals(table.Column(0).name, "gid")
        self.assertEquals(table.Column(1).name, "length")

    def test_table_name_quoting(self):
        """Test whether PostGISTable quotes table names properly"""
        stmt = '''CREATE TABLE "name with ""quotes"" and spaces" (gid integer);
             GRANT SELECT ON "name with ""quotes"" and spaces" TO PUBLIC;'''
        self.server.execute_sql(self.dbname, "admin", stmt)

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        table = PostGISTable(db, 'name with "quotes" and spaces')

        self.assertEquals(table.NumColumns(), 1)
        self.assertEquals(table.Column(0).name, "gid")

    def test_column_name_quoting(self):
        """Test whether PostGISTable quotes column names properly"""
        stmt = '''CREATE TABLE unusual_column_names (
                     gid integer,
                     "with space" float,
                     "with "" quote" integer);
             INSERT INTO unusual_column_names VALUES (1, 1.0, 0);
             INSERT INTO unusual_column_names VALUES (2, 2.0, 1);
             INSERT INTO unusual_column_names VALUES (3, 3.0, 1);
             GRANT SELECT ON unusual_column_names TO PUBLIC;'''
        self.server.execute_sql(self.dbname, "admin", stmt)

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        table = PostGISTable(db, 'unusual_column_names')

        self.assertEquals(table.NumColumns(), 3)
        self.assertEquals(table.Column(0).name, "gid")
        self.assertEquals(table.Column(1).name, "with space")
        self.assertEquals(table.Column(2).name, "with \" quote")

        # do some queries where the names have to be quoted
        self.assertEquals(table.ReadRowAsDict(1),
                          {"gid": 1, "with space": 1.0, "with \" quote": 0})
        self.assertEquals(table.ReadValue(2, 1, row_is_ordinal = True), 3.0)
        self.assertEquals(table.ReadValue(2, 1, row_is_ordinal = False), 2.0)
        self.assertEquals(table.ValueRange("with space"), (1.0, 3.0))

        # The return value of UniqueValues is unsorted, so we need to
        # sort it for comparison.
        unique_values = table.UniqueValues("with \" quote")
        unique_values.sort()
        self.assertEquals(unique_values, [0, 1])
        self.assertEquals(table.SimpleQuery(table.Columns()[2], "==", 1),
                          [2, 3])
        self.assertEquals(table.SimpleQuery(table.Columns()[0], "==",
                                            table.Columns()[1]), [1, 2, 3])

    def test_shapestore_name_quoting(self):
        """Test whether PostGISShapeStore quotes table names properly"""
        postgissupport.skip_if_addgeometrycolumn_does_not_use_quote_ident()

        # Create a table with a name that needs quoting and a geometry
        # column whose name has to be quoted
        stmt = '''CREATE TABLE "table "" name" (gid integer);
             GRANT SELECT ON "table "" name" TO PUBLIC;'''
        self.server.execute_sql(self.dbname, "admin", stmt)
        stmt = """select AddGeometryColumn('%s', 'table \" name',
                                           'the \" geom', '-1', 'POINT', 2);"""
        self.server.execute_sql(self.dbname, "admin", stmt % self.dbname)
        stmt = '''INSERT INTO "table "" name" VALUES (0,
                GeometryFromText('POINT(0 0)', -1));'''
        self.server.execute_sql(self.dbname, "admin", stmt)

        # Instantiate the table
        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        store = PostGISShapeStore(db, "table \" name")

        # do some queries where the names have to be quoted
        self.assertEquals(store.NumShapes(), 1)
        self.assertEquals(store.ShapeType(), SHAPETYPE_POINT)
        self.assertEquals(store.BoundingBox(), (0, 0, 0, 0))
        self.assertEquals(store.Shape(0).ShapeID(), 0)
        self.assertEquals([s.ShapeID() for s in store.AllShapes()], [0])
        self.assertEquals([s.ShapeID()
                           for s in store.ShapesInRegion((-1, -1, 1, 1))], [0])

    def test_shapestore_empty_table(self):
        """Test PostGISShapeStore with emtpy table"""
        conn = psycopg.connect("dbname=%s " % self.dbname
                               + self.server.connection_string("admin"))
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE test (A INT);")
        cursor.execute("SELECT AddGeometryColumn(%(dbname)s, 'test',"
                       " 'geometry', -1, 'POINT', 2);",
                       {"dbname": self.dbname})
        cursor.execute("GRANT SELECT ON test TO PUBLIC;")
        conn.commit()

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        store = PostGISShapeStore(db, "test")
        self.assertEquals(store.BoundingBox(), None)

    def test_shapestore_two_geom_cols(self):
        """Test PostGISShapeStore with two geometry columns"""
        conn = psycopg.connect("dbname=%s " % self.dbname
                               + self.server.connection_string("admin"))
        cursor = conn.cursor()
        cursor.execute("INSERT INTO spatial_ref_sys VALUES"
                       " (1, '', 1, '', 'proj=longlat datum=WGS84')")
        cursor.execute("INSERT INTO spatial_ref_sys VALUES"
                       " (2, '', 2, '', 'proj=longlat datum=WGS84')")

        cursor.execute("CREATE TABLE two_geom_cols"
                       " (gid INTEGER PRIMARY KEY);")
        cursor.execute("SELECT AddGeometryColumn(%(dbname)s, 'two_geom_cols',"
                       " 'point', 1, 'POINT', 2);",
                       {"dbname": self.dbname})
        cursor.execute("SELECT AddGeometryColumn(%(dbname)s, 'two_geom_cols',"
                       " 'poly', 2, 'POLYGON', 2);",
                       {"dbname": self.dbname})
        cursor.execute("INSERT INTO two_geom_cols(gid, point, poly) VALUES (1,"
                       " GeometryFromText('POINT(1 1)', 1),"
                       " GeometryFromText('POLYGON((1 1, 1 2, 2 2,1 1))',2));")
        cursor.execute("GRANT SELECT ON two_geom_cols TO PUBLIC;")
        conn.commit()

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))

        # The table has two geometry columns, and will raise a TypeError
        # when instantiated without specifying which one to use.
        self.assertRaises(TypeError,
                          PostGISShapeStore, db, "two_geom_cols")

        # If the geometry_column is not really a geometry column, we get
        # a TypeError
        self.assertRaises(TypeError,
                          PostGISShapeStore, db, "two_geom_cols",
                          geometry_column = "gid")

        # Instantiate two shape stores, one for each of the geometry
        # columns, and test whether they determine the right shape types
        # and srids (getting the srid wrong would lead to an exception
        # in ShapesInRegion).
        store = PostGISShapeStore(db, "two_geom_cols",
                                  geometry_column = "point")
        self.assertEquals(store.ShapeType(), SHAPETYPE_POINT)
        self.assertEquals([s.ShapeID()
                           for s in store.ShapesInRegion((0, 0, 100,100))],
                          [1])

        store = PostGISShapeStore(db, "two_geom_cols",
                                  geometry_column = "poly")
        self.assertEquals(store.ShapeType(), SHAPETYPE_POLYGON)
        self.assertEquals([s.ShapeID()
                           for s in store.ShapesInRegion((0, 0, 100,100))],
                          [1])


    def test_simple_error_handling(self):
        """Test some simple error handling.

        If an error happens in, say the SimpleQuery method, a subsequent
        ReadValue call should still succeed.
        """
        conn = psycopg.connect("dbname=%s " % self.dbname
                               + self.server.connection_string("admin"))
        cursor = conn.cursor()
        cursor.execute("CREATE TABLE some_table"
                       " (gid INTEGER PRIMARY KEY, length float);"
                       "INSERT INTO some_table VALUES (1, 3.5);")
        cursor.execute("GRANT SELECT ON some_table TO PUBLIC;")
        conn.commit()
        conn.close()

        db = PostGISConnection(dbname = self.dbname,
                               **self.server.connection_params("user"))
        table = PostGISTable(db, "some_table", id_column = "gid")

        # trying to compare the length to a string should fail with an
        # exception.
        self.assertRaises(psycopg.ProgrammingError,
                          table.SimpleQuery,
                          table.Column("length"), "<=", "abc")

        # After the exception, doing a query on the same table should
        # work. Since it's the same database connection as before, it
        # can fail if no rollback has been done or the connection isn't
        # in autocommit mode.  Typical error message is
        #
        #     ERROR: current transaction is aborted, commands ignored
        #     until end of transaction block
        try:
            self.assertEquals(table.ReadValue(1, "length"), 3.5)
        except psycopg.ProgrammingError, val:
            self.fail("table didn't handle exception properly (%s)" % val)


class PostGISStaticTests(unittest.TestCase, support.FloatComparisonMixin):

    """Base class for PostGIS tests with static data."""

    def setUp(self):
        """Start the server and create a database with static data

        This method sets the following instance attributes:

           dbname -- the name of the database

           server -- The server object

           db -- the PostGISConnection object
        """
        postgissupport.skip_if_no_postgis()
        self.server = postgissupport.get_test_server()
        self.postgisdb = self.server.get_default_static_data_db()
        self.db = PostGISConnection(dbname = self.postgisdb.dbname,
                                    **self.server.connection_params("user"))

    def tearDown(self):
        """Close the database connection"""
        self.db.Close()


class TableTests:

    """Mix-in class for the PostGISTable tests

    The tests defined here make some assumptions about the table and the
    data in it:

        self.table should be the PostGISTable instance to test.  Derived
        classes should set this up in setUp.

        self.db should be the DB connection object used by self.table.

        self.table_name should be the name of the table in the database.
        The tests assume that e.g. the title of self.table can be
        derived from this.

        self.gid_column should be the name of the gid column used when
        self.table was instantiated.

        The data in the table should be the data of the
        cultural_landmark-point shapefile with the shape ids used to get
        the contents of the gid column by adding 1000.
    """

    def test_dbconn(self):
        """Test PostGISTable.DBConnection()"""
        self.failUnless(self.table.DBConnection() is self.db)

    def test_dbname(self):
        """Test PostGISTable.TableName()"""
        self.assertEquals(self.table.TableName(), self.table_name)

    def test_title(self):
        """test PostGISTable.Title()"""
        # The title is currently equal to the tablename
        self.assertEquals(self.table.Title(), self.table_name)

    def test_dependencies(self):
        """Test PostGISTable.Dependencies()"""
        # A PostGISTable depends on no other data container
        self.assertEquals(self.table.Dependencies(), ())

    def test_num_rows(self):
        """Test PostGISTable.NumRows()"""
        self.assertEquals(self.table.NumRows(), 34)

    def test_num_columns(self):
        """Test PostGISTable.NumColumns()"""
        # The table in the postgis db has one additional column, "gid",
        # so there's one more column in the PostGISTable than in the DBF
        self.assertEquals(self.table.NumColumns(), 7)

    def test_columns(self):
        """Test PostGISTable.Columns()"""
        self.assertEquals(len(self.table.Columns()), 7)
        self.assertEquals(self.table.Columns()[0].name, self.gid_column)
        self.assertEquals(self.table.Columns()[0].type, FIELDTYPE_INT)
        self.assertEquals(self.table.Columns()[0].index, 0)
        self.assertEquals(self.table.Columns()[1].name, "area")
        self.assertEquals(self.table.Columns()[1].type, FIELDTYPE_DOUBLE)
        self.assertEquals(self.table.Columns()[1].index, 1)
        self.assertEquals(self.table.Columns()[5].name, "clptlabel")
        self.assertEquals(self.table.Columns()[5].type, FIELDTYPE_STRING)
        self.assertEquals(self.table.Columns()[5].index, 5)

    def test_column(self):
        """Test PostGISTable.Column()"""
        self.assertEquals(self.table.Column("area").name, "area")
        self.assertEquals(self.table.Column("area").type, FIELDTYPE_DOUBLE)
        self.assertEquals(self.table.Column("area").index, 1)

    def test_has_column(self):
        """Test PostGISTable.HasColumn()"""
        self.assert_(self.table.HasColumn("area"))
        self.failIf(self.table.HasColumn("foo"))

    def test_read_row_as_dict(self):
        """Test PostGISTable.ReadRowAsDict()"""
        self.assertEquals(self.table.ReadRowAsDict(1003),
                          {self.gid_column: 1003,
                           "area": 0.0,
                           "perimeter": 0.0,
                           "clpoint_": 4,
                           "clpoint_id": 24,
                           "clptlabel": "RUINS",
                           "clptflag": 0})

    def test_read_row_as_dict_row_count_mode(self):
        """Test PostGISTable.ReadRowAsDict() row count address mode"""
        self.assertEquals(self.table.ReadRowAsDict(3, row_is_ordinal = 1),
                          {self.gid_column: 1003,
                           "area": 0.0,
                           "perimeter": 0.0,
                           "clpoint_": 4,
                           "clpoint_id": 24,
                           "clptlabel": "RUINS",
                           "clptflag": 0})

    def test_read_value(self):
        """Test PostGISTable.ReadValue()"""
        self.assertEquals(self.table.ReadValue(1003, 4), 24)
        self.assertEquals(self.table.ReadValue(1003, "clpoint_id"), 24)

    def test_read_value_row_count_mode(self):
        """Test PostGISTable.ReadValue() row count address mode"""
        self.assertEquals(self.table.ReadValue(3, 4, row_is_ordinal = 1), 24)
        self.assertEquals(self.table.ReadValue(3, "clpoint_id",
                                               row_is_ordinal = 1),
                          24)

    def test_row_id_to_ordinal(self):
        """Test PostGISTable.RowIdToOrdinal()"""
        self.assertEquals(self.table.RowIdToOrdinal(1005), 5)

    def test_row_oridnal_to_id(self):
        """Test PostGISTable.RowOrdinalToId()"""
        self.assertEquals(self.table.RowOrdinalToId(5), 1005)

    def test_value_range(self):
        """Test PostGISTable.ValueRange()"""
        self.assertEquals(self.table.ValueRange("clpoint_id"), (21, 74))

    def test_unique_values(self):
        """Test PostGISTable.UniqueValues()"""
        values = self.table.UniqueValues("clptlabel")
        values.sort()
        self.assertEquals(values, ["BUILDING", "FARM", "HUT","LIGHTHOUSE",
                                   "OTHER/UNKNOWN", "RUINS"])

    def test_simple_query(self):
        """Test PostGISTable.SimpleQuery()"""
        table = self.table
        self.assertEquals(table.SimpleQuery(table.Column("clptlabel"),
                                            "==", "FARM"),
                          [1006])
        self.assertEquals(table.SimpleQuery(table.Column("clpoint_id"),
                                            ">", 70),
                          [1024, 1025, 1026])
        self.assertEquals(table.SimpleQuery(table.Column("clpoint_id"),
                                            "<", table.Column("clpoint_")),
                          [1028, 1029, 1030, 1031, 1032, 1033])


class TestPostGISTable(TableTests, PostGISStaticTests):

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISTable"""
        PostGISStaticTests.setUp(self)
        self.table_name = "landmarks"
        self.gid_column = "gid"

        # Note that omit the gid column here to the backwards compatible
        # interface of the constructor where the gid_column defaults to
        # "gid"
        self.table = PostGISTable(self.db, self.table_name)


class TestPostGISTableExplicitGIDColumn(TableTests, PostGISStaticTests):

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISTable"""
        PostGISStaticTests.setUp(self)
        self.table_name = "landmarks_point_id"
        self.gid_column = "point_id"
        self.table = PostGISTable(self.db, self.table_name,
                                  id_column = self.gid_column)


class PointTests:

    """Mix-in class for the tests on POINT tables

    The methods in this class expect self.store to be the shapestore
    used for testing.  Derived classes should initialize self.store with
    PostGISShapeStore in setUp.  Another assumption is that the data in
    self.store is effectively the cultural_landmark-point shapefile with
    shape ids increased by 1000.
    """

    #
    # First, some tests that should be independend of the shapetype, so
    # it shouldn't be necessary to repeat them for other shapetypes
    #

    def test_dependencies(self):
        """Test PostGISShapeStore.Dependencies()"""
        # A PostGISShapeStore depends on no other data container
        self.assertEquals(self.store.Dependencies(), ())

    def test_table(self):
        """Test PostGISShapeStore.Table() with POINT shapes"""
        # A PostGISShapeStore is its own table
        self.assert_(self.store.Table() is self.store)

    def test_orig_shapestore(self):
        """Test PostGISShapeStore.OrigShapeStore() with POINT shapes"""
        # A PostGISShapeStore is not derived from another shape store
        self.assert_(self.store.OrigShapeStore() is None)

    def test_raw_format(self):
        """Test PostGISShapeStore.RawShapeFormat() with POINT shapes"""
        self.assertEquals(self.store.RawShapeFormat(), RAW_WKT)

    def test_all_shapes(self):
        """Test PostGISShapeStore.AllShapes()"""
        self.assertEquals([s.ShapeID() for s in self.store.AllShapes()],
                          range(1000, 1000 + self.store.NumShapes()))

    def test_id_column(self):
        self.assertEquals(self.store.IDColumn().name, self.id_column)

    def test_geometry_column(self):
        self.assertEquals(self.store.GeometryColumn().name,
                          self.geometry_column)

    #
    # Shapetype specific tests
    #

    def test_shape_type(self):
        """Test PostGISShapeStore.ShapeType() with POINT shapes"""
        self.assertEquals(self.store.ShapeType(), SHAPETYPE_POINT)

    def test_num_shapes(self):
        """Test PostGISShapeStore.NumShapes() with POINT shapes"""
        self.assertEquals(self.store.NumShapes(), 34)

    def test_bounding_box(self):
        """Test PostGISShapeStore.BoundingBox() with POINT shapes"""
        self.assertFloatSeqEqual(self.store.BoundingBox(),
                                 [-23.806047439575195, 63.405960083007812,
                                  -15.12291431427002, 66.36572265625])

    def test_shape_shapeid(self):
        """Test PostGISShapeStore.Shape(i).ShapeID() with POINT shapes"""
        self.assertEquals(self.store.Shape(1005).ShapeID(), 1005)

    def test_shape_points(self):
        """Test PostGISShapeStore.Shape(i).Points() with POINT shapes"""
        self.assertPointListEquals(self.store.Shape(1000).Points(),
                                   [[(-22.711074829101562, 66.36572265625)]])

    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with POINT shapes"""
        self.assertEquals(self.store.Shape(1000).RawData(),
                          'POINT(-22.7110748291016 66.36572265625)')

    def test_shapes_in_region(self):
        """Test PostGISShapeStore:ShapesInRegion() with POINT shapes"""
        shapes = self.store.ShapesInRegion((-20.0, 64.0, -24.0, 67))
        self.assertEquals([s.ShapeID() for s in shapes],
                          [1000, 1001, 1002, 1003, 1004, 1005, 1027])


class TestPostGISShapestorePoint(PointTests, PostGISStaticTests):

    """Tests for PostGISShapeStore objects with POINT data an no SRID"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.id_column = "gid"
        self.geometry_column = "the_geom"
        self.store = PostGISShapeStore(self.db, "landmarks")


class TestPostGISShapestorePointSRID(PointTests, PostGISStaticTests):

    """Tests for PostGISShapeStore objects with POINT data and an SRID"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.id_column = "gid"
        self.geometry_column = "the_geom"
        self.store = PostGISShapeStore(self.db, "landmarks_srid")


class TestPostGISShapestorePointExplicitGIDColumn(PointTests,
                                                  PostGISStaticTests):

    """Tests for PostGISShapeStores with POINT data and explicit gid column"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.id_column = "point_id"
        self.geometry_column = "the_geom"
        self.store = PostGISShapeStore(self.db, "landmarks_point_id",
                                       id_column = "point_id")


class TestPostGISShapestorePointOIDAsGIDColumn(PointTests, PostGISStaticTests):

    """Tests for PostGISShapeStores with POINT data using OID as gid column"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.id_column = "oid"
        self.geometry_column = "the_geom"
        self.store = PostGISShapeStore(self.db, "landmarks_point_id",
                                       id_column = self.id_column,
                                       geometry_column = self.geometry_column)

    def find_test_shape_id(self):
        """Return the id of an interesting shape for some test cases.

        This test uses OIDs so we can't easily hard wire ids in the test
        cases
        """
        # get the id of an interesting shape
        shapes = self.store.ShapesInRegion((-20.0, 64.0, -24.0, 67))
        return list(shapes)[0].ShapeID()


    # Override a few tests that won't work with this table because they
    # require knowledge of specific IDs

    def test_shape_shapeid(self):
        """Test PostGISShapeStore.Shape(i).ShapeID() with POINT shapes"""
        gid = self.find_test_shape_id()
        self.assertEquals(self.store.Shape(gid).ShapeID(), gid)

    def test_shape_points(self):
        """Test PostGISShapeStore.Shape(i).Points() with POINT shapes"""
        gid = self.find_test_shape_id()
        self.assertPointListEquals(self.store.Shape(gid).Points(),
                                   [[(-22.711074829101562, 66.36572265625)]])

    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with POINT shapes"""
        gid = self.find_test_shape_id()
        self.assertEquals(self.store.Shape(gid).RawData(),
                          'POINT(-22.7110748291016 66.36572265625)')

    def test_shapes_in_region(self):
        """Test PostGISShapeStore:ShapesInRegion() with POINT shapes"""
        shapes = self.store.ShapesInRegion((-20.0, 64.0, -24.0, 67))
        self.assertEquals(len(list(shapes)), 7)

    def test_all_shapes(self):
        """Test PostGISShapeStore.AllShapes()"""
        self.assertEquals(len(list(self.store.AllShapes())), 34)


class TestPostGISShapestorePointFromViews(PointTests, PostGISStaticTests):

    """Tests for PostGISShapeStores with POINT data in a view"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.id_column = "point_id"
        self.geometry_column = "the_geom"
        self.store = PostGISShapeStore(self.db, "v_landmarks",
                                       id_column = "point_id")


class LineStringTests:

    """Tests shared by the LINESTRING and MULTILINESTRING tests.

    The tests are the same because they are based on the same data.
    """

    def test_shape_type(self):
        """Test PostGISShapeStore.ShapeType() with ARC shapes"""
        self.assertEquals(self.store.ShapeType(), SHAPETYPE_ARC)

    def test_num_shapes(self):
        """Test PostGISShapeStore.NumShapes() with ARC shapes"""
        self.assertEquals(self.store.NumShapes(), 839)

    def test_bounding_box(self):
        """Test PostGISShapeStore.BoundingBox() with ARC shapes"""
        self.assertFloatSeqEqual(self.store.BoundingBox(),
                                 [-24.450359344482422, 63.426830291748047,
                                  -13.55668830871582, 66.520111083984375])

    def test_shape_shapeid(self):
        """Test PostGISShapeStore.Shape(i).ShapeID() with ARC shapes"""
        self.assertEquals(self.store.Shape(5).ShapeID(), 5)

    def test_shape_points(self):
        """Test PostGISShapeStore.Shape(i).Points() with ARC shapes"""
        self.assertPointListEquals(self.store.Shape(32).Points(),
                                   [[(-15.0821743011474, 66.2773818969726),
                                     (-15.0263500213623, 66.2733917236328)]])

    def test_shapes_in_region(self):
        """Test PostGISShapeStore.ShapesInRegion() with ARC shapes"""
        shapes = self.store.ShapesInRegion((-24.0, 64.5, -23.5, 65.0))
        self.assertEquals([s.ShapeID() for s in shapes], [573, 581, 613])


class TestPostGISShapestoreLineString(LineStringTests, PostGISStaticTests):

    """Tests for PostGISShapeStore objects with LINESTRING data"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.store = PostGISShapeStore(self.db, "roads")

    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with ARC shapes"""
        self.assertEquals(self.store.Shape(32).RawData(),
                     "LINESTRING(-15.0821743011475 66.2773818969727,"
                                "-15.0263500213623 66.2733917236328)")


class TestPostGISShapestoreMultiLineString(LineStringTests,
                                           PostGISStaticTests):

    """Tests for PostGISShapeStore objects with MULTILINESTRING data"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.store = PostGISShapeStore(self.db, "roads_multi")

    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with ARC shapes"""
        self.assertEquals(self.store.Shape(32).RawData(),
                     "MULTILINESTRING((-15.0821743011475 66.2773818969727,"
                                      "-15.0263500213623 66.2733917236328))")


class PolygonTests:

    """Test shared by the POLYGON and MULTIPOLYGON tests

    The tests are the same because they are based on the same data.
    """

    def test_shape_type(self):
        """Test PostGISShapeStore.ShapeType() with POLYGON shapes"""
        self.assertEquals(self.store.ShapeType(), SHAPETYPE_POLYGON)

    def test_num_shapes(self):
        """Test PostGISShapeStore.NumShapes() with POLYGON shapes"""
        self.assertEquals(self.store.NumShapes(), 156)

    def test_bounding_box(self):
        """Test PostGISShapeStore.BoundingBox() with POLYGON shapes"""
        self.assertFloatSeqEqual(self.store.BoundingBox(),
                                 [-24.546524047851562, 63.286754608154297,
                                  -13.495815277099609, 66.563774108886719])

    def test_shape_shapeid(self):
        """Test PostGISShapeStore.Shape(i).ShapeID() with POLYGON shapes"""
        self.assertEquals(self.store.Shape(5).ShapeID(), 5)

    def test_shape_points(self):
        """Test PostGISShapeStore.Shape(i).Points() with POLYGON shapes"""
        self.assertPointListEquals(self.store.Shape(4).Points(),
                                   [[(-22.40639114379882, 64.714111328125),
                                     (-22.41621208190918, 64.716003417968),
                                     (-22.40605163574218, 64.719200134277),
                                     (-22.40639114379882, 64.714111328125)]])

    def test_shapes_in_region(self):
        """Test PostGISShapeStore.ShapesInRegion() with POLYGON shapes"""
        shapes = self.store.ShapesInRegion((-23.0, 65.5, -22.8, 65.25))
        self.assertEquals([s.ShapeID() for s in shapes],
                          [47, 56, 59, 61, 62, 71, 144])


class TestPostGISShapestorePolygon(PolygonTests, PostGISStaticTests):

    """Tests for PostGISShapeStore objects with POLYGON data"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.store = PostGISShapeStore(self.db, "political")

    def test_shape_type(self):
        """Test PostGISShapeStore.ShapeType() with POLYGON shapes"""
        self.assertEquals(self.store.ShapeType(), SHAPETYPE_POLYGON)


    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with POLYGON shapes"""
        self.assertEquals(self.store.Shape(4).RawData(),
                          "POLYGON((-22.4063911437988 64.714111328125,"
                                   "-22.4162120819092 64.7160034179688,"
                                   "-22.4060516357422 64.7192001342773,"
                                   "-22.4063911437988 64.714111328125))")


class TestPostGISShapestoreMultiPolygon(PolygonTests, PostGISStaticTests):

    """Tests for PostGISShapeStore objects with MUTLIPOLYGON data"""

    def setUp(self):
        """Extend inherited method to set self.table to a PostGISShapeStore"""
        PostGISStaticTests.setUp(self)
        self.store = PostGISShapeStore(self.db, "political_multi")

    def test_shape_raw_data(self):
        """Test PostGISShapeStore.Shape(i).RawData() with POLYGON shapes"""
        self.assertEquals(self.store.Shape(4).RawData(),
                          "MULTIPOLYGON(((-22.4063911437988 64.714111328125,"
                                         "-22.4162120819092 64.7160034179688,"
                                         "-22.4060516357422 64.7192001342773,"
                                        "-22.4063911437988 64.714111328125)))")


if __name__ == "__main__":
    support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision