(STATUS_NOT_REVIEWED, _("Not Reviewed")),  # (no icon) and default

        (STATUS_APPROVED, _("Approved")),

        (STATUS_REJECTED, _("Rejected")),

        (STATUS_UNDER_REVIEW, _("Under Review")),

    ]

    changeset_status_id = Column('changeset_status_id', Integer(), nullable=False, primary_key=True)

    repo_id = Column('repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)

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

    revision = Column('revision', String(40), nullable=False)

    status = Column('status', String(128), nullable=False, default=DEFAULT)

    changeset_comment_id = Column('changeset_comment_id', Integer(), ForeignKey('changeset_comments.comment_id'))

    modified_at = Column('modified_at', DateTime(), nullable=False, default=datetime.datetime.now)

    version = Column('version', Integer(), nullable=False, default=0)

    pull_request_id = Column("pull_request_id", Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)

    author = relationship('User', lazy='joined')

    repo = relationship('Repository')

    comment = relationship('ChangesetComment', lazy='joined')

    pull_request = relationship('PullRequest', lazy='joined')

    def __unicode__(self):

        return u"<%s('%s:%s')>" % (

            self.__class__.__name__,

            self.status, self.author

        )

    @classmethod

    def get_status_lbl(cls, value):

        return dict(cls.STATUSES).get(value)

    @property

    def status_lbl(self):

        return ChangesetStatus.get_status_lbl(self.status)

class PullRequest(Base, BaseModel):

    __tablename__ = 'pull_requests'

    __table_args__ = (

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'},

    )

    STATUS_NEW = u'new'

    STATUS_OPEN = u'open'

    STATUS_CLOSED = u'closed'

    pull_request_id = Column('pull_request_id', Integer(), nullable=False, primary_key=True)

    title = Column('title', Unicode(256), nullable=True)

    description = Column('description', UnicodeText(10240), nullable=True)

    status = Column('status', Unicode(256), nullable=False, default=STATUS_NEW)

    created_on = Column('created_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    updated_on = Column('updated_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

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

    _revisions = Column('revisions', UnicodeText(20500))  # 500 revisions max

    org_repo_id = Column('org_repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    org_ref = Column('org_ref', Unicode(256), nullable=False)

    other_repo_id = Column('other_repo_id', Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    other_ref = Column('other_ref', Unicode(256), nullable=False)

    @hybrid_property

    def revisions(self):

        return self._revisions.split(':')

    @revisions.setter

    def revisions(self, val):

        self._revisions = ':'.join(val)

    author = relationship('User', lazy='joined')

    reviewers = relationship('PullRequestReviewers',

                             cascade="all, delete, delete-orphan")

    org_repo = relationship('Repository', primaryjoin='PullRequest.org_repo_id==Repository.repo_id')

    other_repo = relationship('Repository', primaryjoin='PullRequest.other_repo_id==Repository.repo_id')

    statuses = relationship('ChangesetStatus')

    comments = relationship('ChangesetComment',

                             cascade="all, delete, delete-orphan")

    def is_closed(self):

        return self.status == self.STATUS_CLOSED

    def __json__(self):

        return dict(

          revisions=self.revisions

        )

class PullRequestReviewers(Base, BaseModel):

    __tablename__ = 'pull_request_reviewers'

    __table_args__ = (

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'},

    )

    def __init__(self, user=None, pull_request=None):

        self.user = user

        self.pull_request = pull_request

    pull_requests_reviewers_id = Column('pull_requests_reviewers_id', Integer(), nullable=False, primary_key=True)

    pull_request_id = Column("pull_request_id", Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=False)

    user_id = Column("user_id", Integer(), ForeignKey('users.user_id'), nullable=True)

    user = relationship('User')

    pull_request = relationship('PullRequest')

class Notification(Base, BaseModel):

    __tablename__ = 'notifications'

    __table_args__ = (

        Index('notification_type_idx', 'type'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'},

    )

    TYPE_CHANGESET_COMMENT = u'cs_comment'

    TYPE_MESSAGE = u'message'

    TYPE_MENTION = u'mention'

    TYPE_REGISTRATION = u'registration'

    TYPE_PULL_REQUEST = u'pull_request'

    TYPE_PULL_REQUEST_COMMENT = u'pull_request_comment'

    notification_id = Column('notification_id', Integer(), nullable=False, primary_key=True)

    subject = Column('subject', Unicode(512), nullable=True)

    body = Column('body', UnicodeText(50000), nullable=True)

    created_by = Column("created_by", Integer(), ForeignKey('users.user_id'), nullable=True)

    created_on = Column('created_on', DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    type_ = Column('type', Unicode(256))

    created_by_user = relationship('User')

    notifications_to_users = relationship('UserNotification', lazy='joined',

                                          cascade="all, delete, delete-orphan")

    @property

    def recipients(self):

        return [x.user for x in UserNotification.query()\

                .filter(UserNotification.notification == self)\

                .order_by(UserNotification.user_id.asc()).all()]

    @classmethod

    def create(cls, created_by, subject, body, recipients, type_=None):

        if type_ is None:

            type_ = Notification.TYPE_MESSAGE

        notification = cls()

        notification.created_by_user = created_by

        notification.subject = subject

        notification.body = body

        notification.type_ = type_

        notification.created_on = datetime.datetime.now()

        for u in recipients:

            assoc = UserNotification()

            assoc.notification = notification

            u.notifications.append(assoc)

        Session().add(notification)

        return notification

    @property

    def description(self):

        from rhodecode.model.notification import NotificationModel

        return NotificationModel().make_description(self)

class UserNotification(Base, BaseModel):

    __tablename__ = 'user_to_notification'

    __table_args__ = (

        UniqueConstraint('user_id', 'notification_id'),

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'}

    )

    user_id = Column('user_id', Integer(), ForeignKey('users.user_id'), primary_key=True)

    notification_id = Column("notification_id", Integer(), ForeignKey('notifications.notification_id'), primary_key=True)

    read = Column('read', Boolean, default=False)

    sent_on = Column('sent_on', DateTime(timezone=False), nullable=True, unique=None)

    user = relationship('User', lazy="joined")

    notification = relationship('Notification', lazy="joined",

                                order_by=lambda: Notification.created_on.desc(),)

    def mark_as_read(self):

        self.read = True

        Session().add(self)

class DbMigrateVersion(Base, BaseModel):

    __tablename__ = 'db_migrate_version'

    __table_args__ = (

        {'extend_existing': True, 'mysql_engine': 'InnoDB',

         'mysql_charset': 'utf8'},

    )

    repository_id = Column('repository_id', String(250), primary_key=True)

    repository_path = Column('repository_path', Text)

    version = Column('version', Integer)

class LazyProperty(object):

    """

    Decorator for easier creation of ``property`` from potentially expensive to

    calculate attribute of the class.

    Usage::

      class Foo(object):

          @LazyProperty

          def bar(self):

              print 'Calculating self._bar'

              return 42

    Taken from http://blog.pythonisito.com/2008/08/lazy-descriptors.html and

    used widely.

    """

    def __init__(self, func):

        self._func = func

        self.__module__ = func.__module__

        self.__name__ = func.__name__

        self.__doc__ = func.__doc__

    def __get__(self, obj, klass=None):

        if obj is None:

            return self

        result = obj.__dict__[self.__name__] = self._func(obj)

        return result

import threading

class ThreadLocalLazyProperty(LazyProperty):

    """

    Same as above but uses thread local dict for cache storage.

    """

    def __get__(self, obj, klass=None):

        if obj is None:

            return self

        if not hasattr(obj, '__tl_dict__'):

            obj.__tl_dict__ = threading.local().__dict__

        result = obj.__tl_dict__[self.__name__] = self._func(obj)

        return result

"""Pylons application test package

This package assumes the Pylons environment is already loaded, such as

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

import datetime

import hashlib

import tempfile

from os.path import join as jn

from unittest import TestCase

from tempfile import _RandomNameSequence

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 import is_windows

from rhodecode.model.meta import Session

from rhodecode.model.db import User

from rhodecode.tests.nose_parametrized import parameterized

import pylons.test

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

if not is_windows:

    time.tzset()

log = logging.getLogger(__name__)

__all__ = [

    'parameterized', 'environ', 'url', 'get_new_dir', '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',

    'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',

    'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',

    'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',

    'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',

    'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',

    'GIT_REMOTE_REPO', 'SCM_TESTS', '_get_repo_create_params'

]

# 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

# nosetests --with-coverage --cover-package=rhodecode.model.validators rhodecode.tests.test_validators

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'

TEST_USER_ADMIN_EMAIL = 'test_admin@mail.com'

TEST_USER_REGULAR_LOGIN = 'test_regular'

TEST_USER_REGULAR_PASS = 'test12'

TEST_USER_REGULAR_EMAIL = 'test_regular@mail.com'

TEST_USER_REGULAR2_LOGIN = 'test_regular2'

TEST_USER_REGULAR2_PASS = 'test12'

TEST_USER_REGULAR2_EMAIL = 'test_regular2@mail.com'

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'

## VCS

SCM_TESTS = ['hg', 'git']

uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))

GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'

TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)

TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)

TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)

HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'

TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)

TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)

TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)

TEST_DIR = tempfile.gettempdir()

TEST_REPO_PREFIX = 'vcs-test'

# cached repos if any !

# comment out to get some other repos from bb or github

GIT_REMOTE_REPO = jn(TESTS_TMP_PATH, GIT_REPO)

HG_REMOTE_REPO = jn(TESTS_TMP_PATH, HG_REPO)

def get_new_dir(title):

    """

    Returns always new directory path.

    """

    from rhodecode.tests.vcs.utils import get_normalized_path

    name = TEST_REPO_PREFIX

    if title:

        name = '-'.join((name, title))

    hex = hashlib.sha1(str(time.time())).hexdigest()

    name = '-'.join((name, hex))

    path = os.path.join(TEST_DIR, name)

    return get_normalized_path(path)

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.Session = Session

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

        ses = response.session['rhodecode_user']

        self.assertEqual(ses.get('username'), username)

        response = response.follow()

        self.assertEqual(ses.get('is_authenticated'), True)

        return response.session['rhodecode_user']

    def _get_logged_user(self):

        return User.get_by_username(self._logged_username)

    def checkSessionFlash(self, response, msg):

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

        if not msg in response.session['flash'][0][1]:

            self.fail(

                'msg `%s` not found in session flash: got `%s` instead' % (

                      msg, response.session['flash'])

            )

## HELPERS ##

def _get_repo_create_params(**custom):

    defs = {

        'repo_name': None,

        'repo_type': 'hg',

        'clone_uri': '',

        'repo_group': '',

        'repo_description': 'DESC',

        'repo_private': False,

        'repo_landing_rev': 'tip'

    }

    defs.update(custom)

    if 'repo_name_full' not in custom:

        defs.update({'repo_name_full': defs['repo_name']})

    return defs