WIP-pyshapelib-bramz/test/test_save.py

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

"""
Test saving a thuban session as XML
"""

__version__ = "$Revision$"
# $Source$
# $Id$

import os
import unittest
from StringIO import StringIO

import xmlsupport
import postgissupport

import support
support.initthuban()

import dbflib

from Thuban import internal_from_unicode
from Thuban.Lib.fileutil import relative_filename
from Thuban.Model.save import XMLWriter, save_session, sort_data_stores
from Thuban.Model.session import Session
from Thuban.Model.map import Map
from Thuban.Model.layer import Layer, RasterLayer
from Thuban.Model.proj import Projection
from Thuban.Model.table import DBFTable
from Thuban.Model.transientdb import TransientJoinedTable
from Thuban.Model.data import DerivedShapeStore, SHAPETYPE_ARC

from Thuban.Model.classification import ClassGroupSingleton, ClassGroupRange, \
    ClassGroupProperties

from Thuban.Model.range import Range

from Thuban.Model.postgisdb import PostGISConnection, PostGISShapeStore


class XMLWriterTest(unittest.TestCase):

    def testEncode(self):
        """Test XMLWriter.encode"""
        writer = XMLWriter()
        eq = self.assertEquals

        eq(writer.encode("hello world"), "hello world")
        eq(writer.encode(unicode("hello world")), unicode("hello world"))

        eq(writer.encode(internal_from_unicode(u"\x80\x90\xc2\x100")),
                         "\xc2\x80\xc2\x90\xc3\x82\x100")
        eq(writer.encode(u"\x80\x90\xc2\x100"),
                         "\xc2\x80\xc2\x90\xc3\x82\x100")
        eq(writer.encode(u"\xFF5E"), "\xc3\xbf5E")

        eq(writer.encode('&"\'<>'), "&amp;&quot;&apos;&lt;&gt;")
        eq(writer.encode(unicode('&"\'<>')), "&amp;&quot;&apos;&lt;&gt;")

class SaveSessionTest(unittest.TestCase, support.FileTestMixin,
                      xmlsupport.ValidationTest):

    dtd = "http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd"
    thubanids = [((dtd, n), (None, "id")) for n in
                 ["fileshapesource", "filetable", "jointable",
                  "derivedshapesource", "dbshapesource", "dbconnection"]]
    thubanidrefs = [((dtd, n), (None, m)) for n, m in
                    [("layer", "shapestore"),
                     ("jointable", "left"),
                     ("jointable", "right"),
                     ("derivedshapesource", "table"),
                     ("derivedshapesource", "shapesource"),
                     ("dbshapesource", "dbconn")]]
    del n, m, dtd

    def tearDown(self):
        """Call self.session.Destroy

        Test cases that create session should bind it to self.session so
        that it gets destroyed properly
        """
        if hasattr(self, "session"):
            self.session.Destroy()
            self.session = None

    def compare_xml(self, xml1, xml2):
        list1 = xmlsupport.sax_eventlist(xml1, ids = self.thubanids,
                                         idrefs = self.thubanidrefs)
        list2 = xmlsupport.sax_eventlist(xml2, ids = self.thubanids,
                                         idrefs = self.thubanidrefs)
        if list1 != list2:
            for a, b in zip(list1, list2):
                if a != b:
                    self.fail("%r != %r" % (a, b))


    def testEmptySession(self):
        """Save an empty session"""
        session = Session("empty session")
        filename = self.temp_file_name("save_emptysession.thuban")
        save_session(session, filename)
        session.Destroy()

        file = open(filename)
        written_contents = file.read()
        file.close()
        self.compare_xml(written_contents,
                         '<?xml version="1.0" encoding="UTF-8"?>\n'
                         '<!DOCTYPE session SYSTEM "thuban-1.1.dtd">\n'
                         '<session title="empty session" '
         'xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">'
                         '\n</session>\n')

        self.validate_data(written_contents)

    def testSingleLayer(self):
        """Save a session with a single map with a single layer"""
        # deliberately put an apersand in the title :)
        session = Session("single map&layer")
        proj = Projection(["proj=utm", "zone=27", "ellps=WGS84",
                           "datum=WGS84", "units=m"],
                          name = "WGS 84 / UTM zone 27N",
                          epsg = "32627")
        map = Map("Test Map", projection = proj)
        session.AddMap(map)
        # use shapefile from the example data
        shpfile = os.path.join(os.path.dirname(__file__),
                               os.pardir, "Data", "iceland", "political.shp")
        layer = Layer("My Layer", session.OpenShapefile(shpfile))
        map.AddLayer(layer)

        filename = self.temp_file_name("save_singlemap.thuban")
        save_session(session, filename)

        file = open(filename)
        written_contents = file.read()
        file.close()
        expected_template = '''<?xml version="1.0" encoding="UTF-8"?>
        <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
        <session title="single map&amp;layer"
           xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
            <fileshapesource id="D1"
                filename="../../Data/iceland/political.shp"
                filetype="shapefile"/>
            <map title="Test Map">
                <projection epsg="32627" name="WGS 84 / UTM zone 27N">
                    <parameter value="proj=utm"/>
                    <parameter value="zone=27"/>
                    <parameter value="ellps=WGS84"/>
                    <parameter value="datum=WGS84"/>
                    <parameter value="units=m"/>
                </projection>
                <layer title="My Layer" shapestore="D1" visible="%s">
                    <classification>
                        <clnull label="">
                            <cldata fill="None" stroke="#000000"
                                stroke_width="1"/>
                        </clnull>
                    </classification>
                </layer>
            </map>
        </session>'''

        expected_contents = expected_template % "true"

        self.compare_xml(written_contents, expected_contents)

        self.validate_data(written_contents)

        # Repeat with an invisible layer
        layer.SetVisible(False)
        save_session(session, filename)

        file = open(filename)
        written_contents = file.read()
        file.close()
        expected_contents = expected_template % "false"
        self.compare_xml(written_contents, expected_contents)
        self.validate_data(written_contents)

        session.Destroy()

    def testLayerProjection(self):
        """Test saving layers with projections"""
        # deliberately put an apersand in the title :)
        session = self.session = Session("single map&layer")
        proj = Projection(["zone=26", "proj=utm", "ellps=clrk66"])
        map = Map("Test Map", projection = proj)
        session.AddMap(map)
        # use shapefile from the example data
        shpfile = os.path.join(os.path.dirname(__file__),
                               os.pardir, "Data", "iceland", "political.shp")
        layer = Layer("My Layer", session.OpenShapefile(shpfile))
        proj = Projection(["proj=lcc", "ellps=clrk66",
                           "lat_1=0", "lat_2=20"],
                          "Layer Projection")
        layer.SetProjection(proj)
        map.AddLayer(layer)

        filename = self.temp_file_name("save_layerproj.thuban")
        save_session(session, filename)

        file = open(filename)
        written_contents = file.read()
        file.close()
        expected_contents = '''<?xml version="1.0" encoding="UTF-8"?>
        <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
        <session title="single map&amp;layer"
           xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
            <fileshapesource id="D1"
                filename="../../Data/iceland/political.shp"
                filetype="shapefile"/>
            <map title="Test Map">
                <projection name="Unknown">
                    <parameter value="zone=26"/>
                    <parameter value="proj=utm"/>
                    <parameter value="ellps=clrk66"/>
                </projection>
                <layer title="My Layer" shapestore="D1" visible="true">
                    <projection name="Layer Projection">
                        <parameter value="proj=lcc"/>
                        <parameter value="ellps=clrk66"/>
                        <parameter value="lat_1=0"/>
                        <parameter value="lat_2=20"/>
                    </projection>
                    <classification>
                        <clnull label="">
                            <cldata fill="None" stroke="#000000"
                                stroke_width="1"/>
                        </clnull>
                    </classification>
                </layer>
            </map>
        </session>'''
        #print written_contents
        #print "********************************************"
        #print expected_contents
        self.compare_xml(written_contents, expected_contents)

        self.validate_data(written_contents)

    def testRasterLayer(self):

        MASK_NONE = RasterLayer.MASK_NONE
        MASK_BIT = RasterLayer.MASK_BIT
        MASK_ALPHA = RasterLayer.MASK_ALPHA

        for opacity, masktype, opname, maskname in \
            [(1,  MASK_BIT,   '', ''),
             (.2, MASK_BIT,   'opacity="0.2"', ''),
             (1,  MASK_ALPHA, '',              'masktype="alpha"'),
             (.5, MASK_ALPHA, 'opacity="0.5"', 'masktype="alpha"'),
             (1,  MASK_NONE,  '',              'masktype="none"'),
             (0,  MASK_NONE,  'opacity="0"',   'masktype="none"') ]:


            # deliberately put an apersand in the title :)
            session = Session("single map&layer")
            map = Map("Test Map")
            session.AddMap(map)
            # use shapefile from the example data
            imgfile = os.path.join(os.path.dirname(__file__),
                                   os.pardir, "Data", "iceland", "island.tif")
            layer = RasterLayer("My RasterLayer", imgfile)

            layer.SetOpacity(opacity)
            layer.SetMaskType(masktype)

            map.AddLayer(layer)

            filename = self.temp_file_name("%s.thuban" % self.id())
            save_session(session, filename)
            session.Destroy()

            file = open(filename)
            written_contents = file.read()
            file.close()
            expected_contents = '''<?xml version="1.0" encoding="UTF-8"?>
            <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
            <session title="single map&amp;layer"
               xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
                <map title="Test Map">
                    <rasterlayer title="My RasterLayer"
                            filename="../../Data/iceland/island.tif"
                            visible="true" %s %s>
                    </rasterlayer>
                </map>
            </session>''' % (opname, maskname)
            #print written_contents
            #print "********************************************"
            #print expected_contents
            self.compare_xml(written_contents, expected_contents)

            self.validate_data(written_contents)

    def testClassifiedLayer(self):
        """Save a session with a single map with classifications"""
        # deliberately put an apersand in the title :)
        session = Session("Map with Classifications")
        proj = Projection(["zone=26", "proj=utm", "ellps=clrk66"])
        map = Map("Test Map", projection = proj)
        session.AddMap(map)
        # use shapefile from the example data
        shpfile = os.path.join(os.path.dirname(__file__),
                               os.pardir, "Data", "iceland", "political.shp")
        layer = Layer("My Layer", session.OpenShapefile(shpfile))
        map.AddLayer(layer)
        layer2 = Layer("My Layer", layer.ShapeStore())
        map.AddLayer(layer2)

        clazz = layer.GetClassification()

        layer.SetClassificationColumn("AREA")

        clazz.AppendGroup(ClassGroupSingleton(42, ClassGroupProperties(),
                                              "single"))
        clazz.AppendGroup(ClassGroupSingleton("text", ClassGroupProperties(),
                                              "single-text"))

        clazz.AppendGroup(ClassGroupRange((0, 42),
                                           ClassGroupProperties(),
                                           "range"))

        range = ClassGroupRange(Range("[0;42]"))
        range.SetProperties(ClassGroupProperties())
        range.SetLabel("new-range")
        clazz.AppendGroup(range)


        clazz = layer2.GetClassification()
        layer2.SetClassificationColumn("POPYCOUN")

        # Classification with Latin 1 text
        clazz.AppendGroup(ClassGroupSingleton(
            internal_from_unicode(u'\xe4\xf6\xfc'), # ae, oe, ue
            ClassGroupProperties(),
            internal_from_unicode(u'\xdcml\xe4uts'))) # Uemlaeuts


        filename = self.temp_file_name("%s.thuban" % self.id())
        save_session(session, filename)

        file = open(filename)
        written_contents = file.read()
        file.close()
        expected_contents = '''<?xml version="1.0" encoding="UTF-8"?>
        <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
        <session title="Map with Classifications"
           xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
            <fileshapesource id="D1"
                filename="../../Data/iceland/political.shp"
                filetype="shapefile"/>
            <map title="Test Map">
                <projection name="Unknown">
                    <parameter value="zone=26"/>
                    <parameter value="proj=utm"/>
                    <parameter value="ellps=clrk66"/>
                </projection>
                <layer title="My Layer" shapestore="D1" visible="true">
                    <classification field="AREA" field_type="double">
                        <clnull label="">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clnull>
                        <clpoint value="42" label="single">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clpoint>
                        <clpoint value="text" label="single-text">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clpoint>
                        <clrange range="[0;42[" label="range">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clrange>
                        <clrange range="[0;42]" label="new-range">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clrange>
                    </classification>
                </layer>
                <layer title="My Layer" shapestore="D1" visible="true">
                    <classification field="POPYCOUN" field_type="string">
                        <clnull label="">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clnull>
                        <clpoint value="\xc3\xa4\xc3\xb6\xc3\xbc"
                             label="\xc3\x9cml\xc3\xa4uts">
                            <cldata fill="None" stroke="#000000" stroke_width="1"/>
                        </clpoint>
                    </classification>
                </layer>
            </map>
        </session>'''

        #print written_contents
        #print "********************************************"
        #print expected_contents
        self.compare_xml(written_contents, expected_contents)

        self.validate_data(written_contents)

        session.Destroy()

    def test_dbf_table(self):
        """Test saving a session with a dbf table link"""
        session = self.session = Session("a DBF Table session")
        # use shapefile from the example data
        dbffile = os.path.join(os.path.dirname(__file__),
                               os.pardir, "Data", "iceland", "political.dbf")
        table = session.AddTable(DBFTable(dbffile))

        filename = self.temp_file_name("save_singletable.thuban")
        save_session(session, filename)

        file = open(filename)
        written_contents = file.read()
        file.close()
        expected_contents = '''<?xml version="1.0" encoding="UTF-8"?>
        <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
        <session title="a DBF Table session"
           xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
            <filetable id="D1" filename="../../Data/iceland/political.dbf"
                filetype="DBF" title="political"/>
        </session>'''

        self.compare_xml(written_contents, expected_contents)
        self.validate_data(written_contents)

    def test_joined_table(self):
        """Test saving a session with joined table"""
        # Create a simple table to use in the join
        dbffile = self.temp_file_name("save_joinedtable.dbf")
        dbf = dbflib.create(dbffile)
        dbf.add_field("RDTYPE", dbflib.FTInteger, 10, 0)
        dbf.add_field("TEXT", dbflib.FTString, 10, 0)
        dbf.write_record(0, {'RDTYPE': 8, "TEXT": "foo"})
        dbf.write_record(1, {'RDTYPE': 2, "TEXT": "bar"})
        dbf.write_record(2, {'RDTYPE': 3, "TEXT": "baz"})
        dbf.close()

        # Create the session and a map
        session = Session("A Joined Table session")
        try:
            map = Map("Test Map")
            session.AddMap(map)

            # Add the dbf file to the session
            dbftable = session.AddTable(DBFTable(dbffile))

            # Create a layer with the shapefile to use in the join
            shpfile = os.path.join(os.path.abspath(os.path.dirname(__file__)),
                                   os.pardir, "Data", "iceland",
                                   "roads-line.shp")
            layer = Layer("My Layer", session.OpenShapefile(shpfile))
            map.AddLayer(layer)

            # Do the join
            store = layer.ShapeStore()
            #for col in store.Table().Columns():
            #    print col.name
            joined = TransientJoinedTable(session.TransientDB(),
                                          store.Table(), "RDLNTYPE",
                                          dbftable, "RDTYPE",
                                          outer_join = True)
            store = session.AddShapeStore(DerivedShapeStore(store, joined))
            layer.SetShapeStore(store)

            # Save the session
            filename = self.temp_file_name("save_joinedtable.thuban")
            save_session(session, filename)

            # Read it back and compare
            file = open(filename)
            written_contents = file.read()
            file.close()
            expected_contents = '''<?xml version="1.0" encoding="UTF-8"?>
            <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
            <session title="A Joined Table session"
             xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
                <fileshapesource filename="../../Data/iceland/roads-line.shp"
                                 filetype="shapefile" id="D142197204"/>
                <filetable filename="save_joinedtable.dbf"
                           title="save_joinedtable"
                           filetype="DBF" id="D141881756"/>
                <jointable id="D142180284"
                           title="Join of roads-line and save_joinedtable"
                           leftcolumn="RDLNTYPE" left="D142197204"
                           rightcolumn="RDTYPE" right="D141881756"
                           jointype="LEFT OUTER" />
                <derivedshapesource id="D141915644"
                                    table="D142180284"
                                    shapesource="D142197204"/>
                <map title="Test Map">
                    <layer title="My Layer"
                           shapestore="D141915644" visible="true">
                        <classification>
                            <clnull label="">
                                <cldata fill="None" stroke="#000000"
                                        stroke_width="1"/>
                            </clnull>
                        </classification>
                    </layer>
                </map>
            </session>'''

            self.compare_xml(written_contents, expected_contents)
            self.validate_data(written_contents)
        finally:
            session.Destroy()
            session = None


    def test_save_postgis(self):
        """Test saving a session with a postgis connection"""

        class NonConnection(PostGISConnection):
            """connection class that doesn't actually connect """
            def connect(self):
                pass

        class NonConnectionStore(PostGISShapeStore):
            """Shapestore that doesn't try to access the server"""
            def _fetch_table_information(self):
                # pretend that we've found a geometry column
                self.geometry_column = "the_geom"
                # pretend this is a ARC shape type.
                self.shape_type = SHAPETYPE_ARC
            def IDColumn(self):
                """Return an object with a name attribute with value 'gid'"""
                class dummycol:
                    name = "gid"
                return dummycol

        session = Session("A PostGIS Session")
        try:
            dbconn = NonConnection(dbname="plugh", host="xyzzy", port="42",
                                   user="grue")
            session.AddDBConnection(dbconn)
            map = Map("Test Map")
            session.AddMap(map)
            store = NonConnectionStore(dbconn, "roads")
            session.AddShapeStore(store)
            layer = Layer("Roads to Nowhere", store)
            map.AddLayer(layer)

            # Save the session
            filename = self.temp_file_name(self.id() + ".thuban")
            save_session(session, filename)

            # Read it back and compare
            file = open(filename)
            written = file.read()
            file.close()
            expected = '''<?xml version="1.0" encoding="UTF-8"?>
            <!DOCTYPE session SYSTEM "thuban-1.1.dtd">
            <session title="A PostGIS Session"
             xmlns="http://thuban.intevation.org/dtds/thuban-1.1-dev.dtd">
                <dbconnection id="DB"
                              dbtype="postgis" dbname="plugh"
                              host="xyzzy" port="42"
                              user="grue"/>
                <dbshapesource id="roads" dbconn="DB" tablename="roads"
                               id_column="gid" geometry_column="the_geom"/>
                <map title="Test Map">
                    <layer title="Roads to Nowhere"
                           shapestore="roads" visible="true">
                        <classification>
                            <clnull label="">
                                <cldata fill="None" stroke="#000000"
                                    stroke_width="1"/>
                            </clnull>
                        </classification>
                    </layer>
                </map>
            </session>'''
            self.compare_xml(written, expected)
            self.validate_data(written)
        finally:
            session.Destroy()


class MockDataStore:

    """A very simple data store that only has dependencies"""

    def __init__(self, name, *dependencies):
        self.name = name
        self.dependencies = dependencies

    def __repr__(self):
        return self.name

    def Dependencies(self):
        return self.dependencies


class TestStoreSort(unittest.TestCase):

    def check_sort(self, containers, sorted):
        """Check whether the list of data containers is sorted"""
        # check whether sorted is in the right order
        seen = {}
        for container in sorted:
            self.failIf(id(container) in seen,
                        "Container %r at least twice in %r" % (container,
                                                               sorted))
            for dep in container.Dependencies():
                self.assert_(id(dep) in seen,
                             "Dependency %r of %r not yet seen" % (dep,
                                                                   container))
            seen[id(container)] = 1
        # check whether all of containers is in sorted
        for container in containers:
            self.assert_(id(container) in seen,
                         "Container %r in containers but not in sorted")
        self.assertEquals(len(containers), len(sorted))

    def test_sort_data_stores(self):
        """Test Thuban.Model.save.sort_data_stores"""
        d1 = MockDataStore("d1")
        d2 = MockDataStore("d2")
        d3 = MockDataStore("d3", d1)
        d4 = MockDataStore("d4", d1, d3)

        containers = [d4, d1, d2, d3]
        self.check_sort(containers, sort_data_stores(containers))
        containers = [d1, d3, d2, d4]
        self.check_sort(containers, sort_data_stores(containers))


if __name__ == "__main__":
    # Fake the __file__ global because it's needed by a test
    import sys
    __file__ = sys.argv[0]
    support.run_tests()
Name	Value
svn:eol-style	native
svn:keywords	Author Date Id Revision