# Copyright (c) 2002, 2003, 2004, 2005, 2007 by Intevation GmbH # Authors: # Bernhard Herzog # # This program is free software under the GPL (>=v2) # Read the file COPYING coming with Thuban for details. """ Test loading a thuban session from a file The tests in this file (test_load.py) are always be the tests for the current version of the thuban file format. Tests for older versions can be found in the version specific test modules, e.g. test_load_0_2 for files created by Thuban 0.2. Maintenance of the test cases: When during a development period the file format is changed with respect to the last released version for the first time, the tests here should be copied to the version specific test file. The round-trip tests which save the session again and compare the XML files should not be copied over as they only make sense here to make sure th that the files checked here are actually ones that may have been written by the current thuban version. """ __version__ = "$Revision: 2826 $" # $Source$ # $Id: test_load.py 2826 2008-01-31 15:41:56Z bernhard $ import os import unittest import support support.initthuban() import postgissupport from xmlsupport import sax_eventlist import dbflib import shapelib from Thuban import internal_from_unicode from Thuban.Model.save import save_session from Thuban.Model.load import load_session, parse_color, LoadError, \ LoadCancelled from Thuban.Model.color import Transparent from Thuban.Model.classification import ClassGroupProperties, ClassGroupRange,\ ClassGroupSingleton, ClassGroupPattern, ClassGroupDefault from Thuban.Model.postgisdb import ConnectionError from Thuban.Model.table import DBFTable, MemoryTable, \ FIELDTYPE_DOUBLE, FIELDTYPE_INT, FIELDTYPE_STRING, \ table_to_dbf from Thuban.Model.label import ALIGN_CENTER, ALIGN_TOP, ALIGN_BOTTOM, \ ALIGN_LEFT, ALIGN_RIGHT, ALIGN_BASELINE def filenames_equal(name1, name2): """Return true if the filenames name1 and name2 are equal. On systems where it is available, simply use os.path.samefile, otherwise return whether the normalized versions of the filenames according to os.path.normpath are equal. """ if hasattr(os.path, "samefile"): return os.path.samefile(name1, name2) return os.path.normpath(name1) == os.path.normpath(name2) class LoadSessionTest(support.FileLoadTestCase): """Base class for .thuban file loading tests Basically the same as the FileLoadTestCase, except that all tests use the '.thuban' extension by default and that setUp and tearDown handle sessions. """ file_extension = ".thuban" def setUp(self): """Create the test files""" support.FileLoadTestCase.setUp(self) self.session = None def tearDown(self): if self.session is not None: self.session.Destroy() self.session = None dtd = "http://thuban.intevation.org/dtds/thuban-1.2.1.dtd" thubanids = [((dtd, n), (None, "id")) for n in ["fileshapesource", "filetable", "jointable", "derivedshapesource"]] thubanidrefs = [((dtd, n), (None, m)) for n, m in [("layer", "shapestore"), ("jointable", "left"), ("jointable", "right"), ("derivedshapesource", "table"), ("derivedshapesource", "shapesource")]] # The filenames in the tests should be understandable on all # currently supported platforms so filenames is an empty list filenames = [] del n, m, dtd def check_format(self): """Check whether the file we loaded from matches the one that would be written. Call this from each test case after loading the session """ filename = self.temp_file_name(self.id() + ".roundtrip.thuban") save_session(self.session, filename) el1 = sax_eventlist(filename = filename, ids = self.thubanids, idrefs = self.thubanidrefs, filenames = self.filenames) el2 = sax_eventlist(filename = self.filename(), ids = self.thubanids, idrefs = self.thubanidrefs, filenames = self.filenames) if 0: for a, b in zip(el1, el2): print a != b and "***************" or "" print a print b self.assertEquals(el1, el2, "loaded file not equivalent to the saved file") class ClassificationTest(LoadSessionTest): """ Base class for tests that do some detailed checking of classifications """ def TestLayers(self, layers, expected): TITLE = 0 NUM_GROUPS = 1 CLASSES = 2 GROUP_TYPE = 0 GROUP_DATA = 1 GROUP_LABEL = 2 GROUP_PROPS = 3 eq = self.assertEquals eq(len(layers), len(expected)) for layer, data in zip(layers, expected): eq(layer.Title(), data[TITLE]) clazz = layer.GetClassification() eq(clazz.GetNumGroups(), data[NUM_GROUPS]) eq(clazz.GetNumGroups() + 1, len(data[CLASSES])) i = 0 for group in clazz: props = ClassGroupProperties() props.SetLineColor( parse_color(data[CLASSES][i][GROUP_PROPS][0])) props.SetLineWidth(data[CLASSES][i][GROUP_PROPS][1]) props.SetFill( parse_color(data[CLASSES][i][GROUP_PROPS][2])) if len(data[CLASSES][i][GROUP_PROPS]) > 3: props.SetSize(data[CLASSES][i][GROUP_PROPS][3]) if data[CLASSES][i][GROUP_TYPE] == "default": g = ClassGroupDefault(props, data[CLASSES][i][GROUP_LABEL]) elif data[CLASSES][i][GROUP_TYPE] == "range": g = ClassGroupRange((data[CLASSES][i][GROUP_DATA][0], data[CLASSES][i][GROUP_DATA][1]), props, data[CLASSES][i][GROUP_LABEL]) elif data[CLASSES][i][GROUP_TYPE] == "single": g = ClassGroupSingleton(data[CLASSES][i][GROUP_DATA], props, data[CLASSES][i][GROUP_LABEL]) elif data[CLASSES][i][GROUP_TYPE] == "pattern": g = ClassGroupPattern(data[CLASSES][i][GROUP_DATA], props, data[CLASSES][i][GROUP_LABEL]) eq(group, g) i += 1 class TestSingleLayer(LoadSessionTest): # Note: The use of & and non-ascii characters is deliberate. We # want to test whether the loading code handles that correctly. file_contents = '''\ ''' def test(self): """Load a session with a single map with a single layer""" eq = self.assertEquals session = load_session(self.filename()) self.session = session # Check the title eq(session.Title(), internal_from_unicode(u"Stra\xdfen & Landmarken")) # the session has one map. maps = session.Maps() eq(len(maps), 1) # Check the map's attributes map = maps[0] eq(map.Title(), internal_from_unicode(u"\xdcbersicht")) proj = map.GetProjection() eq(proj.GetName(), "WGS 84 / UTM zone 27N") eq(proj.EPSGCode(), "32627") params = proj.GetAllParameters() params.sort() eq(params, ["datum=WGS84", "ellps=WGS84", "proj=utm", "units=m", "zone=27"]) # the map has a single layer layers = map.Layers() eq(len(layers), 1) # Check the layer attributes layer = layers[0] eq(layer.Title(), internal_from_unicode(u"K\xfcste")) self.failUnless(filenames_equal(layer.ShapeStore().FileName(), os.path.join(self.temp_dir(), os.pardir, os.pardir, "Data", "iceland", "political.shp"))) eq(layer.GetClassification().GetDefaultFill(), Transparent) eq(layer.GetClassification().GetDefaultLineColor().hex(), "#000000") eq(layer.Visible(), True) self.check_format() self.session.Destroy() self.session = None def test_leak(self): """Test load_session for resource leaks The load_session function had a resource leak in that it created cyclic references. The objects would have been eventually collected by the garbage collector but too late. One symptom is that when layers are removed so that the last normal reference owned indirectly by the session to a shape store goes away, the shape store is not actually removed from the session even though the session only keeps weak references because there are still references owned by the cyclic garbage. """ session = load_session(self.filename()) self.session = session # sanity check self.assertEquals(len(session.ShapeStores()), 1) # remove the map. The shapestore should go away too session.RemoveMap(session.Maps()[0]) self.assertEquals(len(session.ShapeStores()), 0) class TestNonAsciiColumnName(LoadSessionTest): file_contents = '''\ ''' def test(self): """Load a session with a single map with a single layer""" # Create a shapefile and a dbffile with a non-ascii column name dbffile = self.temp_file_name("TestNonAsciiColumnName.dbf") shpfile = self.temp_file_name("TestNonAsciiColumnName.shp") dbf = dbflib.create(dbffile) dbf.add_field('Fl\xe4che', dbflib.FTDouble, 10, 5) dbf.write_record(0, (0.0,)) dbf.close() shp = shapelib.create(shpfile, shapelib.SHPT_POLYGON) shp.write_object(-1, shapelib.SHPObject(shapelib.SHPT_POLYGON, 1, [[(0,0), (10, 10), (10, 0), (0, 0)]])) shp.close() try: session = load_session(self.filename()) except ValueError, v: # Usually if the field name is not decoded properly the # loading fails because the field type mentioned in the file # is not None as returned from the layer for a non-existing # column name so we check for that and report it as failure. # Other exceptions are errors in the test case. if str(v) == "xml field type differs from database!": self.fail("Cannot load file with non-ascii column names") else: raise self.session = session # In case Thuban could load the file anyway (i.e. no ValueError # exception in load_session()), check explicitly whether the # field name was decoded properly. The test will probably lead # to a UnicodeError instead of a test failure so we check that # too layer = session.Maps()[0].Layers()[0] try: self.assertEquals(layer.GetClassificationColumn(), 'Fl\xe4che') except UnicodeError: # FIXME: Obviously this will have to change if Thuban ever # supports unicode properly. self.fail("Column name was not converted to a bytestring") # roundtrip check self.check_format() class TestLayerVisibility(LoadSessionTest): file_contents = '''\ ''' def test(self): """Test that the visible flag is correctly loaded for a layer.""" eq = self.assertEquals session = load_session(self.filename()) self.session = session maps = session.Maps() eq(len(maps), 1) map = maps[0] layers = map.Layers() eq(len(layers), 1) layer = layers[0] eq(layer.Visible(), False) self.check_format() class TestSymbolSize(ClassificationTest): file_contents = '''\ ''' def test(self): """Test that the size attribute for point symbols is correctly loaded for a layer.""" eq = self.assertEquals session = load_session(self.filename()) self.session = session map = session.Maps()[0] # only one map in the sample expected = [("cultural_landmark-point", 2, [("default", (), "", ("#000000", 1, "#000000", 3)), ("single", "RUINS", "", ("#000000", 1, "#ffffff", 6)), ("single", "FARM", "", ("#000000", 1, "#ffff00", 9))])] self.TestLayers(map.Layers(), expected) self.check_format() class TestClassification(ClassificationTest): file_contents = '''\ ''' def test(self): """Load a Thuban session with a map and classified layers.""" session = load_session(self.filename()) self.session = session map = self.session.Maps()[0] # only one map in the sample expected = [("My Layer", 3, [("default", (), "", ("#000000", 1, "None")), ("single", "1", "", ("#000000", 2, "None")), ("single", "1", "", ("#000000", 10, "None")), ("single", internal_from_unicode(u"\xe4\xf6\xfc"), internal_from_unicode(u"\xdcml\xe4uts"), ("#000000", 1, "None"))]), ("My Layer 2", 4, [("default", (), "", ("#000000", 2, "None")), ("range", (0, 1), "", ("#111111", 1, "None")), ("single", .5, "", ("#000000", 1, "#111111")), ("range", (-1, 0), "", ("#000000", 1, "None")), ("single", -.5, "", ("#000000", 1, "None"))]), ("My Layer 3", 2, [("default", (), "", ("#000000", 2, "None")), ("single", "FARM", "", ("#111111", 1, "None")), ("pattern", "BUI", "", ("#000000", 1, "None"))]), ] self.TestLayers(map.Layers(), expected) self.check_format() class TestLabels(ClassificationTest): file_contents = '''\ ''' def test(self): """Load a session and test for reading the group labels.""" eq = self.assertEquals session = load_session(self.filename()) self.session = session map = self.session.Maps()[0] # only one map in the sample expected = [("My Layer", 1, [("default", (), "hallo", ("#000000", 1, "None")), ("single", "1", "welt", ("#000000", 2, "None"))])] self.TestLayers(map.Layers(), expected) self.check_format() class TestLayerProjection(LoadSessionTest): file_contents = '''\ ''' def test(self): """Test loading layers with projections""" eq = self.assertEquals neq = self.assertNotEqual session = load_session(self.filename()) self.session = session map = self.session.Maps()[0] # only one map in the sample layers = map.Layers() # two layers in the sample # test layer with a named projection proj = layers[0].GetProjection() neq(proj, None) eq(proj.GetName(), "hello") eq(proj.GetParameter("proj"), "tmerc") eq(proj.GetParameter("zone"), "13") eq(proj.GetParameter("ellps"), "clrk66") # test layer with an unnamed projection proj = layers[1].GetProjection() neq(proj, None) eq(proj.GetName(), "Unknown") eq(proj.GetParameter("proj"), "lcc") eq(proj.GetParameter("lat_1"), "10") eq(proj.GetParameter("lat_2"), "20") eq(proj.GetParameter("ellps"), "clrk66") self.check_format() class TestRasterLayer(LoadSessionTest): file_contents = '''\ ''' def test(self): eq = self.assertEquals neq = self.assertNotEqual session = load_session(self.filename()) self.session = session map = self.session.Maps()[0] # only one map in the sample layer = map.Layers()[0] # one layer in the sample eq(layer.Title(), "My RasterLayer") eq(layer.Opacity(), 0.4) self.failIf(layer.Visible()) self.failUnless(filenames_equal(layer.GetImageFilename(), os.path.join(self.temp_dir(), os.pardir, os.pardir, "Data", "iceland", "island.tif"))) self.check_format() class TestJoinedTable(LoadSessionTest): file_contents = ''' ''' def setUp(self): """Extend inherited method to create the dbffile for the join""" LoadSessionTest.setUp(self) dbffile = self.temp_file_name("load_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() def test(self): """Test loading a session containing a joined table""" session = load_session(self.filename()) self.session = session tables = session.Tables() self.assertEquals(len(tables), 3) # FIXME: The tests shouldn't assume a certain order of the tables self.assertEquals(tables[0].Title(), "Some Title") self.assertEquals(tables[1].Title(), "Joined") self.assertEquals(tables[1].JoinType(), "LEFT OUTER") self.check_format() class TestLabelLayer(LoadSessionTest): # Note that the labels deliberately contain non-ascii characters to # test whether they're supported correctly. file_contents = ''' ''' def test(self): """Test loading a session with a label layer""" session = load_session(self.filename()) self.session = session label_layer = self.session.Maps()[0].LabelLayer() expected_labels = [(-21.5, 64.25, "RUINS", ALIGN_LEFT, ALIGN_CENTER), (-15.125, 64.75, internal_from_unicode(u"H\xfctte"), ALIGN_RIGHT, ALIGN_TOP), ] for label, values in zip(label_layer.Labels(), expected_labels): self.assertEquals((label.x, label.y, label.text, label.halign, label.valign), values) self.check_format() class TestPostGISLayer(LoadSessionTest): file_contents = ''' ''' def setUp(self): """Extend the inherited method to start the postgis server Furthermore, patch the file contents with the real postgis db information """ postgissupport.skip_if_no_postgis() self.server = postgissupport.get_test_server() self.postgisdb = self.server.get_default_static_data_db() self.file_contents = self.__class__.file_contents % { "dbname": self.postgisdb.dbname, "user": self.server.user_name, "port": self.server.port, "host": self.server.host} LoadSessionTest.setUp(self) def test(self): """Test loading a session containing a postgis shapestore""" session = load_session(self.filename()) self.session = session connections = session.DBConnections() self.assertEquals(len(connections), 1) conn = connections[0] for attr, value in [("host", self.server.host), ("port", str(self.server.port)), ("user", self.server.user_name), ("dbname", self.postgisdb.dbname)]: self.assertEquals(getattr(conn, attr), value) layer = session.Maps()[0].Layers()[0] self.failUnless(layer.ShapeStore().DBConnection() is conn) class TestPostGISLayerPassword(LoadSessionTest): file_contents = ''' ''' def setUp(self): """Extend the inherited method to start the postgis server Furthermore, patch the file contents with the real postgis db information """ postgissupport.skip_if_no_postgis() self.server = postgissupport.get_test_server() self.postgisdb = self.server.get_default_static_data_db() self.file_contents = self.__class__.file_contents % { "dbname": self.postgisdb.dbname, "user": self.server.user_name, "port": self.server.port, "host": self.server.host} LoadSessionTest.setUp(self) self.db_connection_callback_called = False self.server.require_authentication(True) def tearDown(self): """Extend the inherited method to switch off postgresql authentication """ self.server.require_authentication(False) LoadSessionTest.tearDown(self) def db_connection_callback(self, params, message): """Implementation of Thuban.Model.hooks.query_db_connection_parameters """ self.assertEquals(params, {"dbname": self.postgisdb.dbname, "user": self.server.user_name, "port": str(self.server.port), "host": self.server.host}) self.db_connection_callback_called = True params = params.copy() params["password"] = self.server.user_password return params def test_with_callback(self): """Test loading a session with postgis, authentication and a callback """ session = load_session(self.filename(), db_connection_callback = self.db_connection_callback) self.session = session connections = session.DBConnections() self.assertEquals(len(connections), 1) conn = connections[0] for attr, value in [("host", self.server.host), ("port", str(self.server.port)), ("user", self.server.user_name), ("dbname", self.postgisdb.dbname)]: self.assertEquals(getattr(conn, attr), value) layer = session.Maps()[0].Layers()[0] self.failUnless(layer.ShapeStore().DBConnection() is conn) self.failUnless(self.db_connection_callback_called) def test_without_callback(self): """Test loading a session with postgis, authentication and no callback """ # A password is required and there's no callback, so we should # get a ConnectionError self.assertRaises(ConnectionError, load_session, self.filename()) def test_cancel(self): """Test loading a session with postgis and cancelling authentication """ def cancel(*args): self.db_connection_callback_called = True return None # If the user cancels, i.e. if the callbakc returns None, a # LoadCancelled exception is raised. self.assertRaises(LoadCancelled, load_session, self.filename(), cancel) self.failUnless(self.db_connection_callback_called) class TestLoadError(LoadSessionTest): file_contents = '''\ ''' def test(self): """Test loading a session missing a required attribute""" # Don't use assertRaises to make sure that if a session is # actually returned it gets destroyed properly. try: self.session = load_session(self.filename()) except LoadError, value: # Check the actual messge in value to make sure the # LoadError really was about the missing attribute self.assertEquals(str(value), "Element " "(u'http://thuban.intevation.org/dtds/thuban-1.2.1.dtd'," " u'fileshapesource') requires an attribute 'filetype'") else: self.fail("Missing filetype attribute doesn't raise LoadError") class Shapefile_CallBack: def __init__(self, params): """Initialize the callback return values. params must be a dictionary of the potential CB modes (keys), with lists of tuples of return values as values. Depending on the test the callback can be called multiple, each time a return value is poped from the list """ self.params = params def s_cb(self, filename, mode = None, second_try= 0): if self.params.has_key(mode): return self.params[mode].pop(0) else: raise LoadError class TestAltPath(LoadSessionTest): """Test the various cases in the alternative path feature. The test checks the reasonable cases: - First recognition of a path error, fixed with user interaction. - First recognition of a path error, load cancelled. - Path error fixed from list, confirmed by user. - Path error fixed from list, changed by user. - Path error fixed from list, cancelled by user. - Path error wrongly fixed from list, manual fix forced. """ file_contents = '''\ ''' def checkSession(self, session): """Check if session has been loaded successfully.""" eq = self.assertEquals map = session.Maps()[0] layers = map.Layers() eq("AltPath Test session", session.Title()) eq("not the iceland map", map.Title()) eq(3,len(layers)) eq("political",layers[0].Title()) eq("roads-line",layers[1].Title()) eq("something else",layers[2].Title()) def test_01_single_path_error_fix(self): """Test single file path error fix.""" # The usual initial case s_cb = Shapefile_CallBack({ "search": [("../Data/iceland/roads-line.shp",0)], "check": [(None, None)]}) self.session = load_session(self.filename(), shapefile_callback =s_cb.s_cb) self.checkSession(self.session) def test_02_path_error_fix_from_list(self): """Test single file path error fix.""" # This represents the usual case for "from_list" s_cb = Shapefile_CallBack({ "search": [("../Data/iceland/roads-line.shp",1)], "check": [(os.path.abspath("../Data/iceland/roads-line.shp"),1)] }) self.session = load_session(self.filename(), shapefile_callback =s_cb.s_cb) self.checkSession(self.session) def test_03_single_path_error_cancelled(self): """Test alternative path cancelled.""" s_cb = Shapefile_CallBack({ "search": [(None,0)], "check": [(None, None)]}) self.assertRaises(LoadCancelled, load_session, self.filename(), None, s_cb.s_cb) def test_04_path_error_fix_from_list_cancelled(self): """Test alternative path from list cancelled.""" s_cb = Shapefile_CallBack({ "search": [("../Data/iceland/roads-line.shp",1)], "check": [(None,1)] }) self.assertRaises(LoadCancelled, load_session, self.filename(), None, s_cb.s_cb) def test_05_path_error_fix_from_list_changed(self): """Test alternative path from list changed.""" s_cb = Shapefile_CallBack({ "search": [("../Data/iceland/roads-line.shp",1)], "check": [("../Data/iceland/roads-line.shp",0)] }) self.session = load_session(self.filename(), shapefile_callback =s_cb.s_cb) self.checkSession(self.session) def test_06_path_error_fix_from_list_fails(self): """Test alternative path recovery from list.""" s_cb = Shapefile_CallBack({ "search": [("../wrong/iceland/roads-line.shp",1), ("../Data/iceland/roads-line.shp",0)], "check": [(None,None)] }) self.session = load_session(self.filename(), shapefile_callback =s_cb.s_cb) self.assertRaises(IndexError, s_cb.s_cb, None, "search") if __name__ == "__main__": support.run_tests()