__version__ += ' [rev:%s]' % _rev[0]

def get_version():

    """Returns shorter version (digit parts only) as string."""

    return '.'.join((str(each) for each in VERSION[:3]))

BACKENDS = {

    'hg': 'Mercurial repository',

    #'git': 'Git repository',

}

"""Pylons environment configuration"""

import os

import logging

from mako.lookup import TemplateLookup

from pylons.configuration import PylonsConfig

from pylons.error import handle_mako_error

import rhodecode.lib.app_globals as app_globals

import rhodecode.lib.helpers

from rhodecode.config.routing import make_map

from rhodecode.lib import engine_from_config

from rhodecode.lib.auth import set_available_permissions

from rhodecode.lib.utils import repo2db_mapper, make_ui, set_rhodecode_config

from rhodecode.model import init_model

from rhodecode.model.scm import ScmModel

log = logging.getLogger(__name__)

def load_environment(global_conf, app_conf, initial=False):

    """Configure the Pylons environment via the ``pylons.config``

    object

    """

    config = PylonsConfig()

    # Pylons paths

    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

    paths = dict(root=root,

                 controllers=os.path.join(root, 'controllers'),

                 static_files=os.path.join(root, 'public'),

                 templates=[os.path.join(root, 'templates')])

    # Initialize config with the basic options

    config.init_app(global_conf, app_conf, package='rhodecode', paths=paths)

    config['routes.map'] = make_map(config)

    config['pylons.app_globals'] = app_globals.Globals(config)

    config['pylons.h'] = rhodecode.lib.helpers

    # Setup cache object as early as possible

    import pylons

    pylons.cache._push_object(config['pylons.app_globals'].cache)

    # Create the Mako TemplateLookup, with the default auto-escaping

    config['pylons.app_globals'].mako_lookup = TemplateLookup(

        directories=paths['templates'],

        error_handler=handle_mako_error,

        module_directory=os.path.join(app_conf['cache_dir'], 'templates'),

        input_encoding='utf-8', default_filters=['escape'],

        imports=['from webhelpers.html import escape'])

    #sets the c attribute access when don't existing attribute are accessed

    config['pylons.strict_tmpl_context'] = True

    test = os.path.split(config['__file__'])[-1] == 'test.ini'

    if test:

        from rhodecode.lib.utils import create_test_env, create_test_index

        from rhodecode.tests import  TESTS_TMP_PATH

        create_test_env(TESTS_TMP_PATH, config)

    #MULTIPLE DB configs

    # Setup the SQLAlchemy database engine

    sa_engine_db1 = engine_from_config(config, 'sqlalchemy.db1.')

    init_model(sa_engine_db1)

    repos_path = make_ui('db').configitems('paths')[0][1]

    repo2db_mapper(ScmModel().repo_scan(repos_path))

    set_available_permissions(config)

    config['base_path'] = repos_path

    set_rhodecode_config(config)

import logging

import traceback

from pylons import tmpl_context as c, url

from rhodecode.lib.base import BaseController, render

from rhodecode.model.db import Notification

from rhodecode.model.notification import NotificationModel

from rhodecode.lib.auth import LoginRequired

from rhodecode.lib import helpers as h

from rhodecode.model.meta import Session

log = logging.getLogger(__name__)

class NotificationsController(BaseController):

    """REST Controller styled on the Atom Publishing Protocol"""

    # To properly map this controller, ensure your config/routing.py

    # file has a resource setup:

    #     map.resource('notification', 'notifications', controller='_admin/notifications', 

    #         path_prefix='/_admin', name_prefix='_admin_')

    @LoginRequired()

    def __before__(self):

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os

import sys

import socket

import traceback

import logging

from os.path import dirname as dn, join as jn

from hashlib import md5

from decorator import decorator

from vcs.utils.lazy import LazyProperty

from rhodecode.lib import str2bool, safe_str

from rhodecode.lib.pidlock import DaemonLock, LockHeld

from celery.messaging import establish_connection

log = logging.getLogger(__name__)

class ResultWrapper(object):

    def __init__(self, task):

        self.task = task

    @LazyProperty

    def result(self):

        return self.task

def run_task(task, *args, **kwargs):

import logging

from os.path import join as jn

from time import mktime

from operator import itemgetter

from string import lower

from pylons import config, url

from pylons.i18n.translation import _

from vcs import get_backend

from rhodecode.lib import LANGUAGES_EXTENSIONS_MAP, safe_str

from rhodecode.lib.celerylib import run_task, locked_task, str2bool, \

    __get_lockkey, LockHeld, DaemonLock

from rhodecode.lib.helpers import person

from rhodecode.lib.rcmail.smtp_mailer import SmtpMailer

from rhodecode.lib.utils import add_cache, action_logger

from rhodecode.lib.compat import json, OrderedDict

from rhodecode.model import init_model

from rhodecode.model import meta

from rhodecode.model.db import Statistics, Repository, User

from sqlalchemy import engine_from_config

add_cache(config)

__all__ = ['whoosh_index', 'get_commits_stats',

           'reset_user_password', 'send_email']

def get_session():

    if CELERY_ON:

        engine = engine_from_config(config, 'sqlalchemy.db1.')

        init_model(engine)

    sa = meta.Session()

    return sa

def get_logger(cls):

    if CELERY_ON:

        try:

            log = cls.get_logger()

        except:

            added.append(name)

            form_data = {

                         'repo_name': name,

                         'repo_name_full': name,

                         'repo_type': repo.alias,

                         'description': repo.description \

                            if repo.description != 'unknown' else \

                                        '%s repository' % name,

                         'private': False,

                         'group_id': getattr(group, 'group_id', None)

                         }

            rm.create(form_data, user, just_db=True)

    removed = []

    if remove_obsolete:

        #remove from database those repositories that are not in the filesystem

        for repo in sa.query(Repository).all():

            if repo.repo_name not in initial_repo_list.keys():

                removed.append(repo.repo_name)

                sa.delete(repo)

                sa.commit()

    return added, removed

# set cache regions for beaker so celery can utilise it

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os

import logging

import datetime

import traceback

from sqlalchemy import *

from sqlalchemy.ext.hybrid import hybrid_property

from sqlalchemy.orm import relationship, joinedload, class_mapper, validates

from beaker.cache import cache_region, region_invalidate

from vcs import get_backend

from vcs.utils.helpers import get_scm

from vcs.exceptions import VCSError

from vcs.utils.lazy import LazyProperty

from rhodecode.lib.exceptions import UsersGroupsAssignedException

from rhodecode.lib.compat import json

from rhodecode.lib.caching_query import FromCache

from rhodecode.model.meta import Base, Session

log = logging.getLogger(__name__)

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

# BASE CLASSES

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

    __tablename__ = 'user_logs'

    __table_args__ = {'extend_existing':True}

    user_log_id = Column("user_log_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=False, unique=None, default=None)

    repository_id = Column("repository_id", Integer(), ForeignKey('repositories.repo_id'), nullable=False, unique=None, default=None)

    repository_name = Column("repository_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)

    user_ip = Column("user_ip", String(length=255, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)

    action = Column("action", UnicodeText(length=1200000, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)

    action_date = Column("action_date", DateTime(timezone=False), nullable=True, unique=None, default=None)

    @property

    def action_as_day(self):

    user = relationship('User')

    repository = relationship('Repository')

class UsersGroup(Base, BaseModel):

    __tablename__ = 'users_groups'

    __table_args__ = {'extend_existing':True}

    users_group_id = Column("users_group_id", Integer(), nullable=False, unique=True, default=None, primary_key=True)

    users_group_name = Column("users_group_name", String(length=255, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None)

    users_group_active = Column("users_group_active", Boolean(), nullable=True, unique=None, default=None)

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os

import logging

import traceback

import datetime

from pylons.i18n.translation import _

from rhodecode.lib import helpers as h

from rhodecode.model import BaseModel

from rhodecode.model.db import Notification, User, UserNotification

log = logging.getLogger(__name__)

class NotificationModel(BaseModel):

    def __get_user(self, user):

        if isinstance(user, basestring):

            return User.get_by_username(username=user)

        else:

            return self._get_instance(User, user)

    def __get_notification(self, notification):

        if isinstance(notification, Notification):

            return notification

        elif isinstance(notification, int):

            return Notification.get(notification)

        else:

            if notification:

                raise Exception('notification must be int or Instance'

                                ' of Notification got %s' % type(notification))

    def create(self, created_by, subject, body, recipients,

               type_=Notification.TYPE_MESSAGE):

        """

        Creates notification of given type

        :param created_by: int, str or User instance. User who created this

            notification

        :param subject:

        :param body:

        :param recipients: list of int, str or User objects

        :param type_: type of notification

        recipients_objs = []

        for u in recipients:

            obj = self.__get_user(u)

            if obj:

                recipients_objs.append(obj)

        recipients_objs = set(recipients_objs)

        notif = Notification.create(created_by=created_by_obj, subject=subject,

                                    body=body, recipients=recipients_objs,

                                    type_=type_)

        # send email with notification

        for rec in recipients_objs:

            email_subject = NotificationModel().make_description(notif, False)

            type_ = EmailNotificationModel.TYPE_CHANGESET_COMMENT

            email_body = body

            email_body_html = EmailNotificationModel()\

                            .get_email_tmpl(type_, **{'subject':subject,

                                                      'body':h.rst(body)})

            run_task(tasks.send_email, rec.email, email_subject, email_body,

                     email_body_html)

        return notif

                    when=when)

        return tmpl % data

class EmailNotificationModel(BaseModel):

    TYPE_CHANGESET_COMMENT = 'changeset_comment'

    TYPE_PASSWORD_RESET = 'passoword_link'

    TYPE_REGISTRATION = 'registration'

    TYPE_DEFAULT = 'default'

    def __init__(self):

        self.email_types = {

            self.TYPE_CHANGESET_COMMENT:'email_templates/changeset_comment.html',

            self.TYPE_PASSWORD_RESET:'email_templates/password_reset.html',

            self.TYPE_REGISTRATION:'email_templates/registration.html',

            self.TYPE_DEFAULT:'email_templates/default.html'

        }

    def get_email_tmpl(self, type_, **kwargs):

        """

        return generated template for email based on given type

        :param type_:

        """

        base = self.email_types.get(type_, self.TYPE_DEFAULT)

        # translator inject

        _kwargs = {'_':_}

        _kwargs.update(kwargs)

        log.debug('rendering tmpl %s with kwargs %s' % (base, _kwargs))

        return email_template.render(**_kwargs)

when this script is imported from the `nosetests --with-pylons=test.ini`

command.

This module initializes the application via ``websetup`` (`paster

setup-app`) and provides the base testing objects.

"""

import os

import time

import logging

from os.path import join as jn

from unittest import TestCase

from paste.deploy import loadapp

from paste.script.appinstall import SetupCommand

from pylons import config, url

from routes.util import URLGenerator

from webtest import TestApp

from rhodecode.model.db import User

import pylons.test

os.environ['TZ'] = 'UTC'

time.tzset()

log = logging.getLogger(__name__)

__all__ = ['environ', 'url', 'TestController', 'TESTS_TMP_PATH', 'HG_REPO',

           'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO', 'HG_FORK', 'GIT_FORK',

           'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS' ]

# Invoke websetup with the current config file

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

##RUNNING DESIRED TESTS

# nosetests -x rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account

# nosetests --pdb --pdb-failures 

environ = {}

#SOME GLOBALS FOR TESTS

TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())

TEST_USER_ADMIN_LOGIN = 'test_admin'

TEST_USER_ADMIN_PASS = 'test12'

HG_REPO = 'vcs_test_hg'

GIT_REPO = 'vcs_test_git'

NEW_HG_REPO = 'vcs_test_hg_new'

NEW_GIT_REPO = 'vcs_test_git_new'

HG_FORK = 'vcs_test_hg_fork'

GIT_FORK = 'vcs_test_git_fork'

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.index_location = config['app_conf']['index_dir']

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

    def log_user(self, username=TEST_USER_ADMIN_LOGIN,

                 password=TEST_USER_ADMIN_PASS):

        self._logged_username = username

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

                                 {'username':username,

                                  'password':password})

        if 'invalid user name' in response.body:

            self.fail('could not login using %s %s' % (username, password))

        self.assertEqual(response.status, '302 Found')

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def checkSessionFlash(self, response, msg):

        self.assertTrue('flash' in response.session)

        self.assertTrue(msg in response.session['flash'][0][1])

    def test_show(self):

        self.log_user()

        cur_user = self._get_logged_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                               email='u1@rhodecode.org',

                                               name='u1', lastname='u1')

        u2 = UserModel().create_or_update(username='u2', password='qweqwe',

                                               email='u2@rhodecode.org',

                                               name='u2', lastname='u2')

        notification = NotificationModel().create(created_by=cur_user,

                                                  body=u'hi there',

                                                  recipients=[cur_user, u1, u2])

        response = self.app.get(url('notification',

                                    notification_id=notification.notification_id))

#    def test_show_as_xml(self):

#        response = self.app.get(url('formatted_notification', notification_id=1, format='xml'))

#

#    def test_edit(self):

#        response = self.app.get(url('edit_notification', notification_id=1))

#

from rhodecode.tests import *

class TestBranchesController(TestController):

    def test_index(self):

        self.log_user()

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

                                repo_name=HG_REPO, revision=rev))

        # test DB

        self.assertEqual(ChangesetComment.query().count(), 1)

        self.assertTrue('''<div class="comments-number">%s '''

                        '''comment(s) (0 inline)</div>''' % 1 in response.body)

        self.assertEqual(Notification.query().count(), 2)

        users = [x.user.username for x in UserNotification.query().all()]

        # test_regular get's notification by @mention

    def test_delete(self):

        self.log_user()

        rev = '27cd5cce30c96924232dffcd24178a07ffeb5dfc'

        text = u'CommentOnRevision'

        params = {'text':text}

        response = self.app.post(url(controller='changeset', action='comment',

                                     repo_name=HG_REPO, revision=rev),

                                     params=params)

        comments = ChangesetComment.query().all()

        self.assertEqual(len(comments), 1)

        comment_id = comments[0].comment_id

        self.app.delete(url(controller='changeset',

                                    action='delete_comment',

                                    repo_name=HG_REPO,

        comments = ChangesetComment.query().all()

        self.assertEqual(len(comments), 0)

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

                                repo_name=HG_REPO, revision=rev))

        self.assertTrue('''<div class="comments-number">0 comment(s)'''

                        ''' (0 inline)</div>''' in response.body)

                                    repo_name=repo_name))

        self.assertTrue("""There are no forks yet""" in response.body)

    def test_index_with_fork(self):

        self.log_user()

        # create a fork

        fork_name = HG_FORK

        description = 'fork of vcs test'

        repo_name = HG_REPO

                                     action='fork_create',

                                    repo_name=repo_name),

                                     'repo_type':'hg',

                                     'description':description,

                                     'private':'False'})

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

                                    repo_name=repo_name))

        self.assertTrue("""<a href="/%s/summary">"""

                         """vcs_test_hg_fork</a>""" % fork_name

                         in response.body)

        #remove this fork

        response = self.app.delete(url('repo', repo_name=fork_name))

from rhodecode.model.db import Repository

from rhodecode.tests import *

class TestSettingsController(TestController):

    def test_index(self):

        self.log_user()

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

                                    repo_name=HG_REPO))

        # Test response...