| 34 |
import support |
import support |
| 35 |
support.initthuban() |
support.initthuban() |
| 36 |
|
|
| 37 |
|
import postgissupport |
| 38 |
from xmlsupport import sax_eventlist |
from xmlsupport import sax_eventlist |
| 39 |
|
|
| 40 |
import dbflib |
import dbflib |
| 41 |
|
|
| 42 |
from Thuban.Model.save import save_session |
from Thuban.Model.save import save_session |
| 43 |
from Thuban.Model.load import load_session, parse_color, LoadError |
from Thuban.Model.load import load_session, parse_color, LoadError, \ |
| 44 |
|
LoadCancelled |
| 45 |
from Thuban.Model.color import Transparent |
from Thuban.Model.color import Transparent |
| 46 |
from Thuban.Model.classification import ClassGroupProperties, ClassGroupRange,\ |
from Thuban.Model.classification import ClassGroupProperties, ClassGroupRange,\ |
| 47 |
ClassGroupSingleton, ClassGroupDefault |
ClassGroupSingleton, ClassGroupDefault |
| 48 |
|
from Thuban.Model.postgisdb import ConnectionError |
| 49 |
|
|
| 50 |
def filenames_equal(name1, name2): |
def filenames_equal(name1, name2): |
| 51 |
"""Return true if the filenames name1 and name2 are equal. |
"""Return true if the filenames name1 and name2 are equal. |
| 556 |
self.check_format() |
self.check_format() |
| 557 |
|
|
| 558 |
|
|
| 559 |
|
|
| 560 |
|
class TestPostGISLayer(LoadSessionTest): |
| 561 |
|
|
| 562 |
|
file_contents = '''<?xml version="1.0" encoding="UTF-8"?> |
| 563 |
|
<!DOCTYPE session SYSTEM "thuban-0.9.dtd"> |
| 564 |
|
<session xmlns="http://thuban.intevation.org/dtds/thuban-0.9-dev.dtd" |
| 565 |
|
title="unnamed session"> |
| 566 |
|
<dbconnection port="%(port)s" host="%(host)s" user="%(user)s" |
| 567 |
|
dbtype="postgis" id="D142684948" dbname="%(dbname)s"/> |
| 568 |
|
<dbshapesource tablename="landmarks" id="D143149420" dbconn="D142684948"/> |
| 569 |
|
<map title="unnamed map"> |
| 570 |
|
<layer shapestore="D143149420" visible="true" stroke="#000000" |
| 571 |
|
title="landmarks" stroke_width="1" fill="None"/> |
| 572 |
|
</map> |
| 573 |
|
</session> |
| 574 |
|
''' |
| 575 |
|
|
| 576 |
|
def setUp(self): |
| 577 |
|
"""Extend the inherited method to start the postgis server |
| 578 |
|
|
| 579 |
|
Furthermore, patch the file contents with the real postgis db |
| 580 |
|
information |
| 581 |
|
""" |
| 582 |
|
postgissupport.skip_if_no_postgis() |
| 583 |
|
self.server = postgissupport.get_test_server() |
| 584 |
|
self.postgisdb = self.server.get_default_static_data_db() |
| 585 |
|
|
| 586 |
|
self.file_contents = self.__class__.file_contents % { |
| 587 |
|
"dbname": self.postgisdb.dbname, |
| 588 |
|
"user": self.server.user_name, |
| 589 |
|
"port": self.server.port, |
| 590 |
|
"host": self.server.host} |
| 591 |
|
LoadSessionTest.setUp(self) |
| 592 |
|
|
| 593 |
|
def test(self): |
| 594 |
|
"""Test loading a session containing a postgis shapestore""" |
| 595 |
|
session = load_session(self.filename()) |
| 596 |
|
self.session = session |
| 597 |
|
connections = session.DBConnections() |
| 598 |
|
self.assertEquals(len(connections), 1) |
| 599 |
|
conn = connections[0] |
| 600 |
|
for attr, value in [("host", self.server.host), |
| 601 |
|
("port", str(self.server.port)), |
| 602 |
|
("user", self.server.user_name), |
| 603 |
|
("dbname", self.postgisdb.dbname)]: |
| 604 |
|
self.assertEquals(getattr(conn, attr), value) |
| 605 |
|
layer = session.Maps()[0].Layers()[0] |
| 606 |
|
self.failUnless(layer.ShapeStore().DBConnection() is conn) |
| 607 |
|
|
| 608 |
|
|
| 609 |
|
class TestPostGISLayerPassword(LoadSessionTest): |
| 610 |
|
|
| 611 |
|
file_contents = '''<?xml version="1.0" encoding="UTF-8"?> |
| 612 |
|
<!DOCTYPE session SYSTEM "thuban-0.9.dtd"> |
| 613 |
|
<session xmlns="http://thuban.intevation.org/dtds/thuban-0.9-dev.dtd" |
| 614 |
|
title="unnamed session"> |
| 615 |
|
<dbconnection port="%(port)s" host="%(host)s" user="%(user)s" |
| 616 |
|
dbtype="postgis" id="D142684948" dbname="%(dbname)s"/> |
| 617 |
|
<dbshapesource tablename="landmarks" id="D143149420" dbconn="D142684948"/> |
| 618 |
|
<map title="unnamed map"> |
| 619 |
|
<layer shapestore="D143149420" visible="true" stroke="#000000" |
| 620 |
|
title="landmarks" stroke_width="1" fill="None"/> |
| 621 |
|
</map> |
| 622 |
|
</session> |
| 623 |
|
''' |
| 624 |
|
|
| 625 |
|
def setUp(self): |
| 626 |
|
"""Extend the inherited method to start the postgis server |
| 627 |
|
|
| 628 |
|
Furthermore, patch the file contents with the real postgis db |
| 629 |
|
information |
| 630 |
|
""" |
| 631 |
|
postgissupport.skip_if_no_postgis() |
| 632 |
|
self.server = postgissupport.get_test_server() |
| 633 |
|
self.postgisdb = self.server.get_default_static_data_db() |
| 634 |
|
|
| 635 |
|
self.file_contents = self.__class__.file_contents % { |
| 636 |
|
"dbname": self.postgisdb.dbname, |
| 637 |
|
"user": self.server.user_name, |
| 638 |
|
"port": self.server.port, |
| 639 |
|
"host": self.server.host} |
| 640 |
|
LoadSessionTest.setUp(self) |
| 641 |
|
|
| 642 |
|
self.db_connection_callback_called = False |
| 643 |
|
self.server.require_authentication(True) |
| 644 |
|
|
| 645 |
|
def tearDown(self): |
| 646 |
|
"""Extend the inherited method to switch off postgresql authentication |
| 647 |
|
""" |
| 648 |
|
self.server.require_authentication(False) |
| 649 |
|
LoadSessionTest.tearDown(self) |
| 650 |
|
|
| 651 |
|
def db_connection_callback(self, params, message): |
| 652 |
|
"""Implementation of Thuban.Model.hooks.query_db_connection_parameters |
| 653 |
|
""" |
| 654 |
|
self.assertEquals(params, |
| 655 |
|
{"dbname": self.postgisdb.dbname, |
| 656 |
|
"user": self.server.user_name, |
| 657 |
|
"port": str(self.server.port), |
| 658 |
|
"host": self.server.host}) |
| 659 |
|
self.db_connection_callback_called = True |
| 660 |
|
params = params.copy() |
| 661 |
|
params["password"] = self.server.user_password |
| 662 |
|
return params |
| 663 |
|
|
| 664 |
|
def test_with_callback(self): |
| 665 |
|
"""Test loading a session with postgis, authentication and a callback |
| 666 |
|
""" |
| 667 |
|
session = load_session(self.filename(), |
| 668 |
|
db_connection_callback = self.db_connection_callback) |
| 669 |
|
self.session = session |
| 670 |
|
connections = session.DBConnections() |
| 671 |
|
self.assertEquals(len(connections), 1) |
| 672 |
|
conn = connections[0] |
| 673 |
|
for attr, value in [("host", self.server.host), |
| 674 |
|
("port", str(self.server.port)), |
| 675 |
|
("user", self.server.user_name), |
| 676 |
|
("dbname", self.postgisdb.dbname)]: |
| 677 |
|
self.assertEquals(getattr(conn, attr), value) |
| 678 |
|
layer = session.Maps()[0].Layers()[0] |
| 679 |
|
self.failUnless(layer.ShapeStore().DBConnection() is conn) |
| 680 |
|
self.failUnless(self.db_connection_callback_called) |
| 681 |
|
|
| 682 |
|
def test_without_callback(self): |
| 683 |
|
"""Test loading a session with postgis, authentication and no callback |
| 684 |
|
""" |
| 685 |
|
# A password is required and there's no callback, so we should |
| 686 |
|
# get a ConnectionError |
| 687 |
|
self.assertRaises(ConnectionError, load_session, self.filename()) |
| 688 |
|
|
| 689 |
|
def test_cancel(self): |
| 690 |
|
"""Test loading a session with postgis and cancelling authentication |
| 691 |
|
""" |
| 692 |
|
def cancel(*args): |
| 693 |
|
self.db_connection_callback_called = True |
| 694 |
|
return None |
| 695 |
|
|
| 696 |
|
# If the user cancels, i.e. if the callbakc returns None, a |
| 697 |
|
# LoadCancelled exception is raised. |
| 698 |
|
self.assertRaises(LoadCancelled, |
| 699 |
|
load_session, self.filename(), cancel) |
| 700 |
|
self.failUnless(self.db_connection_callback_called) |
| 701 |
|
|
| 702 |
|
|
| 703 |
class TestLoadError(LoadSessionTest): |
class TestLoadError(LoadSessionTest): |
| 704 |
|
|
| 705 |
file_contents = '''\ |
file_contents = '''\ |
Parent Directory
| 
Revision Log
| 
Patch