diff --git a/ckanext/datastore/tests/test_configure.py b/ckanext/datastore/tests/test_configure.py index 2594e988700..019bb317f48 100644 --- a/ckanext/datastore/tests/test_configure.py +++ b/ckanext/datastore/tests/test_configure.py @@ -1,22 +1,24 @@ import unittest -import ckan.plugins as p -from nose.tools import raises +import nose.tools +import pyutilib.component.core +import ckan.plugins import ckanext.datastore.plugin as plugin - class TestConfiguration(unittest.TestCase): def setUp(self): - self.p = p.load('datastore') + self._original_plugin = ckan.plugins.unload('datastore') + pyutilib.component.core.PluginGlobals.singleton_services()[plugin.DatastorePlugin] = True + self.p = pyutilib.component.core.PluginGlobals.singleton_services()[plugin.DatastorePlugin] = ckan.plugins.load('datastore') def tearDown(self): - p.unload('datastore') + ckan.plugins.unload('datastore') + pyutilib.component.core.PluginGlobals.singleton_services()[plugin.DatastorePlugin] = self._original_plugin def test_legacy_mode_default(self): assert not self.p.legacy_mode def test_set_legacy_mode(self): - assert not self.p.legacy_mode c = { 'sqlalchemy.url': 'bar', 'ckan.datastore.write_url': 'foo' @@ -47,65 +49,67 @@ def test_same_ckan_and_datastore_db(self): self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' assert not self.p._same_ckan_and_datastore_db() + def test_setup_plugin_for_check_urls_and_permissions_tests_should_leave_the_plugin_in_a_valid_state(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() + self.p._check_urls_and_permissions() # Should be OK -class TestCheckUrlsAndPermissions(unittest.TestCase): - def setUp(self): - self.p = p.load('datastore') + def test_check_urls_and_permissions_requires_different_ckan_and_datastore_dbs(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() - self.p.legacy_mode = False + self.p._same_ckan_and_datastore_db = lambda: False + self.p._check_urls_and_permissions() # Should be OK - # initialize URLs - self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' - self.p.write_url = 'postgresql://u:pass@localhost/ds' - self.p.read_url = 'postgresql://u2:pass@localhost/ds' + self.p._same_ckan_and_datastore_db = lambda: True + nose.tools.assert_raises(InvalidUrlsOrPermissionsException, self.p._check_urls_and_permissions) - # initialize mock for privileges check - def true_privileges_mock(): - return True - self.p._read_connection_has_correct_privileges = true_privileges_mock + def test_check_urls_and_permissions_requires_different_read_and_write_urls_when_not_in_legacy_mode(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() + self.p.legacy_mode = False - def raise_datastore_exception(message): - raise plugin.DatastoreException(message) - self.p._log_or_raise = raise_datastore_exception + self.p._same_read_and_write_url = lambda: False + self.p._check_urls_and_permissions() # Should be OK - def tearDown(self): - p.unload('datastore') + self.p._same_read_and_write_url = lambda: True + nose.tools.assert_raises(InvalidUrlsOrPermissionsException, self.p._check_urls_and_permissions) - def test_everything_correct_does_not_raise(self): - self.p._check_urls_and_permissions() + def test_check_urls_and_permissions_doesnt_require_different_read_and_write_urls_when_in_legacy_mode(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() + self.p.legacy_mode = True - @raises(plugin.DatastoreException) - def test_raises_when_ckan_and_datastore_db_are_the_same(self): - self.p.read_url = 'postgresql://u2:pass@localhost/ckan' - self.p.ckan_url = 'postgresql://u:pass@localhost/ckan' + self.p._same_read_and_write_url = lambda: False + self.p._check_urls_and_permissions() # Should be OK - self.p._check_urls_and_permissions() + self.p._same_read_and_write_url = lambda: True + self.p._check_urls_and_permissions() # Should be OK - @raises(plugin.DatastoreException) - def test_raises_when_same_read_and_write_url(self): - self.p.read_url = 'postgresql://u:pass@localhost/ds' - self.p.write_url = 'postgresql://u:pass@localhost/ds' + def test_check_urls_and_permissions_requires_read_connection_with_correct_privileges_when_not_in_legacy_mode(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() + self.p.legacy_mode = False - self.p._check_urls_and_permissions() + self.p._read_connection_has_correct_privileges = lambda: True + self.p._check_urls_and_permissions() # Should be OK - def test_same_read_and_write_url_in_legacy_mode(self): - self.p.read_url = 'postgresql://u:pass@localhost/ds' - self.p.write_url = 'postgresql://u:pass@localhost/ds' + self.p._read_connection_has_correct_privileges = lambda: False + nose.tools.assert_raises(InvalidUrlsOrPermissionsException, self.p._check_urls_and_permissions) + + def test_check_urls_and_permissions_doesnt_care_about_read_connection_privileges_when_in_legacy_mode(self): + self.setUp_plugin_for_check_urls_and_permissions_tests() self.p.legacy_mode = True - self.p._check_urls_and_permissions() + self.p._read_connection_has_correct_privileges = lambda: True + self.p._check_urls_and_permissions() # Should be OK + + self.p._read_connection_has_correct_privileges = lambda: False + self.p._check_urls_and_permissions() # Should be OK - @raises(plugin.DatastoreException) - def test_raises_when_we_have_write_permissions(self): - def false_privileges_mock(): - return False - self.p._read_connection_has_correct_privileges = false_privileges_mock - self.p._check_urls_and_permissions() + def setUp_plugin_for_check_urls_and_permissions_tests(self): + def _raise_invalid_urls_or_permissions_exception(message): + raise InvalidUrlsOrPermissionsException(message) - def test_have_write_permissions_in_legacy_mode(self): - def false_privileges_mock(): - return False - self.p._read_connection_has_correct_privileges = false_privileges_mock + self.p._same_ckan_and_datastore_db = lambda: False self.p.legacy_mode = True + self.p._same_read_and_write_url = lambda: False + self.p._read_connection_has_correct_privileges = lambda: True + self.p._log_or_raise = _raise_invalid_urls_or_permissions_exception - self.p._check_urls_and_permissions() +class InvalidUrlsOrPermissionsException(Exception): pass