Merge pull request #6 from conwetlab/ckan_2.3

Bug fixing

ckanext/oauth2/plugin.py

@@ -175,22 +175,27 @@ def login(self):
     def abort(self, status_code, detail, headers, comment):
         log.debug('abort')
 
-        # The user is authenticated, but they cannot access to a protected resource so we redirect
-        # them to the previous page. Otherwise the system will try to reauthenticate the user
-        # generating a redirect loop:
+        # If the user is authenticated, but they cannot access a protected resource, the system
+        # should redirect them to the previous page. If the user is not redirected, the system
+        # will try to reauthenticate the user generating a redirect loop:
         # (authenticate -> user not allowed -> auto log out -> authenticate -> ...)
-
-        # When the user is logged in, he/she should be redirected to the main page when
-        # the system cannot get the previous page
-        came_from_url = self._get_previous_page('/')
-
-        # Init headers and set Location
-        if headers is None:
-            headers = {}
-        headers['Location'] = came_from_url
-
-        # 302 -> Found
-        return 302, detail, headers, comment
+        # If the user is not authenticated, the system should start the authentication process
+
+        if toolkit.c.user:  # USER IS AUTHENTICATED
+            # When the user is logged in, he/she should be redirected to the main page when
+            # the system cannot get the previous page
+            came_from_url = self._get_previous_page('/')
+
+            # Init headers and set Location
+            if headers is None:
+                headers = {}
+            headers['Location'] = came_from_url
+
+            # 302 -> Found
+            return 302, detail, headers, comment
+        else:                # USER IS NOT AUTHENTICATED
+            # By not modifying the received parameters, the authentication process will start
+            return status_code, detail, headers, comment
 
     def get_auth_functions(self):
         # we need to prevent some actions being authorized.

ckanext/oauth2/tests/test_plugin.py

@@ -194,7 +194,7 @@ def authenticate_side_effect(identity):
     ])
     def test_login(self, referer=None, came_from=None, expected_referer='/dashboard'):
 
-        # The login function will check this variables
+        # The login function will check these variables
         plugin.toolkit.request.headers = {}
         plugin.toolkit.request.params = {}
 
@@ -211,23 +211,25 @@ def test_login(self, referer=None, came_from=None, expected_referer='/dashboard'
 
     @parameterized.expand([
         (),
-        (None,                        None,               None,                                     '/'),
-        (None,                        None,               {'Param1': 'value1', 'paRam2': 'value2'}, '/'),
-        ('/about',                    None,               None,                                     '/about'),
-        ('/about',                    '/ckan-admin',      None,                                     '/ckan-admin'),
-        (None,                        '/ckan-admin',      None,                                     '/ckan-admin'),
-        ('/',                         None,               None,                                     '/'),
-        ('/user/logged_out_redirect', None,               None,                                     '/'),
-        ('/',                         '/ckan-admin',      None,                                     '/ckan-admin'),
-        ('/user/logged_out_redirect', '/ckan-admin',      None,                                     '/ckan-admin'),
-        ('http://google.es',          None,               None,                                     '/'),
-        ('http://google.es',          None,               None,                                     '/'),
-        ('http://' + HOST + '/about', None,               None,                                     'http://' + HOST + '/about'),
-        ('http://' + HOST + '/about', '/other_url',       None,                                     '/other_url')
+        ('user', None,                        None,               None,                                     '/'),
+        ('user', None,                        None,               {'Param1': 'value1', 'paRam2': 'value2'}, '/'),
+        ('user', '/about',                    None,               None,                                     '/about'),
+        ('user', '/about',                    '/ckan-admin',      None,                                     '/ckan-admin'),
+        ('user', None,                        '/ckan-admin',      None,                                     '/ckan-admin'),
+        ('user', '/',                         None,               None,                                     '/'),
+        ('user', '/user/logged_out_redirect', None,               None,                                     '/'),
+        ('user', '/',                         '/ckan-admin',      None,                                     '/ckan-admin'),
+        ('user', '/user/logged_out_redirect', '/ckan-admin',      None,                                     '/ckan-admin'),
+        ('user', 'http://google.es',          None,               None,                                     '/'),
+        ('user', 'http://google.es',          None,               None,                                     '/'),
+        ('user', 'http://' + HOST + '/about', None,               None,                                     'http://' + HOST + '/about'),
+        ('user', 'http://' + HOST + '/about', '/other_url',       None,                                     '/other_url'),
+        (None,   '/about',                    '/other',           None,                                     None),
     ])
-    def test_abort(self, referer=None, came_from=None, headers=None, expected_location='/'):
+    def test_abort(self, user='user', referer=None, came_from=None, headers=None, expected_location='/'):
 
-        # The abort function will check this variable
+        # The abort function will check these variables
+        plugin.toolkit.c.user = user
         plugin.toolkit.request.host = HOST
         plugin.toolkit.request.headers = {}
         plugin.toolkit.request.params = {}
@@ -238,19 +240,28 @@ def test_abort(self, referer=None, came_from=None, headers=None, expected_locati
         if came_from:
             plugin.toolkit.request.params['came_from'] = came_from
 
-        # Save previous headers (if they are not None) and check that
-        # they are kept and not discarded
-        headers_copy = None if not headers else headers.copy()
-
         # Call the function
-        status_code, detail, new_headers, comment = self._plugin.abort(401, None, headers, None)
+        initial_status_code = 401
+        initial_detail = 'DETAIL'
+        initial_headers = None if not headers else headers.copy()
+        initial_comment = 'COMMENT'
+
+        # headers will be modified inside the function, but we should retain a copy (initial_headers)
+        status_code, detail, new_headers, comment = self._plugin.abort(initial_status_code, initial_detail, headers, initial_comment)
 
         # Verifications
-        self.assertEquals(302, status_code)
-        self.assertEquals(new_headers['Location'], expected_location)
+        self.assertEquals(initial_detail, detail)
+        self.assertEquals(initial_comment, comment)
+
+        if user:
+            self.assertEquals(302, status_code)
+            self.assertEquals(new_headers['Location'], expected_location)
+        else:
+            self.assertEquals(initial_status_code, status_code)
+            self.assertEquals(initial_headers, new_headers)
 
         # Check previous headers if they were not None
-        if headers_copy:
-            for header in headers_copy:
+        if initial_headers:
+            for header in initial_headers:
                 self.assertIn(header, new_headers)
-                self.assertEquals(headers_copy[header], new_headers[header])
+                self.assertEquals(initial_headers[header], new_headers[header])

ckanext/oauth2/tests/test_selenium.py

@@ -25,8 +25,13 @@
 from selenium import webdriver
 from subprocess import Popen
 
+IDM_URL = "https://account.lab.fiware.org"
+FILAB2_MAIL = "filab2@mailinator.com"
+FILAB3_MAIL = "filab3@mailinator.com"
+FILAB_PASSWORD = "filab1234"
 
-class BasicLoginDifferentReferer(unittest.TestCase):
+
+class IntegrationTest(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
@@ -51,31 +56,35 @@ def tearDown(self):
         self.driver.quit()
         self.assertEqual([], self.verificationErrors)
 
-    def _log_in(self, referer, user_name, password):
+    def _introduce_log_in_parameters(self, username=FILAB2_MAIL, password=FILAB_PASSWORD):
         driver = self.driver
-        driver.get(referer)
-        driver.find_element_by_link_text("Log in").click()
         driver.find_element_by_id("user_email").clear()
-        driver.find_element_by_id("user_email").send_keys(user_name)
+        driver.find_element_by_id("user_email").send_keys(username)
         driver.find_element_by_id("user_password").clear()
         driver.find_element_by_id("user_password").send_keys(password)
         driver.find_element_by_name("commit").click()
 
+    def _log_in(self, referer, username=FILAB2_MAIL, password=FILAB_PASSWORD):
+        driver = self.driver
+        driver.get(referer)
+        driver.find_element_by_link_text("Log in").click()
+        self._introduce_log_in_parameters(username, password)
+
     def test_basic_login(self):
         driver = self.driver
-        self._log_in(self.base_url, "filab2@mailinator.com", "filab1234")
+        self._log_in(self.base_url)
         self.assertEqual("filab2 Example User", driver.find_element_by_link_text("filab2 Example User").text)
         self.assertEqual(self.base_url + 'dashboard', driver.current_url)
         driver.find_element_by_link_text("About").click()
         self.assertEqual("filab2 Example User", driver.find_element_by_css_selector("span.username").text)
         self.assertEqual(self.base_url + "about", driver.current_url)
         driver.find_element_by_css_selector("a[title=\"Edit settings\"]").click()
         time.sleep(3)   # Wait the OAuth2 Server to return the page
-        assert driver.current_url.startswith("https://account.lab.fiware.org/settings")
+        assert driver.current_url.startswith(IDM_URL + "/settings")
 
     def test_basic_login_different_referer(self):
         driver = self.driver
-        self._log_in(self.base_url + "about", "filab2@mailinator.com", "filab1234")
+        self._log_in(self.base_url + "about")
         self.assertEqual("filab2 Example User", driver.find_element_by_css_selector("span.username").text)
         self.assertEqual(self.base_url + "about", driver.current_url)
         driver.find_element_by_link_text("Datasets").click()
@@ -85,29 +94,43 @@ def test_basic_login_different_referer(self):
     def test_user_denies_ckan_access_to_their_account(self):
         # User rejects the application to access his/her information
         driver = self.driver
-        self._log_in(self.base_url, "filab3@mailinator.com", "filab1234")
+        self._log_in(self.base_url, FILAB3_MAIL)
         driver.find_element_by_name("cancel").click()
         assert driver.find_element_by_xpath("//div/div/div/div").text.startswith("The end-user or authorization server denied the request.")
 
     def test_user_access_unauthorized_page(self):
         driver = self.driver
-        self._log_in(self.base_url, "filab2@mailinator.com", "filab1234")
+        self._log_in(self.base_url)
+        time.sleep(3)   # Wait until the log in proccess is completed
         driver.get(self.base_url + "ckan-admin")
 
         # Check that the user has been redirected to the main page
         self.assertEquals(self.base_url, driver.current_url)
         # Check that an error message is shown
         assert driver.find_element_by_xpath("//div/div/div/div").text.startswith("Need to be system administrator to administer")
 
+    def test_user_access_unauthorized_page_not_logged(self):
+        driver = self.driver
+        driver.get(self.base_url + "ckan-admin")
+
+        # Check that the user has been redirected to the log in page
+        self.assertEquals(IDM_URL + "/users/sign_in", driver.current_url)
+
+        # Log in the user
+        self._introduce_log_in_parameters()
+
+        # Check that the user is logged in now
+        self.assertEqual("filab2 Example User", driver.find_element_by_css_selector("span.username").text)
+
     def test_register_btn(self):
         driver = self.driver
         driver.get(self.base_url)
         driver.find_element_by_link_text("Register").click()
-        self.assertEqual("https://account.lab.fiware.org/users/sign_up", driver.current_url)
+        self.assertEqual(IDM_URL + "/users/sign_up", driver.current_url)
 
     @parameterized.expand([
-        ("user/register", "https://account.lab.fiware.org/users/sign_up"),
-        ("user/reset", "https://account.lab.fiware.org/users/password/new")
+        ("user/register", IDM_URL + "/users/sign_up"),
+        ("user/reset", IDM_URL + "/users/password/new")
     ])
     def test_register(self, action, expected_url):
         driver = self.driver