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

BACKENDS = {

    'hg': 'Mercurial repository',

    #'git': 'Git repository',

}

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__)

                 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

    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)

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

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

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

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

                            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)

# 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

    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):

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)

            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

        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()\

    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)

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()

##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'

    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

                                  '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])

                                               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))

from rhodecode.tests import *

class TestBranchesController(TestController):

    def test_index(self):

        self.log_user()

        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'

        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.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))

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

                         in response.body)

        #remove this fork

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

    def test_index(self):

        self.log_user()

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

                                    repo_name=HG_REPO))

        # Test response...