diff --git a/ckanext/datastore/plugin.py b/ckanext/datastore/plugin.py index f055878b9d9..40c163b8c33 100644 --- a/ckanext/datastore/plugin.py +++ b/ckanext/datastore/plugin.py @@ -1,6 +1,5 @@ import logging import pylons -from sqlalchemy.exc import ProgrammingError import ckan.plugins as p import ckanext.datastore.logic.action as action @@ -64,12 +63,7 @@ def configure(self, config): "only database. Permission checks and the creation " "of _table_metadata are skipped.") else: - if self.config.get('debug'): - handler = log.critical - else: - def handler(message): - raise DatastoreException(message) - self._check_urls_and_permissions(handler) + self._check_urls_and_permissions(self._log_or_raise) self._create_alias_table() @@ -106,21 +100,30 @@ def new_resource_show(context, data_dict): new_resource_show._datastore_wrapped = True logic._actions['resource_show'] = new_resource_show - def _check_urls_and_permissions(self, handler): + def _log_or_raise(self, message): + if self.config.get('debug'): + log.critical(message) + else: + raise DatastoreException(message) + + def _check_urls_and_permissions(self, error_handler): # Make sure that the right permissions are set # so that no harmful queries can be made - if not self.legacy_mode and self._same_read_and_write_url(): - handler("The write and read-only database " - "connection urls are the same.") - if self._same_ckan_and_datastore_db(): - handler("CKAN and DataStore database " - "cannot be the same.") + error_handler("CKAN and DataStore database " + "cannot be the same.") + + # in legacy mode, the read and write url are ths same (both write url) + # consequently the same url check and and write privilege check + # don't make sense + if not self.legacy_mode: + if self._same_read_and_write_url(): + error_handler("The write and read-only database " + "connection urls are the same.") - if not self._read_connection_has_correct_privileges(): - handler("We have write permissions " - "on the read-only database.") + if not self._read_connection_has_correct_privileges(): + error_handler("The read-only user has write privileges.") def _is_read_only_database(self): ''' diff --git a/ckanext/datastore/tests/test_configure.py b/ckanext/datastore/tests/test_configure.py index 50a83e71057..2ab40412a8f 100644 --- a/ckanext/datastore/tests/test_configure.py +++ b/ckanext/datastore/tests/test_configure.py @@ -1,9 +1,12 @@ import unittest - -import sqlalchemy +from nose.tools import assert_equal import ckanext.datastore.plugin as plugin +# global variable used for callback tests +msg = '' +called = 0 + class TestTypeGetters(unittest.TestCase): def setUp(self): @@ -38,49 +41,75 @@ def test_check_separate_write_and_read_url(self): def test_same_ckan_and_datastore_db(self): self.p.read_url = 'postgresql://u2:pass@localhost/ckan' self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' - assert self.p._same_ckan_and_datastore_db() self.p.read_url = 'postgresql://u:pass@localhost/dt' self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' - assert not self.p._same_ckan_and_datastore_db() def test_check_urls_and_permissions(self): + global msg + + def handler(message): + global msg, called + msg = message + called += 1 + + def true_privileges_mock(): + return True + + def false_privileges_mock(): + return False + self.p.legacy_mode = False self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' self.p.write_url = 'postgresql://u:pass@localhost/ds' - self.p.read_url = 'postgresql://u:pass@localhost/ds' + self.p.read_url = 'postgresql://u2:pass@localhost/ds' + self.p._read_connection_has_correct_privileges = true_privileges_mock - def handler(message): - assert 'urls are the same' in message, message - try: - self.p._check_urls_and_permissions(handler) - except sqlalchemy.exc.OperationalError: - pass - else: - assert False + # all urls are correct + self.p._check_urls_and_permissions(handler) + assert_equal(msg, '') + assert_equal(called, 0) - self.p.ckan_url = 'postgresql://u:pass@localhost/ds' + # same url for read and write but in legacy mode self.p.legacy_mode = True + self.p.read_url = 'postgresql://u:pass@localhost/ds' + self.p._check_urls_and_permissions(handler) + assert_equal(msg, '') + assert_equal(called, 0) - def handler2(message): - assert 'cannot be the same' in message, message - try: - self.p._check_urls_and_permissions(handler2) - except sqlalchemy.exc.OperationalError: - pass - else: - assert False + # same url for read and write + self.p.legacy_mode = False + self.p._check_urls_and_permissions(handler) + assert 'urls are the same' in msg, msg + assert_equal(called, 1) + # same ckan and ds db + self.p.ckan_url = 'postgresql://u:pass@localhost/ds' self.p.read_url = 'postgresql://u2:pass@localhost/ds' + self.p._check_urls_and_permissions(handler) + assert 'cannot be the same' in msg, msg + assert_equal(called, 2) + + # have write permissions but in legacy mode + self.p.legacy_mode = True + msg = '' + self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' + self.p._read_connection_has_correct_privileges = false_privileges_mock + self.p._check_urls_and_permissions(handler) + assert_equal(msg, '') + assert_equal(called, 2) + + # have write permissions self.p.legacy_mode = False + self.p._check_urls_and_permissions(handler) + assert 'user has write privileges' in msg, msg + assert_equal(called, 3) - def handler3(message): - assert 'cannot be the same' in message, message - try: - self.p._check_urls_and_permissions(handler3) - except sqlalchemy.exc.OperationalError: - pass - else: - assert False + # everything is wrong + self.p.ckan_url = 'postgresql://u:pass@localhost/ds' + self.p.write_url = 'postgresql://u:pass@localhost/ds' + self.p.read_url = 'postgresql://u:pass@localhost/ds' + self.p._check_urls_and_permissions(handler) + assert_equal(called, 6)