# List of modules to import when celery starts.

import sys

import os

import ConfigParser

root = os.getcwd()

sys.path.append(root)

config = ConfigParser.ConfigParser({'here':root})

config.read('%s/%s' % (root, PYLONS_CONFIG_NAME))

PYLONS_CONFIG = config

CELERY_IMPORTS = ("pylons_app.lib.celerylib.tasks",)

## Result store settings.

CELERY_RESULT_BACKEND = "database"

CELERY_RESULT_DBURI = dict(config.items('app:main'))['sqlalchemy.db1.url']

CELERY_RESULT_SERIALIZER = 'json'

BROKER_CONNECTION_MAX_RETRIES = 30

## Broker settings.

BROKER_HOST = "localhost"

BROKER_PORT = 5672

BROKER_VHOST = "rabbitmqhost"

BROKER_USER = "rabbitmq"

BROKER_PASSWORD = "qweqwe"

## Worker settings

## If you're doing mostly I/O you can have more processes,

## but if mostly spending CPU, try to keep it close to the

## number of CPUs on your machine. If not set, the number of CPUs/cores

## available will be used.

CELERYD_CONCURRENCY = 2

# CELERYD_LOG_FILE = "celeryd.log"

CELERYD_LOG_LEVEL = "DEBUG"

CELERYD_MAX_TASKS_PER_CHILD = 3

#Tasks will never be sent to the queue, but executed locally instead.

CELERY_ALWAYS_EAGER = False

#===============================================================================

# EMAIL SETTINGS

#===============================================================================

pylons_email_config = dict(config.items('DEFAULT'))

CELERY_SEND_TASK_ERROR_EMAILS = True

#List of (name, email_address) tuples for the admins that should receive error e-mails.

ADMINS = [('Administrator', pylons_email_config.get('email_to'))]

#The e-mail address this worker sends e-mails from. Default is "celery@localhost".

SERVER_EMAIL = pylons_email_config.get('error_email_from')

#The mail server to use. Default is "localhost".

MAIL_HOST = pylons_email_config.get('smtp_server')

#Username (if required) to log on to the mail server with.

MAIL_HOST_USER = pylons_email_config.get('smtp_username')

#Password (if required) to log on to the mail server with.

MAIL_HOST_PASSWORD = pylons_email_config.get('smtp_password')

MAIL_PORT = pylons_email_config.get('smtp_port')

        self.check_for_db(override)

        if override:

            log.info("database exisist and it's going to be destroyed")

            if self.tests:

                destroy = True

            else:

                destroy = ask_ok('Are you sure to destroy old database ? [y/n]')

            if not destroy:

                sys.exit()

            if self.db_exists and destroy:

                os.remove(jn(ROOT, self.dbname))

        checkfirst = not override

        meta.Base.metadata.create_all(checkfirst=checkfirst)

        log.info('Created tables for %s', self.dbname)

    def admin_prompt(self):

        if not self.tests:

            import getpass

            username = raw_input('Specify admin username:')

            password = getpass.getpass('Specify admin password:')

            email = raw_input('Specify admin email:')

            self.create_user(username, password, email, True)

        else:

            log.info('creating admin and regular test users')

    def config_prompt(self, test_repo_path=''):

        log.info('Setting up repositories config')

        if not self.tests and not test_repo_path:

            path = raw_input('Specify valid full path to your repositories'

                        ' you can change this later in application settings:')

        else:

            path = test_repo_path

        if not os.path.isdir(path):

            log.error('You entered wrong path: %s', path)

            sys.exit()

        hooks1 = HgAppUi()

        hooks1.ui_section = 'hooks'

        hooks1.ui_key = 'changegroup.update'

        hooks1.ui_value = 'hg update >&2'

        hooks2 = HgAppUi()

        hooks2.ui_section = 'hooks'

        hooks2.ui_key = 'changegroup.repo_size'

log = logging.getLogger(__name__) 

import pylons.test

__all__ = ['environ', 'url', 'TestController']

# Invoke websetup with the current config file

#SetupCommand('setup-app').run([config_file])

environ = {}

class TestController(TestCase):

    def __init__(self, *args, **kwargs):

        wsgiapp = pylons.test.pylonsapp

        config = wsgiapp.config

        self.app = TestApp(wsgiapp)

        url._push_object(URLGenerator(config['routes.map'], environ))

        self.sa = meta.Session

        TestCase.__init__(self, *args, **kwargs)

        response = self.app.post(url(controller='login', action='index'),

                                 {'username':username,

                                  'password':password})

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        return response.follow()

        response = self.app.post(url('admin_settings_my_account_update'), params=dict(

                                                            _method='put',

                                                            username='test_admin',

                                                            new_password='test',

                                                            password='',

                                                            name='NewName',

                                                            lastname='NewLastname',

                                                            email=new_email,))

        response.follow()

        print response

        print 'x' * 100

        print response.session

        assert 'Your account was updated succesfully' in response.session['flash'][0][1], 'no flash message about success of change'

        user = self.sa.query(User).filter(User.username == 'test_admin').one()

        assert user.email == new_email , 'incorrect user email after update got %s vs %s' % (user.email, new_email)

    def test_my_account_update_own_email_ok(self):

        self.log_user()

        new_email = 'new@mail.pl'

        response = self.app.post(url('admin_settings_my_account_update'), params=dict(

                                                            _method='put',

                                                            username='test_admin',

                                                            name='NewName',

                                                            lastname='NewLastname',

                                                            email=new_email,))

        print response

    def test_my_account_update_err_email_exists(self):

        self.log_user()

        new_email = 'test_regular@mail.com'#already exisitn email

        response = self.app.post(url('admin_settings_my_account_update'), params=dict(

                                                            _method='put',

                                                            username='test_admin',

                                                            name='NewName',

                                                            lastname='NewLastname',

                                                            email=new_email,))

        print response

        assert 'That e-mail address is already taken' in response.body, 'Missing error message about existing email'

    def test_my_account_update_err(self):

        new_email = 'newmail.pl'

        response = self.app.post(url('admin_settings_my_account_update'), params=dict(

                                                            _method='put',

                                                            name='NewName',

                                                            lastname='NewLastname',

                                                            email=new_email,))

        print response

        assert 'An email address must contain a single @' in response.body, 'Missing error message about wrong email'

        assert 'This username already exists' in response.body, 'Missing error message about existing user'

from pylons_app.model.db import User

from pylons_app.lib.auth import check_password

class TestLoginController(TestController):

    def test_index(self):

        response = self.app.get(url(controller='login', action='index'))

        assert response.status == '200 OK', 'Wrong response from login page got %s' % response.status

        # Test response...

    def test_login_admin_ok(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username':'test_admin',

                                  'password':'test'})

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['hg_app_user'].username == 'test_admin', 'wrong logged in user'

        response = response.follow()

        assert 'auto description for vcs_test' in response.body

    def test_login_regular_ok(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username':'test_regular',

                                  'password':'test'})

        assert response.status == '302 Found', 'Wrong response code from login got %s' % response.status

        assert response.session['hg_app_user'].username == 'test_regular', 'wrong logged in user'

        response = response.follow()

        assert 'auto description for vcs_test' in response.body

        assert '<a title="Admin" href="/_admin">' not in response.body

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

        response = self.app.post(url(controller='login', action='index', came_from=test_came_from),

                                 {'username':'test_admin',

                                  'password':'test'})

        assert response.status == '302 Found', 'Wrong response code from came from redirection'

        response = response.follow()

        assert response.status == '200 OK', 'Wrong response from login page got %s' % response.status

        assert 'Users administration' in response.body, 'No proper title in response'

        response = self.app.post(url(controller='login', action='index'),

                                 {'username':'error',

                                  'password':'test'})

        assert response.status == '200 OK', 'Wrong response from login page'

        assert 'invalid user name' in response.body, 'No error username message in response'

        assert 'invalid password' in response.body, 'No error password message in response'

    def test_register(self):

        response = self.app.get(url(controller='login', action='register'))

        assert 'Sign Up to hg-app' in response.body, 'wrong page for user registration'

    def test_register_err_same_username(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username':'test_admin',

                                             'password':'test',

                                             'email':'goodmail@domain.com',

                                             'name':'test',

                                             'lastname':'test'})

        assert response.status == '200 OK', 'Wrong response from register page got %s' % response.status

        assert 'This username already exists' in response.body 

    def test_register_err_wrong_data(self):

        response = self.app.post(url(controller='login', action='register'),

                                            {'username':'xs',

                                             'password':'',

                                             'email':'goodmailm',

                                             'name':'test',

                                             'lastname':'test'})

        assert response.status == '200 OK', 'Wrong response from register page got %s' % response.status

        assert 'An email address must contain a single @' in response.body

    def test_register_ok(self):

        username = 'test_regular4'

        password = 'qweqwe'

        email = 'marcin@test.com'

        name = 'testname'

        lastname = 'testlastname'

        response = self.app.post(url(controller='login', action='register'),

                                            {'username':username,

                                             'password':password,

                                             'email':email,

                                             'name':name,

                                             'lastname':lastname})

        print response.body

        assert response.status == '302 Found', 'Wrong response from register page got %s' % response.status        

        assert 'You have successfully registered into hg-app' in response.session['flash'][0], 'No flash message about user registration'

        ret = self.sa.query(User).filter(User.username == 'test_regular4').one()

        assert ret.username == username , 'field mismatch %s %s' % (ret.username, username)

        assert check_password(password, ret.password) == True , 'password mismatch'

        assert ret.email == email , 'field mismatch %s %s' % (ret.email, email)

from pylons_app.tests import *

class TestSettingsController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='settings', action='index',

                                    repo_name='vcs_test'))

        # Test response...