# Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you under the Apache License, Version 2.0 (the
#       "License"); you may not use this file except in compliance
#       with the License.  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#       Unless required by applicable law or agreed to in writing,
#       software distributed under the License is distributed on an
#       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#       KIND, either express or implied.  See the License for the
#       specific language governing permissions and limitations
#       under the License.

import calendar
from datetime import datetime, time, timedelta
import json
from urlparse import urlparse, parse_qs
from urllib import urlencode
from bson import ObjectId

import re
from ming.orm.ormsession import ThreadLocalORMSession, session
from tg import config, expose
from mock import patch
import mock
from nose.tools import (
    assert_equal,
    assert_not_equal,
    assert_is_none,
    assert_is_not_none,
    assert_in,
    assert_not_in,
    assert_true,
    assert_false,
)
from pylons import tmpl_context as c
from webob import exc

from allura.tests import TestController
from allura.tests import decorators as td
from alluratest.controller import setup_trove_categories
from allura import model as M
from allura.lib import plugin
from allura.lib import helpers as h


def unentity(s):
    return s.replace('&quot;', '"')


class TestAuth(TestController):
    def test_login(self):
        self.app.get('/auth/')
        r = self.app.post('/auth/send_verification_link', params=dict(a='test@example.com'))
        email = M.User.query.get(username='test-admin').email_addresses[0]
        r = self.app.post('/auth/send_verification_link', params=dict(a=email))
        ThreadLocalORMSession.flush_all()
        r = self.app.get('/auth/verify_addr', params=dict(a='foo'))
        assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r)
        ea = M.EmailAddress.query.find().first()
        r = self.app.get('/auth/verify_addr', params=dict(a=ea.nonce))
        assert json.loads(self.webflash(r))['status'] == 'ok', self.webflash(r)
        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo'))
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='food'))
        assert 'Invalid login' in str(r), r.showbrowser()
        r = self.app.post('/auth/do_login', params=dict(
            username='test-usera', password='foo'))
        assert 'Invalid login' in str(r), r.showbrowser()

    def test_track_login(self):
        user = M.User.by_username('test-user')
        assert_equal(user.last_access['login_date'], None)
        assert_equal(user.last_access['login_ip'], None)
        assert_equal(user.last_access['login_ua'], None)

        self.app.post('/auth/do_login',
                      headers={'User-Agent': 'browser'},
                      extra_environ={'REMOTE_ADDR': 'addr'},
                      params=dict(
                          username='test-user',
                          password='foo'
                      ))
        user = M.User.by_username('test-user')
        assert_not_equal(user.last_access['login_date'], None)
        assert_equal(user.last_access['login_ip'], 'addr')
        assert_equal(user.last_access['login_ua'], 'browser')

    def test_rememberme(self):
        username = M.User.query.get(username='test-user').username

        # Login as test-user with remember me checkbox off
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo'))
        assert_equal(r.session['username'], username)
        assert_equal(r.session['login_expires'], True)

        for header, contents in r.headerlist:
            if header == 'Set-cookie':
                assert_not_in('expires', contents)

        # Login as test-user with remember me checkbox on
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo', rememberme='on'))
        assert_equal(r.session['username'], username)
        assert_not_equal(r.session['login_expires'], True)

        for header, contents in r.headerlist:
            if header == 'Set-cookie':
                assert_in('expires', contents)

    @td.with_user_project('test-admin')
    def test_user_can_not_claim_duplicate_emails(self):
        email_address = 'test_abcd_123@domain.net'
        user = M.User.query.get(username='test-admin')
        addresses_number = len(user.email_addresses)
        self.app.post('/auth/preferences/update_emails',
                      params={
                          'new_addr.addr': email_address,
                          'new_addr.claim': 'Claim Address',
                          'primary_addr': 'test-admin@users.localhost',
                          'preferences.email_format': 'plain',
                          'password': 'foo',
                      },
                      extra_environ=dict(username='test-admin'))

        assert M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-admin@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                          },
                          extra_environ=dict(username='test-admin'))

        assert json.loads(self.webflash(r))['status'] == 'error', self.webflash(r)
        assert M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user._id)).count() == 1
        assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_added_claimed_address_by_other_user_confirmed(self, gen_message_id, sendsimplemail):
        email_address = 'test_abcd_123@domain.net'

        # test-user claimed & confirmed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.query.find(dict(email=email_address)).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()

        # Claiming the same email address by test-admin
        # the email should be added to the email_addresses list but notifications should not be sent

        admin = M.User.query.get(username='test-admin')
        addresses_number = len(admin.email_addresses)
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-admin@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                          },
                          extra_environ=dict(username='test-admin'))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'A verification email has been sent.  ' \
                                                          'Please check your email and click to confirm.'

        args, kwargs = sendsimplemail.post.call_args

        assert sendsimplemail.post.call_count == 1
        assert kwargs['toaddr'] == email_address
        assert kwargs['subject'] == u'%s - Email address claim attempt' % config['site_name']
        assert "You tried to add %s to your Allura account, " \
               "but it is already claimed by your %s account." % (email_address, user.username) in kwargs['text']

        assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1
        assert len(M.EmailAddress.query.find(dict(email=email_address)).all()) == 2

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_added_claimed_address_by_other_user_not_confirmed(self, gen_message_id, sendsimplemail):
        email_address = 'test_abcd_1235@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.query.find(dict(email=email_address)).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()
        # Claiming the same email address by test-admin
        # the email should be added to the email_addresses list but notifications should not be sent

        user1 = M.User.query.get(username='test-user-1')
        addresses_number = len(user1.email_addresses)
        r = self.app.post('/auth/preferences/update_emails',
                          params={
                              'new_addr.addr': email_address,
                              'new_addr.claim': 'Claim Address',
                              'primary_addr': 'test-user-1@users.localhost',
                              'preferences.email_format': 'plain',
                              'password': 'foo',
                          },
                          extra_environ=dict(username='test-user-1'))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'A verification email has been sent.  ' \
                                                          'Please check your email and click to confirm.'
        assert sendsimplemail.post.called
        assert len(M.User.query.get(username='test-admin').email_addresses) == addresses_number + 1
        assert len(M.EmailAddress.query.find(dict(email=email_address)).all()) == 2

    @td.with_user_project('test-admin')
    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_cannot_claim_more_than_max_limit(self, gen_message_id, sendsimplemail):
        with h.push_config(config, **{'user_prefs.maximum_claimed_emails': '1'}):
            r = self.app.post('/auth/preferences/update_emails',
                              params={
                                  'new_addr.addr': 'test_abcd_1@domain.net',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-user-1@users.localhost',
                                  'preferences.email_format': 'plain',
                                  'password': 'foo',
                              },
                              extra_environ=dict(username='test-user-1'))
            assert json.loads(self.webflash(r))['status'] == 'ok'

            r = self.app.post('/auth/preferences/update_emails',
                              params={
                                  'new_addr.addr': 'test_abcd_2@domain.net',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-user-1@users.localhost',
                                  'preferences.email_format': 'plain',
                                  'password': 'foo',
                              },
                              extra_environ=dict(username='test-user-1'))

            assert json.loads(self.webflash(r))['status'] == 'error'
            assert json.loads(self.webflash(r))['message'] == 'You cannot claim more than 1 email addresses.'

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_verification_link_for_confirmed_email(self, gen_message_id, sendsimplemail):
        email_address = 'test_abcd@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        email.confirmed = True

        user1 = M.User.query.get(username='test-user-1')
        user1.claim_address(email_address)
        email = M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user1._id)).first()
        email.confirmed = False

        ThreadLocalORMSession.flush_all()

        r = self.app.post('/auth/send_verification_link',
                          params=dict(a=email_address),
                          extra_environ=dict(username='test-user-1'))

        assert json.loads(self.webflash(r))['status'] == 'ok'
        assert json.loads(self.webflash(r))['message'] == 'Verification link sent'

        args, kwargs = sendsimplemail.post.call_args
        assert sendsimplemail.post.call_count == 1
        assert kwargs['toaddr'] == email_address
        assert kwargs['subject'] == u'%s - Email address claim attempt' % config['site_name']
        assert "You tried to add %s to your Allura account, " \
               "but it is already claimed by your %s account." % (email_address, user.username) in kwargs['text']

    def test_invalidate_verification_link_if_email_was_confirmed(self):
        email_address = 'test_abcd@domain.net'

        # test-user claimed email address
        user = M.User.query.get(username='test-user')
        user.claim_address(email_address)
        email = M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()

        self.app.post('/auth/send_verification_link',
                          params=dict(a=email_address),
                          extra_environ=dict(username='test-user'))


        user1 = M.User.query.get(username='test-user-1')
        user1.claim_address(email_address)
        email1 = M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user1._id)).first()
        email1.confirmed = True
        ThreadLocalORMSession.flush_all()
        # Verify first email with the verification link
        r = self.app.get('/auth/verify_addr', params=dict(a=email.nonce), extra_environ=dict(username='test-user'))

        assert json.loads(self.webflash(r))['status'] == 'error'
        email = M.EmailAddress.query.find(dict(email=email_address, claimed_by_user_id=user._id)).first()
        assert not email.confirmed


    @td.with_user_project('test-admin')
    def test_prefs(self):
        r = self.app.get('/auth/preferences/',
                         extra_environ=dict(username='test-admin'))
        # check preconditions of test data
        assert 'test@example.com' not in r
        assert 'test-admin@users.localhost' in r
        assert_equal(M.User.query.get(username='test-admin').get_pref('email_address'),
                     'test-admin@users.localhost')

        # add test@example
        with td.audits('New email address: test@example.com', user=True):
            r = self.app.post('/auth/preferences/update_emails',
                              extra_environ=dict(username='test-admin'),
                              params={
                                  'new_addr.addr': 'test@example.com',
                                  'new_addr.claim': 'Claim Address',
                                  'primary_addr': 'test-admin@users.localhost',
                                  'password': 'foo',
                                  'preferences.email_format': 'plain'
                              })
        r = self.app.get('/auth/preferences/')
        assert 'test@example.com' in r
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('email_address'), 'test-admin@users.localhost')

        # remove test-admin@users.localhost
        with td.audits('Email address deleted: test-admin@users.localhost', user=True):
            r = self.app.post('/auth/preferences/update_emails',
                              extra_environ=dict(username='test-admin'),
                              params={
                                  'addr-1.ord': '1',
                                  'addr-1.delete': 'on',
                                  'addr-2.ord': '2',
                                  'new_addr.addr': '',
                                  'primary_addr': 'test-admin@users.localhost',
                                  'password': 'foo',
                                  'preferences.email_format': 'plain'
                              })
        r = self.app.get('/auth/preferences/')
        assert 'test-admin@users.localhost' not in r
        # preferred address has not changed if email is not verified
        user = M.User.query.get(username='test-admin')
        assert_equal(user.get_pref('email_address'), None)

        with td.audits('Display Name changed Test Admin => Admin', user=True):
            r = self.app.post('/auth/preferences/update',
                              params={'preferences.display_name': 'Admin'},
                              extra_environ=dict(username='test-admin'))

    @td.with_user_project('test-admin')
    def test_email_prefs_change_requires_password(self):
        # Claim new email
        new_email_params = {
            'new_addr.addr': 'test@example.com',
            'new_addr.claim': 'Claim Address',
            'primary_addr': 'test-admin@users.localhost',
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to claim new email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())
        new_email_params['password'] = 'bad pass'

        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to claim new email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())
        new_email_params['password'] = 'foo'  # valid password

        r = self.app.post('/auth/preferences/update_emails',
                          params=new_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to claim new email', self.webflash(r))
        assert_in('test@example.com', r.follow())

        # Change primary address
        change_primary_params = {
            'new_addr.addr': '',
            'primary_addr': 'test@example.com',
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test-admin@users.localhost')
        change_primary_params['password'] = 'bad pass'

        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test-admin@users.localhost')
        change_primary_params['password'] = 'foo'  # valid password

        r = self.app.post('/auth/preferences/update_emails',
                          params=change_primary_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to change primary address', self.webflash(r))
        assert_equal(M.User.by_username('test-admin').get_pref('email_address'), 'test@example.com')

        # Remove email
        remove_email_params = {
            'addr-1.ord': '1',
            'addr-2.ord': '2',
            'addr-2.delete': 'on',
            'new_addr.addr': '',
            'primary_addr': 'test-admin@users.localhost',
        }
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to delete an email', self.webflash(r))
        assert_in('test@example.com', r.follow())
        remove_email_params['password'] = 'bad pass'
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_in('You must provide your current password to delete an email', self.webflash(r))
        assert_in('test@example.com', r.follow())
        remove_email_params['password'] = 'foo'  # vallid password
        r = self.app.post('/auth/preferences/update_emails',
                          params=remove_email_params,
                          extra_environ=dict(username='test-admin'))
        assert_not_in('You must provide your current password to delete an email', self.webflash(r))
        assert_not_in('test@example.com', r.follow())

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions(self):
        r = self.app.get('/auth/subscriptions/',
                         extra_environ=dict(username='test-admin'))
        subscriptions = M.Mailbox.query.find(dict(
            user_id=c.user._id, is_flash=False)).all()
        # make sure page actually lists all the user's subscriptions
        assert len(
            subscriptions) > 0, 'Test user has no subscriptions, cannot verify that they are shown'
        for m in subscriptions:
            assert m._id in r, "Page doesn't list subscription for Mailbox._id = %s" % m._id

        # make sure page lists all tools which user can subscribe
        user = M.User.query.get(username='test-admin')
        tools = []
        for p in user.my_projects():
            for ac in p.app_configs:
                if not M.Mailbox.subscribed(project_id=p._id, app_config_id=ac._id):
                    tools.append(ac._id)
        for tool_id in tools:
            assert tool_id in r, "Page doesn't list tool with app_config_id = %s" % tool_id

    def _find_subscriptions_form(self, r):
        form = None
        for f in r.forms.itervalues():
            if f.action == 'update_subscriptions':
                form = f
                break
        assert form is not None, "Can't find subscriptions form"
        return form

    def _find_subscriptions_field(self, form, subscribed=False):
        field_name = None
        for k, v in form.fields.iteritems():
            if subscribed:
                check = c and v[0].value == 'on'
            else:
                check = c and v[0].value != 'on'
            if k and k.endswith('.subscribed') and check:
                field_name = k.replace('.subscribed', '')
        assert field_name, "Can't find unsubscribed tool for user"
        return field_name

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions_subscribe(self):
        resp = self.app.get('/auth/subscriptions/',
                            extra_environ=dict(username='test-admin'))
        form = self._find_subscriptions_form(resp)
        # find not subscribed tool, subscribe and verify
        field_name = self._find_subscriptions_field(form, subscribed=False)
        t_id = ObjectId(form.fields[field_name + '.tool_id'][0].value)
        p_id = ObjectId(form.fields[field_name + '.project_id'][0].value)
        subscribed = M.Mailbox.subscribed(project_id=p_id, app_config_id=t_id)
        assert not subscribed, "User already subscribed for tool %s" % t_id
        form.fields[field_name + '.subscribed'][0].value = 'on'
        form.submit()
        subscribed = M.Mailbox.subscribed(project_id=p_id, app_config_id=t_id)
        assert subscribed, "User is not subscribed for tool %s" % t_id

    @td.with_user_project('test-admin')
    def test_prefs_subscriptions_unsubscribe(self):
        resp = self.app.get('/auth/subscriptions/',
                            extra_environ=dict(username='test-admin'))
        form = self._find_subscriptions_form(resp)
        field_name = self._find_subscriptions_field(form, subscribed=True)
        s_id = ObjectId(form.fields[field_name + '.subscription_id'][0].value)
        s = M.Mailbox.query.get(_id=s_id)
        assert s, "User has not subscription with Mailbox._id = %s" % s_id
        form.fields[field_name + '.subscribed'][0].value = None
        form.submit()
        s = M.Mailbox.query.get(_id=s_id)
        assert not s, "User still has subscription with Mailbox._id %s" % s_id

    def test_format_email(self):
        self.app.post('/auth/subscriptions/update_subscriptions',
                      params={'email_format': 'html', 'subscriptions': ''})
        r = self.app.get('/auth/subscriptions/')
        assert '<option selected value="html">HTML</option>' in r
        self.app.post('/auth/subscriptions/update_subscriptions',
                      params={'email_format': 'plain', 'subscriptions': ''})
        r = self.app.get('/auth/subscriptions/')
        assert '<option selected value="plain">Plain Text</option>' in r
        self.app.post('/auth/subscriptions/update_subscriptions',
                      params={'email_format': 'both', 'subscriptions': ''})
        r = self.app.get('/auth/subscriptions/')
        assert '<option selected value="both">Combined</option>' in r

    def test_create_account(self):
        r = self.app.get('/auth/create_account')
        assert 'Create an Account' in r
        r = self.app.post('/auth/save_new',
                          params=dict(username='aaa', pw='123'))
        assert 'Enter a value 6 characters long or more' in r
        r = self.app.post(
            '/auth/save_new',
            params=dict(
                username='aaa',
                pw='12345678',
                pw2='12345678',
                display_name='Test Me'))
        r = r.follow()
        assert 'User "aaa" registered' in unentity(r.body)
        r = self.app.post(
            '/auth/save_new',
            params=dict(
                username='aaa',
                pw='12345678',
                pw2='12345678',
                display_name='Test Me'))
        assert 'That username is already taken. Please choose another.' in r
        r = self.app.get('/auth/logout')
        r = self.app.post(
            '/auth/do_login',
            params=dict(username='aaa', password='12345678'),
            status=302)

    def test_create_account_require_email(self):
        with h.push_config(config, **{'auth.require_email_addr': 'false'}):
            self.app.post(
                '/auth/save_new',
                params=dict(
                    username='aaa',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com'))
            user = M.User.query.get(username='aaa')
            assert not user.pending
        with h.push_config(config, **{'auth.require_email_addr': 'true'}):
            self.app.post(
                '/auth/save_new',
                params=dict(
                    username='bbb',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com'))
            user = M.User.query.get(username='bbb')
            assert user.pending

    def test_verify_email(self):
        with h.push_config(config, **{'auth.require_email_addr': 'true'}):
            r = self.app.post(
                '/auth/save_new',
                params=dict(
                    username='aaa',
                    pw='12345678',
                    pw2='12345678',
                    display_name='Test Me',
                    email='test@example.com'))
            r = r.follow()
            user = M.User.query.get(username='aaa')
            em = M.EmailAddress.query.get(email='test@example.com')
            assert user._id == em.claimed_by_user_id
            r = self.app.get('/auth/verify_addr', params=dict(a=em.nonce))
            user = M.User.query.get(username='aaa')
            em = M.EmailAddress.query.get(email='test@example.com')
            assert not user.pending
            assert em.confirmed

    def test_create_account_disabled_header_link(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            r = self.app.get('/')
            assert 'Register' not in r

    def test_create_account_disabled_form_gone(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            r = self.app.get('/auth/create_account', status=404)
            assert 'Create an Account' not in r

    def test_create_account_disabled_submit_fails(self):
        with h.push_config(config, **{'auth.allow_user_registration': 'false'}):
            self.app.post('/auth/save_new',
                          params=dict(
                              username='aaa',
                              pw='12345678',
                              pw2='12345678',
                              display_name='Test Me'),
                          status=404)

    def test_one_project_role(self):
        """Make sure when a user goes to a new project only one project role is created.
           There was an issue with extra project roles getting created if a user went directly to
           an admin page."""
        p_nbhd = M.Neighborhood.query.get(name='Projects')
        p = M.Project.query.get(shortname='test', neighborhood_id=p_nbhd._id)
        self.app.post('/auth/save_new', params=dict(
            username='aaa',
            pw='12345678',
            pw2='12345678',
            display_name='Test Me',
            email='test@example.com')).follow()
        user = M.User.query.get(username='aaa')
        user.pending = False
        session(user).flush(user)
        assert M.ProjectRole.query.find(
            dict(user_id=user._id, project_id=p._id)).count() == 0

        self.app.get('/p/test/admin/permissions',
                     extra_environ=dict(username='aaa'), status=403)
        assert M.ProjectRole.query.find(
            dict(user_id=user._id, project_id=p._id)).count() <= 1

    def test_default_lookup(self):
        # Make sure that default _lookup() throws 404
        self.app.get('/auth/foobar', status=404)

    def test_disabled_user(self):
        user = M.User.query.get(username='test-admin')
        sess = session(user)
        assert not user.disabled
        r = self.app.get('/p/test/admin/',
                         extra_environ={'username': 'test-admin'})
        assert_equal(r.status_int, 200, 'Redirect to %s' % r.location)
        user.disabled = True
        sess.save(user)
        sess.flush()
        user = M.User.query.get(username='test-admin')
        assert user.disabled
        r = self.app.get('/p/test/admin/',
                         extra_environ={'username': 'test-admin'})
        assert_equal(r.status_int, 302)
        assert_equal(r.location, 'http://localhost/auth/?return_to=%2Fp%2Ftest%2Fadmin%2F')

    def test_no_open_return_to(self):
        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='/foo'))
        assert_equal(r.location, 'http://localhost/foo')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='https://localhost/foo'))
        assert_equal(r.location, 'https://localhost/foo')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='http://example.com/foo'))
        assert_equal(r.location, 'http://localhost/')

        r = self.app.get('/auth/logout')
        r = self.app.post('/auth/do_login', params=dict(
            username='test-user', password='foo',
            return_to='//example.com/foo'))
        assert_equal(r.location, 'http://localhost/')


class TestPreferences(TestController):
    @td.with_user_project('test-admin')
    def test_personal_data(self):
        from pytz import country_names

        setsex, setbirthdate, setcountry, setcity, settimezone = \
            ('Male', '19/08/1988', 'IT', 'Milan', 'Europe/Rome')
        self.app.get('/auth/user_info/')

        # Check if personal data is properly set
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(
                              sex=setsex,
                              birthdate=setbirthdate,
                              country=setcountry,
                              city=setcity,
                              timezone=settimezone))
        user = M.User.query.get(username='test-admin')
        sex = user.sex
        assert sex == setsex
        birthdate = user.birthdate.strftime('%d/%m/%Y')
        assert birthdate == setbirthdate
        country = user.localization.country
        assert country_names.get(setcountry) == country
        city = user.localization.city
        assert city == setcity
        timezone = user.timezone
        assert timezone == settimezone

        # Check if setting a wrong date everything works correctly
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(birthdate='30/02/1998'))
        assert 'Please enter a valid date' in str(r)
        user = M.User.query.get(username='test-admin')
        sex = user.sex
        assert sex == setsex
        birthdate = user.birthdate.strftime('%d/%m/%Y')
        assert birthdate == setbirthdate
        country = user.localization.country
        assert country_names.get(setcountry) == country
        city = user.localization.city
        assert city == setcity
        timezone = user.timezone
        assert timezone == settimezone

        # Check deleting birthdate
        r = self.app.post('/auth/user_info/change_personal_data',
                          params=dict(
                              sex=setsex,
                              birthdate='',
                              country=setcountry,
                              city=setcity,
                              timezone=settimezone))
        user = M.User.query.get(username='test-admin')
        assert user.birthdate is None

    @td.with_user_project('test-admin')
    def test_contacts(self):
        # Add skype account
        testvalue = 'testaccount'
        self.app.get('/auth/user_info/contacts/')
        self.app.post('/auth/user_info/contacts/skype_account',
                      params=dict(skypeaccount=testvalue))
        user = M.User.query.get(username='test-admin')
        assert user.skypeaccount == testvalue

        # Add social network account
        socialnetwork = 'Facebook'
        accounturl = 'http://www.facebook.com/test'
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(socialnetwork=socialnetwork,
                                  accounturl=accounturl))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1 \
               and user.socialnetworks[0].socialnetwork == socialnetwork \
               and user.socialnetworks[0].accounturl == accounturl

        # Add second social network account
        socialnetwork2 = 'Twitter'
        accounturl2 = 'http://twitter.com/test'
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(socialnetwork=socialnetwork2,
                                  accounturl='@test'))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 2 and \
               ({'socialnetwork': socialnetwork, 'accounturl': accounturl} in user.socialnetworks and
                {'socialnetwork': socialnetwork2, 'accounturl': accounturl2} in user.socialnetworks)

        # Remove first social network account
        self.app.post('/auth/user_info/contacts/remove_social_network',
                      params=dict(socialnetwork=socialnetwork,
                                  account=accounturl))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1 and \
               {'socialnetwork': socialnetwork2, 'accounturl': accounturl2} in user.socialnetworks

        # Add empty social network account
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(accounturl=accounturl, socialnetwork=''))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1 and \
               {'socialnetwork': socialnetwork2, 'accounturl': accounturl2} in user.socialnetworks

        # Add invalid social network account
        self.app.post('/auth/user_info/contacts/add_social_network',
                      params=dict(accounturl=accounturl, socialnetwork='invalid'))
        user = M.User.query.get(username='test-admin')
        assert len(user.socialnetworks) == 1 and \
               {'socialnetwork': socialnetwork2, 'accounturl': accounturl2} in user.socialnetworks

        # Add telephone number
        telnumber = '+3902123456'
        self.app.post('/auth/user_info/contacts/add_telnumber',
                      params=dict(newnumber=telnumber))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 1 and (user.telnumbers[0] == telnumber))

        # Add second telephone number
        telnumber2 = '+3902654321'
        self.app.post('/auth/user_info/contacts/add_telnumber',
                      params=dict(newnumber=telnumber2))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 2 and telnumber in user.telnumbers and telnumber2 in user.telnumbers)

        # Remove first telephone number
        self.app.post('/auth/user_info/contacts/remove_telnumber',
                      params=dict(oldvalue=telnumber))
        user = M.User.query.get(username='test-admin')
        assert (len(user.telnumbers) == 1 and telnumber2 in user.telnumbers)

        # Add website
        website = 'http://www.testurl.com'
        self.app.post('/auth/user_info/contacts/add_webpage',
                      params=dict(newwebsite=website))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 1 and (website in user.webpages))

        # Add second website
        website2 = 'http://www.testurl2.com'
        self.app.post('/auth/user_info/contacts/add_webpage',
                      params=dict(newwebsite=website2))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 2 and website in user.webpages and website2 in user.webpages)

        # Remove first website
        self.app.post('/auth/user_info/contacts/remove_webpage',
                      params=dict(oldvalue=website))
        user = M.User.query.get(username='test-admin')
        assert (len(user.webpages) == 1 and website2 in user.webpages)

    @td.with_user_project('test-admin')
    def test_availability(self):
        # Add availability timeslot
        weekday = 'Monday'
        starttime = time(9, 0, 0)
        endtime = time(12, 0, 0)

        self.app.get('/auth/user_info/availability/')
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday,
                              starttime=starttime.strftime('%H:%M'),
                              endtime=endtime.strftime('%H:%M')))
        user = M.User.query.get(username='test-admin')
        timeslot1dict = dict(
            week_day=weekday, start_time=starttime, end_time=endtime)
        assert len(
            user.availability) == 1 and timeslot1dict in user.get_availability_timeslots()

        weekday2 = 'Tuesday'
        starttime2 = time(14, 0, 0)
        endtime2 = time(16, 0, 0)

        # Add second availability timeslot
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday2,
                              starttime=starttime2.strftime('%H:%M'),
                              endtime=endtime2.strftime('%H:%M')))
        user = M.User.query.get(username='test-admin')
        timeslot2dict = dict(week_day=weekday2,
                             start_time=starttime2, end_time=endtime2)
        assert len(user.availability) == 2 \
               and timeslot1dict in user.get_availability_timeslots() \
               and timeslot2dict in user.get_availability_timeslots()

        # Remove availability timeslot
        r = self.app.post('/auth/user_info/availability/remove_timeslot',
                          params=dict(
                              weekday=weekday,
                              starttime=starttime.strftime('%H:%M'),
                              endtime=endtime.strftime('%H:%M')))
        user = M.User.query.get(username='test-admin')
        assert len(user.availability) == 1 and timeslot2dict in user.get_availability_timeslots()

        # Add invalid availability timeslot
        r = self.app.post('/auth/user_info/availability/add_timeslot',
                          params=dict(
                              weekday=weekday2,
                              starttime=endtime2.strftime('%H:%M'),
                              endtime=starttime2.strftime('%H:%M')))
        assert 'Invalid period:' in str(r)
        user = M.User.query.get(username='test-admin')
        timeslot2dict = dict(week_day=weekday2,
                             start_time=starttime2, end_time=endtime2)
        assert len(user.availability) == 1 and timeslot2dict in user.get_availability_timeslots()

    @td.with_user_project('test-admin')
    def test_inactivity(self):
        # Add inactivity period
        now = datetime.utcnow().date()
        now = datetime(now.year, now.month, now.day)
        startdate = now + timedelta(days=1)
        enddate = now + timedelta(days=7)
        self.app.get('/auth/user_info/availability/')
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate=startdate.strftime('%d/%m/%Y'),
                              enddate=enddate.strftime('%d/%m/%Y')))
        user = M.User.query.get(username='test-admin')
        period1dict = dict(start_date=startdate, end_date=enddate)
        assert len(user.inactiveperiod) == 1 and period1dict in user.get_inactive_periods()

        # Add second inactivity period
        startdate2 = now + timedelta(days=24)
        enddate2 = now + timedelta(days=28)
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate=startdate2.strftime('%d/%m/%Y'),
                              enddate=enddate2.strftime('%d/%m/%Y')))
        user = M.User.query.get(username='test-admin')
        period2dict = dict(start_date=startdate2, end_date=enddate2)
        assert len(user.inactiveperiod) == 2 \
               and period1dict in user.get_inactive_periods() \
               and period2dict in user.get_inactive_periods()

        # Remove first inactivity period
        r = self.app.post(
            '/auth/user_info/availability/remove_inactive_period',
            params=dict(
                startdate=startdate.strftime('%d/%m/%Y'),
                enddate=enddate.strftime('%d/%m/%Y')))
        user = M.User.query.get(username='test-admin')
        assert len(user.inactiveperiod) == 1 and period2dict in user.get_inactive_periods()

        # Add invalid inactivity period
        r = self.app.post('/auth/user_info/availability/add_inactive_period',
                          params=dict(
                              startdate='NOT/A/DATE',
                              enddate=enddate2.strftime('%d/%m/%Y')))
        user = M.User.query.get(username='test-admin')
        assert 'Please enter a valid date' in str(r)
        assert len(user.inactiveperiod) == 1 and period2dict in user.get_inactive_periods()

    @td.with_user_project('test-admin')
    def test_skills(self):
        setup_trove_categories()
        # Add a skill
        skill_cat = M.TroveCategory.query.get(show_as_skill=True)
        level = 'low'
        comment = 'test comment'
        result = self.app.get('/auth/user_info/skills/')
        r = self.app.post('/auth/user_info/skills/save_skill',
                          params=dict(
                              level=level,
                              comment=comment,
                              selected_skill=str(skill_cat.trove_cat_id)))
        user = M.User.query.get(username='test-admin')
        skilldict = dict(category_id=skill_cat._id,
                         comment=comment, level=level)
        assert len(user.skills) == 1 and skilldict in user.skills

        # Add again the same skill
        level = 'medium'
        comment = 'test comment 2'
        result = self.app.get('/auth/user_info/skills/')
        r = self.app.post('/auth/user_info/skills/save_skill',
                          params=dict(
                              level=level,
                              comment=comment,
                              selected_skill=str(skill_cat.trove_cat_id)))
        user = M.User.query.get(username='test-admin')
        skilldict = dict(category_id=skill_cat._id,
                         comment=comment, level=level)
        assert len(user.skills) == 1 and skilldict in user.skills

        # Add an invalid skill
        level2 = 'not a level'
        comment2 = 'test comment 2'
        r = self.app.post('/auth/user_info/skills/save_skill',
                          params=dict(
                              level=level2,
                              comment=comment2,
                              selected_skill=str(skill_cat.trove_cat_id)))
        user = M.User.query.get(username='test-admin')
        # Check that everything is as it was before
        assert len(user.skills) == 1 and skilldict in user.skills

        # Remove a skill
        self.app.get('/auth/user_info/skills/')
        self.app.post('/auth/user_info/skills/remove_skill',
                      params=dict(
                          categoryid=str(skill_cat.trove_cat_id)))
        user = M.User.query.get(username='test-admin')
        assert len(user.skills) == 0

    @td.with_user_project('test-admin')
    def test_user_message(self):
        assert not M.User.query.get(username='test-admin').get_pref('disable_user_messages')
        self.app.post('/auth/preferences/user_message')
        assert M.User.query.get(username='test-admin').get_pref('disable_user_messages')
        self.app.post('/auth/preferences/user_message',
                      params={'allow_user_messages': 'on'})
        assert not M.User.query.get(username='test-admin').get_pref('disable_user_messages')

    @td.with_user_project('test-admin')
    def test_additional_page(self):
        class MyPP(plugin.UserPreferencesProvider):
            def not_page(self):
                return 'not page'

            @expose()
            def new_page(self):
                return 'new page'

        with mock.patch.object(plugin.UserPreferencesProvider, 'get') as upp_get:
            upp_get.return_value = MyPP()
            r = self.app.get('/auth/new_page')
            assert_equal(r.body, 'new page')
            self.app.get('/auth/not_page', status=404)


class TestPasswordReset(TestController):
    test_primary_email = 'testprimaryaddr@mail.com'

    @patch('allura.tasks.mail_tasks.sendmail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_email_unconfirmed(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.query.find(
            {'claimed_by_user_id': user._id}).first()
        email.confirmed = False
        ThreadLocalORMSession.flush_all()
        self.app.post('/auth/password_recovery_hash', {'email': email.email})
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        assert hash is None

    @patch('allura.tasks.mail_tasks.sendmail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_user_disabled(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.query.find(
            {'claimed_by_user_id': user._id}).first()
        user.disabled = True
        ThreadLocalORMSession.flush_all()
        self.app.post('/auth/password_recovery_hash', {'email': email.email})
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        assert hash is None

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_only_primary_email_reset_allowed(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        user.claim_address(self.test_primary_email)
        user.set_pref('email_address', self.test_primary_email)

        email = M.EmailAddress.query.find({'email': self.test_primary_email}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()

        with h.push_config(config, **{'auth.allow_non_primary_email_password_reset': 'false'}):
            self.app.post('/auth/password_recovery_hash', {'email': self.test_primary_email})
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            assert hash is not None
            args, kwargs = sendmail.post.call_args
            assert_equal(kwargs['toaddr'], self.test_primary_email)

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_non_primary_email_reset_allowed(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email1 = M.EmailAddress.query.find({'claimed_by_user_id': user._id}).first()
        user.claim_address(self.test_primary_email)
        user.set_pref('email_address', self.test_primary_email)
        email = M.EmailAddress.query.find({'email': self.test_primary_email}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        with h.push_config(config, **{'auth.allow_non_primary_email_password_reset': 'true'}):
            self.app.post('/auth/password_recovery_hash', {'email': email1.email})
            hash = user.get_tool_data('AuthPasswordReset', 'hash')
            assert hash is not None
            args, kwargs = sendmail.post.call_args
            assert_equal(kwargs['toaddr'], email1.email)

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_password_reset(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.query.find(
            {'claimed_by_user_id': user._id}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        old_pw_hash = user.password
        with td.audits('Password recovery link sent to: test-admin@users.localhost', user=True):
            r = self.app.post('/auth/password_recovery_hash', {'email': email.email})
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
        assert hash is not None
        assert hash_expiry is not None

        r = self.app.get('/auth/forgotten_password/%s' % hash)
        assert_in('Enter a new password for: test-admin', r)
        assert_in('New Password:', r)
        assert_in('New Password (again):', r)
        form = r.forms[0]
        form['pw'] = form['pw2'] = new_password = '154321'
        with td.audits('Password changed \(through recovery process\)', user=True):
            # escape parentheses, so they would not be treated as regex group
            r = form.submit()
        user = M.User.query.get(username='test-admin')
        assert_not_equal(old_pw_hash, user.password)
        provider = plugin.LocalAuthenticationProvider(None)
        assert_true(provider._validate_password(user, new_password))

        text = '''Your username is test-admin

To reset your password on %s, please visit the following URL:

%s/auth/forgotten_password/%s''' % (config['site_name'], config['base_url'], hash)

        sendmail.post.assert_called_once_with(
            toaddr=email.email,
            fromaddr=config['forgemail.return_path'],
            reply_to=config['forgemail.return_path'],
            subject='Allura Password recovery',
            message_id=gen_message_id(),
            text=text)
        user = M.User.query.get(username='test-admin')
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        hash_expiry = user.get_tool_data('AuthPasswordReset', 'hash_expiry')
        assert_equal(hash, '')
        assert_equal(hash_expiry, '')

    @patch('allura.tasks.mail_tasks.sendsimplemail')
    @patch('allura.lib.helpers.gen_message_id')
    def test_hash_expired(self, gen_message_id, sendmail):
        user = M.User.query.get(username='test-admin')
        email = M.EmailAddress.query.find(
            {'claimed_by_user_id': user._id}).first()
        email.confirmed = True
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/auth/password_recovery_hash', {'email': email.email})
        user = M.User.by_username('test-admin')
        hash = user.get_tool_data('AuthPasswordReset', 'hash')
        user.set_tool_data('AuthPasswordReset',
                           hash_expiry=datetime(2000, 10, 10))
        r = self.app.get('/auth/forgotten_password/%s' % hash.encode('utf-8'))
        assert_in('Unable to process reset, please try again', r.follow().body)
        r = self.app.post('/auth/set_new_password/%s' %
                          hash.encode('utf-8'), {'pw': '154321', 'pw2': '154321'})
        assert_in('Unable to process reset, please try again', r.follow().body)

    @patch('allura.lib.plugin.AuthenticationProvider')
    def test_provider_disabled(self, AP):
        user = M.User.query.get(username='test-admin')
        ap = AP.get()
        ap.forgotten_password_process = False
        ap.authenticate_request()._id = user._id
        ap.by_username().username = user.username
        self.app.get('/auth/forgotten_password', status=404)
        self.app.post('/auth/set_new_password',
                      {'pw': 'foo', 'pw2': 'foo'}, status=404)
        self.app.post('/auth/password_recovery_hash',
                      {'email': 'foo'}, status=404)


class TestOAuth(TestController):
    def test_register_deregister_app(self):
        # register
        r = self.app.get('/auth/oauth/')
        r = self.app.post('/auth/oauth/register',
                          params={'application_name': 'oautstapp', 'application_description': 'Oauth rulez'}).follow()
        assert 'oautstapp' in r
        # deregister
        assert_equal(r.forms[0].action, 'deregister')
        r.forms[0].submit()
        r = self.app.get('/auth/oauth/')
        assert 'oautstapp' not in r

    def test_generate_revoke_access_token(self):
        # generate
        r = self.app.post('/auth/oauth/register',
                          params={'application_name': 'oautstapp', 'application_description': 'Oauth rulez'}).follow()
        assert_equal(r.forms[1].action, 'generate_access_token')
        r.forms[1].submit()
        r = self.app.get('/auth/oauth/')
        assert 'Bearer Token:' in r
        assert_not_equal(
            M.OAuthAccessToken.for_user(M.User.by_username('test-admin')), [])
        # revoke
        assert_equal(r.forms[0].action, 'revoke_access_token')
        r.forms[0].submit()
        r = self.app.get('/auth/oauth/')
        assert_not_equal(r.forms[0].action, 'revoke_access_token')
        assert_equal(
            M.OAuthAccessToken.for_user(M.User.by_username('test-admin')), [])

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_interactive(self, Request, Server):
        M.OAuthConsumerToken.consumer = mock.Mock()
        user = M.User.by_username('test-admin')
        consumer_token = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        ThreadLocalORMSession.flush_all()
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_callback': 'http://my.domain.com/callback',
        }
        r = self.app.post('/rest/oauth/request_token', params={})
        rtok = parse_qs(r.body)['oauth_token'][0]
        r = self.app.post('/rest/oauth/authorize',
                          params={'oauth_token': rtok})
        r = r.forms[0].submit('yes')
        assert r.location.startswith('http://my.domain.com/callback')
        pin = parse_qs(urlparse(r.location).query)['oauth_verifier'][0]
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': rtok,
            'oauth_verifier': pin,
        }
        r = self.app.get('/rest/oauth/access_token')
        atok = parse_qs(r.body)
        assert_equal(len(atok['oauth_token']), 1)
        assert_equal(len(atok['oauth_token_secret']), 1)

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_request_token_valid(self, Request, Server):
        M.OAuthConsumerToken.consumer = mock.Mock()
        user = M.User.by_username('test-user')
        consumer_token = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        req = Request.from_request.return_value = {'oauth_consumer_key': 'api_key'}
        r = self.app.post('/rest/oauth/request_token', params={'key': 'value'})
        Request.from_request.assert_called_once_with(
            'POST', 'http://localhost/rest/oauth/request_token',
            headers={'Host': 'localhost:80', 'Content-Type': 'application/x-www-form-urlencoded; charset="utf-8"'},
            parameters={'key': 'value'},
            query_string='')
        Server().verify_request.assert_called_once_with(req, consumer_token.consumer, None)
        request_token = M.OAuthRequestToken.query.get(consumer_token_id=consumer_token._id)
        assert_is_not_none(request_token)
        assert_equal(r.body, request_token.to_string())

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_request_token_no_consumer_token(self, Request, Server):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key'}
        r = self.app.post('/rest/oauth/request_token',
                          params={'key': 'value'}, status=403)

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_request_token_invalid(self, Request, Server):
        Server().verify_request.side_effect = ValueError
        M.OAuthConsumerToken.consumer = mock.Mock()
        user = M.User.by_username('test-user')
        consumer_token = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        req = Request.from_request.return_value = {'oauth_consumer_key': 'api_key'}
        self.app.post('/rest/oauth/request_token', params={'key': 'value'}, status=403)

    def test_authorize_ok(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/authorize', params={'oauth_token': 'api_key'})
        assert_in('ctok_desc', r.body)
        assert_in('api_key', r.body)

    def test_authorize_invalid(self):
        self.app.post('/rest/oauth/authorize', params={'oauth_token': 'api_key'}, status=403)

    def test_do_authorize_no(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        self.app.post('/rest/oauth/do_authorize',
                      params={'no': '1', 'oauth_token': 'api_key'})
        assert_is_none(M.OAuthRequestToken.query.get(api_key='api_key'))

    def test_do_authorize_oob(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='oob',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key'})
        assert_is_not_none(r.html.find(text=re.compile('^PIN: ')))

    def test_do_authorize_cb(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key'})
        assert r.location.startswith('http://my.domain.com/callback?oauth_token=api_key&oauth_verifier=')

    def test_do_authorize_cb_params(self):
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.post('/rest/oauth/do_authorize', params={'yes': '1', 'oauth_token': 'api_key'})
        assert r.location.startswith('http://my.domain.com/callback?myparam=foo&oauth_token=api_key&oauth_verifier=')

    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_access_token_no_consumer(self, Request):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': 'api_key',
            'oauth_verifier': 'good',
        }
        self.app.get('/rest/oauth/access_token', status=403)

    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_access_token_no_request(self, Request):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': 'api_key',
            'oauth_verifier': 'good',
        }
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        ThreadLocalORMSession.flush_all()
        self.app.get('/rest/oauth/access_token', status=403)

    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_access_token_bad_pin(self, Request):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': 'api_key',
            'oauth_verifier': 'bad',
        }
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good',
        )
        ThreadLocalORMSession.flush_all()
        self.app.get('/rest/oauth/access_token', status=403)

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_access_token_bad_sig(self, Request, Server):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': 'api_key',
            'oauth_verifier': 'good',
        }
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good',
        )
        ThreadLocalORMSession.flush_all()
        Server().verify_request.side_effect = ValueError
        self.app.get('/rest/oauth/access_token', status=403)

    @mock.patch('allura.controllers.rest.oauth.Server')
    @mock.patch('allura.controllers.rest.oauth.Request')
    def test_access_token_ok(self, Request, Server):
        req = Request.from_request.return_value = {
            'oauth_consumer_key': 'api_key',
            'oauth_token': 'api_key',
            'oauth_verifier': 'good',
        }
        user = M.User.by_username('test-admin')
        ctok = M.OAuthConsumerToken(
            api_key='api_key',
            user_id=user._id,
            description='ctok_desc',
        )
        rtok = M.OAuthRequestToken(
            api_key='api_key',
            consumer_token_id=ctok._id,
            callback='http://my.domain.com/callback?myparam=foo',
            user_id=user._id,
            validation_pin='good',
        )
        ThreadLocalORMSession.flush_all()
        r = self.app.get('/rest/oauth/access_token')
        atok = parse_qs(r.body)
        assert_equal(len(atok['oauth_token']), 1)
        assert_equal(len(atok['oauth_token_secret']), 1)


class TestDisableAccount(TestController):
    def test_not_authenticated(self):
        r = self.app.get(
            '/auth/disable/',
            extra_environ={'username': '*anonymous'})
        assert_equal(r.status_int, 302)
        assert_equal(r.location,
                     'http://localhost/auth/?return_to=%2Fauth%2Fdisable%2F')

    def test_lists_user_projects(self):
        r = self.app.get('/auth/disable/')
        user = M.User.by_username('test-admin')
        for p in user.my_projects_by_role_name('Admin'):
            assert_in(p.name, r)
            assert_in(p.url(), r)

    def test_has_asks_password(self):
        r = self.app.get('/auth/disable/')
        form = r.html.find('form', {'action': 'do_disable'})
        assert form is not None

    def test_bad_password(self):
        r = self.app.post('/auth/disable/do_disable', {'password': 'bad'})
        assert_in('Invalid password', r)
        user = M.User.by_username('test-admin')
        assert_equal(user.disabled, False)

    def test_disable(self):
        r = self.app.post('/auth/disable/do_disable', {'password': 'foo'})
        assert_equal(r.status_int, 302)
        assert_equal(r.location, 'http://localhost/')
        flash = json.loads(self.webflash(r))
        assert_equal(flash['status'], 'ok')
        assert_equal(flash['message'], 'Your account was successfully disabled!')
        user = M.User.by_username('test-admin')
        assert_equal(user.disabled, True)


class TestPasswordExpire(TestController):

    def login(self, username='test-user', pwd='foo', query_string=''):
        r = self.app.get('/auth/' + query_string, extra_environ={'username': '*anonymous'})
        f = r.forms[0]
        f['username'] = username
        f['password'] = pwd
        return f.submit(extra_environ={'username': '*anonymous'})

    def assert_redirects(self, where='/'):
        try:
            self.app.get(where, extra_environ={'username': 'test-user'}, status=302)
        except exc.HTTPFound as e:
            assert_equal(e.location, '/auth/pwd_expired?' + urlencode({'return_to': where}))

    def assert_not_redirects(self):
        self.app.get('/', extra_environ={'username': 'test-user'}, status=200)

    def test_disabled(self):
        r = self.login()
        assert_false(r.session.get('pwd-expired'))
        self.assert_not_redirects()

    def expired(self, r):
        return r.session.get('pwd-expired')

    def set_expire_for_user(self, username='test-user', days=100):
        user = M.User.by_username(username)
        user.last_password_updated = datetime.utcnow() - timedelta(days=days)
        session(user).flush(user)
        return user

    def test_days(self):
        self.set_expire_for_user()

        with h.push_config(config, **{'auth.pwdexpire.days': 180}):
            r = self.login()
            assert_false(self.expired(r))
            self.assert_not_redirects()

        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

    def test_before(self):
        self.set_expire_for_user()

        before = datetime.utcnow() - timedelta(days=180)
        before = calendar.timegm(before.timetuple())
        with h.push_config(config, **{'auth.pwdexpire.before': before}):
            r = self.login()
            assert_false(self.expired(r))
            self.assert_not_redirects()

        before = datetime.utcnow() - timedelta(days=90)
        before = calendar.timegm(before.timetuple())
        with h.push_config(config, **{'auth.pwdexpire.before': before}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

    def test_logout(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()
            r = self.app.get('/auth/logout', extra_environ={'username': 'test-user'})
            assert_false(self.expired(r))
            self.assert_not_redirects()

    def test_change_pwd(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

            user = M.User.by_username('test-user')
            old_update_time = user.last_password_updated
            old_password = user.password
            r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
            f = r.forms[0]
            f['oldpw'] = 'foo'
            f['pw'] = 'qwerty'
            f['pw2'] = 'qwerty'
            r = f.submit(extra_environ={'username': 'test-user'}, status=302)
            assert_equal(r.location, 'http://localhost/')
            assert_false(self.expired(r))
            user = M.User.by_username('test-user')
            assert_true(user.last_password_updated > old_update_time)
            assert_not_equal(user.password, old_password)

            # Can log in with new password and change isn't required anymore
            r = self.login(pwd='qwerty')
            assert_equal(r.location, 'http://localhost/')
            assert_not_in('Invalid login', r)
            assert_false(self.expired(r))
            self.assert_not_redirects()

            # and can't log in with old password
            r = self.login(pwd='foo')
            assert_in('Invalid login', r)

    def check_validation(self, oldpw, pw, pw2):
        user = M.User.by_username('test-user')
        old_update_time = user.last_password_updated
        old_password = user.password
        r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
        f = r.forms[0]
        f['oldpw'] = oldpw
        f['pw'] = pw
        f['pw2'] = pw2
        r = f.submit(extra_environ={'username': 'test-user'})
        assert_true(self.expired(r))
        user = M.User.by_username('test-user')
        assert_equal(user.last_password_updated, old_update_time)
        assert_equal(user.password, old_password)
        return r

    def test_change_pwd_validation(self):
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login()
            assert_true(self.expired(r))
            self.assert_redirects()

            r = self.check_validation('', '', '')
            assert_in('Please enter a value', r)
            r = self.check_validation('', 'qwe', 'qwerty')
            assert_in('Enter a value 6 characters long or more', r)
            r = self.check_validation('bad', 'qwerty1', 'qwerty')
            assert_in('Passwords must match', r)
            r = self.check_validation('bad', 'qwerty', 'qwerty')
            assert_in('Incorrect password', self.webflash(r))
            assert_equal(r.location, 'http://localhost/auth/pwd_expired?return_to=')

            with h.push_config(config, **{'auth.min_password_len': 3}):
                r = self.check_validation('foo', 'foo', 'foo')
                assert_in('Your old and new password should not be the same', r)

    def test_return_to(self):
        return_to = '/p/test/tickets/?milestone=1.0&page=2'
        self.set_expire_for_user()
        with h.push_config(config, **{'auth.pwdexpire.days': 90}):
            r = self.login(query_string='?' + urlencode({'return_to': return_to}))
            # don't go to the return_to yet
            assert_equal(r.location, 'http://localhost/auth/pwd_expired?' + urlencode({'return_to': return_to}))

            # but if user tries to go directly there anyway, intercept and redirect back
            self.assert_redirects(where=return_to)

            r = self.app.get('/auth/pwd_expired', extra_environ={'username': 'test-user'})
            f = r.forms[0]
            f['oldpw'] = 'foo'
            f['pw'] = 'qwerty'
            f['pw2'] = 'qwerty'
            f['return_to'] = return_to
            r = f.submit(extra_environ={'username': 'test-user'}, status=302)
            assert_equal(r.location, 'http://localhost/p/test/tickets/?milestone=1.0&page=2')