[#642] Ignore permission check in legacy mode and improve configurati…

…on tests

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):
         '''

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)