Synnefo

# Copyright 2011 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from astakos.im.tests.common import *

class ShibbolethTests(TestCase):

    """

    Testing shibboleth authentication.

    """

    def setUp(self):

        self.client = ShibbolethClient()

        astakos_settings.IM_MODULES = ['local', 'shibboleth']

        astakos_settings.MODERATION_ENABLED = True

    def tearDown(self):

        AstakosUser.objects.all().delete()

    @im_settings(FORCE_PROFILE_UPDATE=False)

    def test_create_account(self):

        client = ShibbolethClient()

        # shibboleth views validation

        # eepn required

        r = client.get('/im/login/shibboleth?', follow=True)

        self.assertContains(r, messages.SHIBBOLETH_MISSING_EPPN % {

            'domain': astakos_settings.BASEURL,

            'contact_email': settings.CONTACT_EMAIL

        })

        client.set_tokens(eppn="kpapeppn")

        astakos_settings.SHIBBOLETH_REQUIRE_NAME_INFO = True

        # shibboleth user info required

        r = client.get('/im/login/shibboleth?', follow=True)

        self.assertContains(r, messages.SHIBBOLETH_MISSING_NAME)

        astakos_settings.SHIBBOLETH_REQUIRE_NAME_INFO = False

        # shibboleth logged us in

        client.set_tokens(mail="kpap@synnefo.org", eppn="kpapeppn",

                          cn="Kostas Papadimitriou",

                          ep_affiliation="Test Affiliation")

        r = client.get('/im/login/shibboleth?', follow=True)

        token = PendingThirdPartyUser.objects.get().token

        self.assertRedirects(r, '/im/signup?third_party_token=%s' % token)

        self.assertEqual(r.status_code, 200)

        # a new pending user created

        pending_user = PendingThirdPartyUser.objects.get(

            third_party_identifier="kpapeppn")

        self.assertEqual(PendingThirdPartyUser.objects.count(), 1)

        # keep the token for future use

        token = pending_user.token

        # from now on no shibboleth headers are sent to the server

        client.reset_tokens()

        # this is the old way, it should fail, to avoid pending user take over

        r = client.get('/im/shibboleth/signup/%s' % pending_user.username)

        self.assertEqual(r.status_code, 404)

        # this is the signup unique url associated with the pending user

        # created

        r = client.get('/im/signup/?third_party_token=%s' % token)

        identifier = pending_user.third_party_identifier

        post_data = {'third_party_identifier': identifier,

                     'first_name': 'Kostas',

                     'third_party_token': token,

                     'last_name': 'Mitroglou',

                     'provider': 'shibboleth'}

        signup_url = reverse('signup')

        # invlid email

        post_data['email'] = 'kpap'

        r = client.post(signup_url, post_data)

        self.assertContains(r, token)

        # existing email

        existing_user = get_local_user('test@test.com')

        post_data['email'] = 'test@test.com'

        r = client.post(signup_url, post_data)

        self.assertContains(r, messages.EMAIL_USED)

        existing_user.delete()

        # and finally a valid signup

        post_data['email'] = 'kpap@synnefo.org'

        r = client.post(signup_url, post_data, follow=True)

        self.assertContains(r, messages.VERIFICATION_SENT)

        # entires commited as expected

        self.assertEqual(AstakosUser.objects.count(), 1)

        self.assertEqual(AstakosUserAuthProvider.objects.count(), 1)

        self.assertEqual(PendingThirdPartyUser.objects.count(), 0)

        # provider info stored

        provider = AstakosUserAuthProvider.objects.get(module="shibboleth")

        self.assertEqual(provider.affiliation, 'Test Affiliation')

        self.assertEqual(provider.info, {u'email': u'kpap@synnefo.org',

                                         u'eppn': u'kpapeppn',

                                         u'name': u'Kostas Papadimitriou'})

        # login (not activated yet)

        client.set_tokens(mail="kpap@synnefo.org", eppn="kpapeppn",

                          cn="Kostas Papadimitriou", )

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertContains(r, 'is pending moderation')

        # admin activates the user

        u = AstakosUser.objects.get(username="kpap@synnefo.org")

        backend = activation_backends.get_backend()

        activation_result = backend.verify_user(u, u.verification_code)

        activation_result = backend.accept_user(u)

        self.assertFalse(activation_result.is_error())

        backend.send_result_notifications(activation_result, u)

        self.assertEqual(u.is_active, True)

        # we see our profile

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertRedirects(r, '/im/landing')

        self.assertEqual(r.status_code, 200)

    def test_existing(self):

        """

        Test adding of third party login to an existing account

        """

        # this is our existing user

        existing_user = get_local_user('kpap@synnefo.org')

        existing_inactive = get_local_user('kpap-inactive@synnefo.org')

        existing_inactive.is_active = False

        existing_inactive.save()

        existing_unverified = get_local_user('kpap-unverified@synnefo.org')

        existing_unverified.is_active = False

        existing_unverified.activation_sent = None

        existing_unverified.email_verified = False

        existing_unverified.is_verified = False

        existing_unverified.save()

        client = ShibbolethClient()

        # shibboleth logged us in, notice that we use different email

        client.set_tokens(mail="kpap@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimitriou", )

        r = client.get("/im/login/shibboleth?", follow=True)

        # a new pending user created

        pending_user = PendingThirdPartyUser.objects.get()

        token = pending_user.token

        self.assertEqual(PendingThirdPartyUser.objects.count(), 1)

        pending_key = pending_user.token

        client.reset_tokens()

        self.assertRedirects(r, "/im/signup?third_party_token=%s" % token)

        form = r.context['login_form']

        signupdata = copy.copy(form.initial)

        signupdata['email'] = 'kpap@synnefo.org'

        signupdata['third_party_token'] = token

        signupdata['provider'] = 'shibboleth'

        signupdata.pop('id', None)

        # the email exists to another user

        r = client.post("/im/signup", signupdata)

        self.assertContains(r, "There is already an account with this email "

                               "address")

        # change the case, still cannot create

        signupdata['email'] = 'KPAP@synnefo.org'

        r = client.post("/im/signup", signupdata)

        self.assertContains(r, "There is already an account with this email "

                               "address")

        # inactive user

        signupdata['email'] = 'KPAP-inactive@synnefo.org'

        r = client.post("/im/signup", signupdata)

        self.assertContains(r, "There is already an account with this email "

                               "address")

        # unverified user, this should pass, old entry will be deleted

        signupdata['email'] = 'KAPAP-unverified@synnefo.org'

        r = client.post("/im/signup", signupdata)

        post_data = {'password': 'password',

                     'username': 'kpap@synnefo.org'}

        r = client.post('/im/local', post_data, follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        client.set_tokens(mail="kpap@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimitriou", )

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertContains(r, "enabled for this account")

        client.reset_tokens()

        user = existing_user

        self.assertTrue(user.has_auth_provider('shibboleth'))

        self.assertTrue(user.has_auth_provider('local',

                                               auth_backend='astakos'))

        client.logout()

        # look Ma, i can login with both my shibboleth and local account

        client.set_tokens(mail="kpap@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimitriou")

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        self.assertTrue(r.context['request'].user.email == "kpap@synnefo.org")

        self.assertRedirects(r, '/im/landing')

        self.assertEqual(r.status_code, 200)

        client.logout()

        client.reset_tokens()

        # logged out

        r = client.get("/im/profile", follow=True)

        self.assertFalse(r.context['request'].user.is_authenticated())

        # login with local account also works

        post_data = {'password': 'password',

                     'username': 'kpap@synnefo.org'}

        r = self.client.post('/im/local', post_data, follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        self.assertTrue(r.context['request'].user.email == "kpap@synnefo.org")

        self.assertRedirects(r, '/im/landing')

        self.assertEqual(r.status_code, 200)

        # cannot add the same eppn

        client.set_tokens(mail="secondary@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimitriou", )

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertRedirects(r, '/im/landing')

        self.assertTrue(r.status_code, 200)

        self.assertEquals(existing_user.auth_providers.count(), 2)

        # only one allowed by default

        client.set_tokens(mail="secondary@shibboleth.gr", eppn="kpapeppn2",

                          cn="Kostas Papadimitriou", ep_affiliation="affil2")

        prov = auth_providers.get_provider('shibboleth')

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertContains(r, "Failed to add")

        self.assertRedirects(r, '/im/profile')

        self.assertTrue(r.status_code, 200)

        self.assertEquals(existing_user.auth_providers.count(), 2)

        client.logout()

        client.reset_tokens()

        # cannot login with another eppn

        client.set_tokens(mail="kpap@synnefo.org", eppn="kpapeppninvalid",

                          cn="Kostas Papadimitriou")

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertFalse(r.context['request'].user.is_authenticated())

        # cannot

        # lets remove local password

        user = AstakosUser.objects.get(username="kpap@synnefo.org",

                                       email="kpap@synnefo.org")

        remove_local_url = user.get_auth_provider('local').get_remove_url

        remove_shibbo_url = user.get_auth_provider('shibboleth',

                                                   'kpapeppn').get_remove_url

        client.set_tokens(mail="kpap@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimtriou")

        r = client.get("/im/login/shibboleth?", follow=True)

        client.reset_tokens()

        # only POST is allowed (for CSRF protection)

        r = client.get(remove_local_url, follow=True)

        self.assertEqual(r.status_code, 405)

        r = client.post(remove_local_url, follow=True)

        # 2 providers left

        self.assertEqual(user.auth_providers.count(), 1)

        # cannot remove last provider

        r = client.post(remove_shibbo_url)

        self.assertEqual(r.status_code, 403)

        self.client.logout()

        # cannot login using local credentials (notice we use another client)

        post_data = {'password': 'password',

                     'username': 'kpap@synnefo.org'}

        r = self.client.post('/im/local', post_data, follow=True)

        self.assertFalse(r.context['request'].user.is_authenticated())

        # we can reenable the local provider by setting a password

        r = client.get("/im/password_change", follow=True)

        r = client.post("/im/password_change", {'new_password1': '111',

                                                'new_password2': '111'},

                        follow=True)

        user = r.context['request'].user

        self.assertTrue(user.has_auth_provider('local'))

        self.assertTrue(user.has_auth_provider('shibboleth'))

        self.assertTrue(user.check_password('111'))

        self.assertTrue(user.has_usable_password())

        self.client.logout()

        # now we can login

        post_data = {'password': '111',

                     'username': 'kpap@synnefo.org'}

        r = self.client.post('/im/local', post_data, follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        client.reset_tokens()

        # we cannot take over another shibboleth identifier

        user2 = get_local_user('another@synnefo.org')

        user2.add_auth_provider('shibboleth', identifier='existingeppn')

        # login

        client.set_tokens(mail="kpap@shibboleth.gr", eppn="kpapeppn",

                          cn="Kostas Papadimitriou")

        r = client.get("/im/login/shibboleth?", follow=True)

        # try to assign existing shibboleth identifier of another user

        client.set_tokens(mail="kpap_second@shibboleth.gr",

                          eppn="existingeppn", cn="Kostas Papadimitriou")

        r = client.get("/im/login/shibboleth?", follow=True)

        self.assertContains(r, "is already in use")

class TestLocal(TestCase):

    def setUp(self):

        settings.ADMINS = (('admin', 'support@cloud.synnefo.org'),)

        settings.SERVER_EMAIL = 'no-reply@synnefo.org'

        self._orig_moderation = astakos_settings.MODERATION_ENABLED

        settings.ASTAKOS_MODERATION_ENABLED = True

    def tearDown(self):

        settings.ASTAKOS_MODERATION_ENABLED = self._orig_moderation

        AstakosUser.objects.all().delete()

    def test_no_moderation(self):

        # disable moderation

        astakos_settings.MODERATION_ENABLED = False

        # create a new user

        r = self.client.get("/im/signup")

        self.assertEqual(r.status_code, 200)

        data = {'email': 'kpap@synnefo.org', 'password1': 'password',

                'password2': 'password', 'first_name': 'Kostas',

                'last_name': 'Mitroglou', 'provider': 'local'}

        r = self.client.post("/im/signup", data)

        # user created

        self.assertEqual(AstakosUser.objects.count(), 1)

        user = AstakosUser.objects.get(username="kpap@synnefo.org",

                                       email="kpap@synnefo.org")

        self.assertEqual(user.username, 'kpap@synnefo.org')

        self.assertEqual(user.has_auth_provider('local'), True)

        self.assertFalse(user.is_active)

        # user (but not admin) gets notified

        self.assertEqual(len(get_mailbox('support@cloud.synnefo.org')), 0)

        self.assertEqual(len(get_mailbox('kpap@synnefo.org')), 1)

        astakos_settings.MODERATION_ENABLED = True

    def test_email_case(self):

        data = {

            'email': 'kPap@synnefo.org',

            'password1': '1234',

            'password2': '1234'

        }

        form = forms.LocalUserCreationForm(data)

        self.assertTrue(form.is_valid())

        user = form.save()

        form.store_user(user, {})

        u = AstakosUser.objects.get()

        self.assertEqual(u.email, 'kPap@synnefo.org')

        self.assertEqual(u.username, 'kpap@synnefo.org')

        u.is_active = True

        u.email_verified = True

        u.save()

        data = {'username': 'kpap@synnefo.org', 'password': '1234'}

        login = forms.LoginForm(data=data)

        self.assertTrue(login.is_valid())

        data = {'username': 'KpaP@synnefo.org', 'password': '1234'}

        login = forms.LoginForm(data=data)

        self.assertTrue(login.is_valid())

        data = {

            'email': 'kpap@synnefo.org',

            'password1': '1234',

            'password2': '1234'

        }

        form = forms.LocalUserCreationForm(data)

        self.assertFalse(form.is_valid())

    @im_settings(HELPDESK=(('support', 'support@synnefo.org'),),

                 FORCE_PROFILE_UPDATE=False, MODERATION_ENABLED=True)

    def test_local_provider(self):

        self.helpdesk_email = astakos_settings.HELPDESK[0][1]

        # create a user

        r = self.client.get("/im/signup")

        self.assertEqual(r.status_code, 200)

        data = {'email': 'kpap@synnefo.org', 'password1': 'password',

                'password2': 'password', 'first_name': 'Kostas',

                'last_name': 'Mitroglou', 'provider': 'local'}

        r = self.client.post("/im/signup", data)

        # user created

        self.assertEqual(AstakosUser.objects.count(), 1)

        user = AstakosUser.objects.get(username="kpap@synnefo.org",

                                       email="kpap@synnefo.org")

        self.assertEqual(user.username, 'kpap@synnefo.org')

        self.assertEqual(user.has_auth_provider('local'), True)

        self.assertFalse(user.is_active)  # not activated

        self.assertFalse(user.email_verified)  # not verified

        self.assertTrue(user.activation_sent)  # activation automatically sent

        self.assertFalse(user.moderated)

        self.assertFalse(user.email_verified)

        # admin gets notified and activates the user from the command line

        self.assertEqual(len(get_mailbox('kpap@synnefo.org')), 1)

        r = self.client.post('/im/local', {'username': 'kpap@synnefo.org',

                                           'password': 'password'},

                             follow=True)

        self.assertContains(r, messages.VERIFICATION_SENT)

        backend = activation_backends.get_backend()

        user = AstakosUser.objects.get(username="kpap@synnefo.org")

        backend.send_user_verification_email(user)

        # user activation fields updated and user gets notified via email

        user = AstakosUser.objects.get(pk=user.pk)

        self.assertTrue(user.activation_sent)

        self.assertFalse(user.email_verified)

        self.assertFalse(user.is_active)

        self.assertEqual(len(get_mailbox('kpap@synnefo.org')), 2)

        # user forgot she got registered and tries to submit registration

        # form. Notice the upper case in email

        data = {'email': 'KPAP@synnefo.org', 'password1': 'password',

                'password2': 'password', 'first_name': 'Kostas',

                'last_name': 'Mitroglou', 'provider': 'local'}

        r = self.client.post("/im/signup", data, follow=True)

        self.assertRedirects(r, reverse('index'))

        self.assertContains(r, messages.VERIFICATION_SENT)

        user = AstakosUser.objects.get()

        # previous user replaced

        self.assertTrue(user.activation_sent)

        self.assertFalse(user.email_verified)

        self.assertFalse(user.is_active)

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 1)

        # hmmm, email exists; lets request a password change

        r = self.client.get('/im/local/password_reset')

        self.assertEqual(r.status_code, 200)

        data = {'email': 'kpap@synnefo.org'}

        r = self.client.post('/im/local/password_reset', data, follow=True)

        # she can't because account is not active yet

        self.assertContains(r, 'pending activation')

        # moderation is enabled and an activation email has already been sent

        # so user can trigger resend of the activation email

        r = self.client.get('/im/send/activation/%d' % user.pk, follow=True)

        self.assertContains(r, 'has been sent to your email address.')

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 2)

        # also she cannot login

        data = {'username': 'kpap@synnefo.org', 'password': 'password'}

        r = self.client.post('/im/local', data, follow=True)

        self.assertContains(r, 'Resend activation')

        self.assertFalse(r.context['request'].user.is_authenticated())

        self.assertFalse('_pithos2_a' in self.client.cookies)

        # user sees the message and resends activation

        r = self.client.get('/im/send/activation/%d' % user.pk, follow=True)

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 3)

        # logged in user cannot activate another account

        tmp_user = get_local_user("test_existing_user@synnefo.org")

        tmp_client = Client()

        tmp_client.login(username="test_existing_user@synnefo.org",

                         password="password")

        r = tmp_client.get(user.get_activation_url(), follow=True)

        self.assertContains(r, messages.LOGGED_IN_WARNING)

        r = self.client.get(user.get_activation_url(), follow=True)

        # previous code got invalidated

        self.assertEqual(r.status_code, 404)

        user = AstakosUser.objects.get(pk=user.pk)

        self.assertEqual(len(get_mailbox(self.helpdesk_email)), 0)

        r = self.client.get(user.get_activation_url(), follow=True)

        self.assertRedirects(r, reverse('index'))

        # user sees that account is pending approval from admins

        self.assertContains(r, messages.NOTIFICATION_SENT)

        self.assertEqual(len(get_mailbox(self.helpdesk_email)), 1)

        user = AstakosUser.objects.get(email="KPAP@synnefo.org")

        result = backend.handle_moderation(user)

        backend.send_result_notifications(result, user)

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 4)

        self.assertEqual(len(get_mailbox(self.helpdesk_email)), 2)

        user = AstakosUser.objects.get(email="KPAP@synnefo.org")

        r = self.client.get('/im/profile', follow=True)

        self.assertFalse(r.context['request'].user.is_authenticated())

        self.assertFalse('_pithos2_a' in self.client.cookies)

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 4)

        user = AstakosUser.objects.get(pk=user.pk)

        r = self.client.post('/im/local', {'username': 'kpap@synnefo.org',

                                           'password': 'password'},

                             follow=True)

        # user activated and logged in, token cookie set

        self.assertTrue(r.context['request'].user.is_authenticated())

        self.assertTrue('_pithos2_a' in self.client.cookies)

        cookies = self.client.cookies

        self.assertTrue(quote(user.auth_token) in

                        cookies.get('_pithos2_a').value)

        r = self.client.get('/im/logout', follow=True)

        r = self.client.get('/im/')

        # user logged out, token cookie removed

        self.assertFalse(r.context['request'].user.is_authenticated())

        self.assertFalse(self.client.cookies.get('_pithos2_a').value)

        #https://docs.djangoproject.com/en/dev/topics/testing/#persistent-state

        del self.client.cookies['_pithos2_a']

        # user can login

        r = self.client.post('/im/local', {'username': 'kpap@synnefo.org',

                                           'password': 'password'},

                             follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        self.assertTrue('_pithos2_a' in self.client.cookies)

        cookies = self.client.cookies

        self.assertTrue(quote(user.auth_token) in

                        cookies.get('_pithos2_a').value)

        self.client.get('/im/logout', follow=True)

        # user forgot password

        old_pass = user.password

        r = self.client.get('/im/local/password_reset')

        self.assertEqual(r.status_code, 200)

        r = self.client.post('/im/local/password_reset', {'email':

                                                          'kpap@synnefo.org'})

        self.assertEqual(r.status_code, 302)

        # email sent

        self.assertEqual(len(get_mailbox('KPAP@synnefo.org')), 5)

        # user visits change password link

        user = AstakosUser.objects.get(pk=user.pk)

        r = self.client.get(user.get_password_reset_url())

        r = self.client.post(user.get_password_reset_url(),

                             {'new_password1': 'newpass',

                              'new_password2': 'newpass'})

        user = AstakosUser.objects.get(pk=user.pk)

        self.assertNotEqual(old_pass, user.password)

        # old pass is not usable

        r = self.client.post('/im/local', {'username': 'kpap@synnefo.org',

                                           'password': 'password'})

        self.assertContains(r, 'Please enter a correct username and password')

        r = self.client.post('/im/local', {'username': 'kpap@synnefo.org',

                                           'password': 'newpass'},

                             follow=True)

        self.assertTrue(r.context['request'].user.is_authenticated())

        self.client.logout()

        # tests of special local backends

        user = AstakosUser.objects.get(pk=user.pk)

        user.auth_providers.filter(module='local').update(auth_backend='ldap')

        user.save()

        # non astakos local backends do not support password reset

        r = self.client.get('/im/local/password_reset')

        self.assertEqual(r.status_code, 200)

        r = self.client.post('/im/local/password_reset', {'email':

                                                          'kpap@synnefo.org'})

        # she can't because account is not active yet

        self.assertContains(r, "Changing password is not")

class UserActionsTests(TestCase):

    def test_email_change(self):

        # to test existing email validation

        get_local_user('existing@synnefo.org')

        # local user

        user = get_local_user('kpap@synnefo.org')

        # login as kpap

        self.client.login(username='kpap@synnefo.org', password='password')

        r = self.client.get('/im/profile', follow=True)

        user = r.context['request'].user

        self.assertTrue(user.is_authenticated())

        # change email is enabled

        r = self.client.get('/im/email_change')

        self.assertEqual(r.status_code, 200)

        self.assertFalse(user.email_change_is_pending())

        # request email change to an existing email fails

        data = {'new_email_address': 'existing@synnefo.org'}

        r = self.client.post('/im/email_change', data)

        self.assertContains(r, messages.EMAIL_USED)

        # proper email change

        data = {'new_email_address': 'kpap@gmail.com'}

        r = self.client.post('/im/email_change', data, follow=True)

        self.assertRedirects(r, '/im/profile')

        self.assertContains(r, messages.EMAIL_CHANGE_REGISTERED)

        change1 = EmailChange.objects.get()

        # user sees a warning

        r = self.client.get('/im/email_change')

        self.assertEqual(r.status_code, 200)

        self.assertContains(r, messages.PENDING_EMAIL_CHANGE_REQUEST)

        self.assertTrue(user.email_change_is_pending())

        # link was sent

        self.assertEqual(len(get_mailbox('kpap@synnefo.org')), 0)

        self.assertEqual(len(get_mailbox('kpap@gmail.com')), 1)

        # proper email change

        data = {'new_email_address': 'kpap@yahoo.com'}

        r = self.client.post('/im/email_change', data, follow=True)

        self.assertRedirects(r, '/im/profile')

        self.assertContains(r, messages.EMAIL_CHANGE_REGISTERED)

        self.assertEqual(len(get_mailbox('kpap@synnefo.org')), 0)

        self.assertEqual(len(get_mailbox('kpap@yahoo.com')), 1)

        change2 = EmailChange.objects.get()

        r = self.client.get(change1.get_url())

        self.assertEquals(r.status_code, 404)

        self.client.logout()

        invalid_client = Client()

        r = invalid_client.post('/im/local?',

                                {'username': 'existing@synnefo.org',

                                 'password': 'password'})

        r = invalid_client.get(change2.get_url(), follow=True)

        self.assertEquals(r.status_code, 403)

        r = self.client.post('/im/local?next=' + change2.get_url(),

                             {'username': 'kpap@synnefo.org',

                              'password': 'password',

                              'next': change2.get_url()},

                             follow=True)

        self.assertRedirects(r, '/im/profile')

        user = r.context['request'].user

        self.assertEquals(user.email, 'kpap@yahoo.com')

        self.assertEquals(user.username, 'kpap@yahoo.com')

        self.client.logout()

        r = self.client.post('/im/local?next=' + change2.get_url(),

                             {'username': 'kpap@synnefo.org',

                              'password': 'password',

                              'next': change2.get_url()},

                             follow=True)

        self.assertContains(r, "Please enter a correct username and password")

        self.assertEqual(user.emailchanges.count(), 0)

class TestAuthProviderViews(TestCase):

    def tearDown(self):

        AstakosUser.objects.all().delete()

    @shibboleth_settings(CREATION_GROUPS_POLICY=['academic-login'],

                         AUTOMODERATE_POLICY=True)

    @im_settings(IM_MODULES=['shibboleth', 'local'], MODERATION_ENABLED=True,

                 FORCE_PROFILE_UPDATE=False)

    def test_user(self):

        Profile = AuthProviderPolicyProfile

        Pending = PendingThirdPartyUser

        User = AstakosUser

        User.objects.create(email="newuser@synnefo.org")

        get_local_user("olduser@synnefo.org")

        cl_olduser = ShibbolethClient()

        get_local_user("olduser2@synnefo.org")

        ShibbolethClient()

        cl_newuser = ShibbolethClient()

        cl_newuser2 = Client()

        academic_group, created = Group.objects.get_or_create(

            name='academic-login')

        academic_users = academic_group.user_set

        assert created

        policy_only_academic = Profile.objects.add_policy('academic_strict',

                                                          'shibboleth',

                                                          academic_group,

                                                          exclusive=True,

                                                          login=False,

                                                          add=False)

        # new academic user

        self.assertFalse(academic_users.filter(email='newuser@synnefo.org'))

        cl_newuser.set_tokens(eppn="newusereppn")

        r = cl_newuser.get('/im/login/shibboleth?', follow=True)

        pending = Pending.objects.get()

        identifier = pending.third_party_identifier

        signup_data = {'third_party_identifier': identifier,

                       'first_name': 'Academic',

                       'third_party_token': pending.token,

                       'last_name': 'New User',

                       'provider': 'shibboleth'}

        r = cl_newuser.post('/im/signup', signup_data)

        self.assertContains(r, "This field is required", )

        signup_data['email'] = 'olduser@synnefo.org'

        r = cl_newuser.post('/im/signup', signup_data)

        self.assertContains(r, "already an account with this email", )

        signup_data['email'] = 'newuser@synnefo.org'

        r = cl_newuser.post('/im/signup', signup_data, follow=True)

        r = cl_newuser.post('/im/signup', signup_data, follow=True)

        self.assertEqual(r.status_code, 404)

        newuser = User.objects.get(email="newuser@synnefo.org")

        activation_link = newuser.get_activation_url()

        self.assertTrue(academic_users.get(email='newuser@synnefo.org'))

        # new non-academic user

        signup_data = {'first_name': 'Non Academic',

                       'last_name': 'New User',

                       'provider': 'local',

                       'password1': 'password',

                       'password2': 'password'}

        signup_data['email'] = 'olduser@synnefo.org'

        r = cl_newuser2.post('/im/signup', signup_data)

        self.assertContains(r, 'There is already an account with this '

                               'email address')

        signup_data['email'] = 'newuser@synnefo.org'

        r = cl_newuser2.post('/im/signup/', signup_data)

        self.assertFalse(academic_users.filter(email='newuser@synnefo.org'))

        r = self.client.get(activation_link, follow=True)

        self.assertEqual(r.status_code, 404)

        newuser = User.objects.get(email="newuser@synnefo.org")

        self.assertTrue(newuser.activation_sent)

        # activation sent, user didn't open verification url so additional

        # registrations invalidate the previous signups.

        self.assertFalse(academic_users.filter(email='newuser@synnefo.org'))

        r = cl_newuser.get('/im/login/shibboleth?', follow=True)

        pending = Pending.objects.get()

        identifier = pending.third_party_identifier

        signup_data = {'third_party_identifier': identifier,

                       'first_name': 'Academic',

                       'third_party_token': pending.token,

                       'last_name': 'New User',

                       'provider': 'shibboleth'}

        signup_data['email'] = 'newuser@synnefo.org'

        r = cl_newuser.post('/im/signup', signup_data)

        self.assertEqual(r.status_code, 302)

        newuser = User.objects.get(email="newuser@synnefo.org")

        self.assertTrue(newuser.activation_sent)

        activation_link = newuser.get_activation_url()

        self.assertTrue(academic_users.get(email='newuser@synnefo.org'))

        r = cl_newuser.get(newuser.get_activation_url(), follow=True)

        self.assertRedirects(r, '/im/landing')

        newuser = User.objects.get(email="newuser@synnefo.org")

        self.assertEqual(newuser.is_active, True)

        self.assertEqual(newuser.email_verified, True)

        cl_newuser.logout()

        # cannot reactivate if suspended

        newuser.is_active = False

        newuser.save()

        r = cl_newuser.get(newuser.get_activation_url())

        newuser = User.objects.get(email="newuser@synnefo.org")

        self.assertFalse(newuser.is_active)

        # release suspension

        newuser.is_active = True

        newuser.save()

        cl_newuser.get('/im/login/shibboleth?', follow=True)

        local = auth.get_provider('local', newuser)

        self.assertEqual(local.get_add_policy, False)

        self.assertEqual(local.get_login_policy, False)

        r = cl_newuser.get(local.get_add_url, follow=True)

        self.assertRedirects(r, '/im/profile')

        self.assertContains(r, 'disabled for your')

        cl_olduser.login(username='olduser@synnefo.org', password="password")

        r = cl_olduser.get('/im/profile', follow=True)

        self.assertEqual(r.status_code, 200)

        r = cl_olduser.get('/im/login/shibboleth?', follow=True)

        self.assertContains(r, 'Your request is missing a unique token')

        cl_olduser.set_tokens(eppn="newusereppn")

        r = cl_olduser.get('/im/login/shibboleth?', follow=True)

        self.assertContains(r, 'already in use')

        cl_olduser.set_tokens(eppn="oldusereppn")

        r = cl_olduser.get('/im/login/shibboleth?', follow=True)

        self.assertContains(r, 'Academic login enabled for this account')

        user = User.objects.get(email="olduser@synnefo.org")

        shib_provider = user.get_auth_provider('shibboleth', 'oldusereppn')

        local_provider = user.get_auth_provider('local')

        self.assertEqual(shib_provider.get_remove_policy, True)

        self.assertEqual(local_provider.get_remove_policy, True)

        policy_only_academic = Profile.objects.add_policy('academic_strict2',

                                                          'shibboleth',

                                                          academic_group,

                                                          remove=False)

        user.groups.add(academic_group)

        shib_provider = user.get_auth_provider('shibboleth', 'oldusereppn')

        local_provider = user.get_auth_provider('local')

        self.assertEqual(shib_provider.get_remove_policy, False)

        self.assertEqual(local_provider.get_remove_policy, True)

        self.assertEqual(local_provider.get_login_policy, False)

        cl_olduser.logout()

        login_data = {'username': 'olduser@synnefo.org',

                      'password': 'password'}

        r = cl_olduser.post('/im/local', login_data, follow=True)

        self.assertContains(r, "href='/im/login/shibboleth'>Academic login")

        Group.objects.all().delete()

class TestAuthProvidersAPI(TestCase):

    """

    Test auth_providers module API

    """

    def tearDown(self):

        Group.objects.all().delete()

    @im_settings(IM_MODULES=['local', 'shibboleth'])

    def test_create(self):

        user = AstakosUser.objects.create(email="kpap@synnefo.org")

        user2 = AstakosUser.objects.create(email="kpap2@synnefo.org")

        module = 'shibboleth'

        identifier = 'SHIB_UUID'

        provider_params = {

            'affiliation': 'UNIVERSITY',

            'info': {'age': 27}

        }

        provider = auth.get_provider(module, user2, identifier,

                                     **provider_params)

        provider.add_to_user()

        provider = auth.get_provider(module, user, identifier,

                                     **provider_params)

        provider.add_to_user()

        user.email_verified = True

        user.save()

        self.assertRaises(Exception, provider.add_to_user)

        provider = user.get_auth_provider(module, identifier)

        self.assertEqual(user.get_auth_provider(

            module, identifier)._instance.info.get('age'), 27)

        module = 'local'

        identifier = None

        provider_params = {'auth_backend': 'ldap', 'info':

                          {'office': 'A1'}}

        provider = auth.get_provider(module, user, identifier,

                                     **provider_params)

        provider.add_to_user()

        self.assertFalse(provider.get_add_policy)

        self.assertRaises(Exception, provider.add_to_user)

        shib = user.get_auth_provider('shibboleth',

                                      'SHIB_UUID')

        self.assertTrue(shib.get_remove_policy)

        local = user.get_auth_provider('local')

        self.assertTrue(local.get_remove_policy)

        local.remove_from_user()

        self.assertFalse(shib.get_remove_policy)

        self.assertRaises(Exception, shib.remove_from_user)

        provider = user.get_auth_providers()[0]

        self.assertRaises(Exception, provider.add_to_user)

    @im_settings(IM_MODULES=['local', 'shibboleth'])

    @shibboleth_settings(ADD_GROUPS_POLICY=['group1', 'group2'],

                         CREATION_GROUPS_POLICY=['group-create', 'group1',

                                                 'group2'])

    @localauth_settings(ADD_GROUPS_POLICY=['localgroup'],

                        CREATION_GROUPS_POLICY=['localgroup-create',

                                                'group-create'])

    def test_add_groups(self):

        user = AstakosUser.objects.create(email="kpap@synnefo.org")

        provider = auth.get_provider('shibboleth', user, 'test123')

        provider.add_to_user()

        user = AstakosUser.objects.get()

        self.assertEqual(sorted(user.groups.values_list('name', flat=True)),

                              sorted([u'group1', u'group2', u'group-create']))

        local = auth.get_provider('local', user)

        local.add_to_user()

        provider = user.get_auth_provider('shibboleth')

        self.assertEqual(provider.get_add_groups_policy, ['group1', 'group2'])

        provider.remove_from_user()

        user = AstakosUser.objects.get()

        self.assertEqual(len(user.get_auth_providers()), 1)

        self.assertEqual(sorted(user.groups.values_list('name', flat=True)),

                              sorted([u'group-create', u'localgroup']))

        local = user.get_auth_provider('local')

        self.assertRaises(Exception, local.remove_from_user)

        provider = auth.get_provider('shibboleth', user, 'test123')

        provider.add_to_user()

        user = AstakosUser.objects.get()

        self.assertEqual(sorted(user.groups.values_list('name', flat=True)),

                              sorted([u'group-create', u'group1', u'group2',

                               u'localgroup']))

        Group.objects.all().delete()

    @im_settings(IM_MODULES=['local', 'shibboleth'])

    def test_policies(self):

        group_old, created = Group.objects.get_or_create(name='olduser')

        astakos_settings.MODERATION_ENABLED = True

        settings.ASTAKOS_AUTH_PROVIDER_SHIBBOLETH_CREATION_GROUPS_POLICY = \

            ['academic-user']

        settings.ASTAKOS_AUTH_PROVIDER_GOOGLE_ADD_GROUPS_POLICY = \

            ['google-user']

        user = AstakosUser.objects.create(email="kpap@synnefo.org")

        user.groups.add(group_old)

        user.add_auth_provider('local')

        user2 = AstakosUser.objects.create(email="kpap2@synnefo.org")

        user2.add_auth_provider('shibboleth', identifier='shibid')

        user3 = AstakosUser.objects.create(email="kpap3@synnefo.org")

        user3.groups.add(group_old)