DEFAULT_ENCODINGS = aslist(rhodecode.CONFIG.get('default_encoding',

                                                        'utf8'), sep=',')

        to_encoding = DEFAULT_ENCODINGS

    if not isinstance(to_encoding, (list, tuple)):

        to_encoding = [to_encoding]

    for enc in to_encoding:

        try:

            return unicode_.encode(enc)

        except UnicodeEncodeError:

            pass

    try:

        import chardet

        encoding = chardet.detect(unicode_)['encoding']

        if encoding is None:

            raise UnicodeEncodeError()

        return unicode_.encode(encoding)

    except (ImportError, UnicodeEncodeError):

        return unicode_.encode(to_encoding[0], 'replace')

    return safe_str

def remove_suffix(s, suffix):

    if s.endswith(suffix):

        s = s[:-1 * len(suffix)]

    return s

def remove_prefix(s, prefix):

    if s.startswith(prefix):

        s = s[len(prefix):]

    return s

def engine_from_config(configuration, prefix='sqlalchemy.', **kwargs):

    """

    Custom engine_from_config functions that makes sure we use NullPool for

    file based sqlite databases. This prevents errors on sqlite. This only

    applies to sqlalchemy versions < 0.7.0

    """

    import sqlalchemy

    from sqlalchemy import engine_from_config as efc

    import logging

    if int(sqlalchemy.__version__.split('.')[1]) < 7:

        # This solution should work for sqlalchemy < 0.7.0, and should use

        # proxy=TimerProxy() for execution time profiling

        from sqlalchemy.pool import NullPool

        url = configuration[prefix + 'url']

        if url.startswith('sqlite'):

            kwargs.update({'poolclass': NullPool})

        return efc(configuration, prefix, **kwargs)

    else:

        import time

        from sqlalchemy import event

        from sqlalchemy.engine import Engine

        log = logging.getLogger('sqlalchemy.engine')

        BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = xrange(30, 38)

        engine = efc(configuration, prefix, **kwargs)

        def color_sql(sql):

            COLOR_SEQ = "\033[1;%dm"

            COLOR_SQL = YELLOW

            normal = '\x1b[0m'

            return ''.join([COLOR_SEQ % COLOR_SQL, sql, normal])

        if configuration['debug']:

            #attach events only for debug configuration

            def before_cursor_execute(conn, cursor, statement,

                                    parameters, context, executemany):

                context._query_start_time = time.time()

                log.info(color_sql(">>>>> STARTING QUERY >>>>>"))

            def after_cursor_execute(conn, cursor, statement,

                                    parameters, context, executemany):

                total = time.time() - context._query_start_time

                log.info(color_sql("<<<<< TOTAL TIME: %f <<<<<" % total))

            event.listen(engine, "before_cursor_execute",

                         before_cursor_execute)

            event.listen(engine, "after_cursor_execute",

                         after_cursor_execute)

    return engine

    """

    turns a datetime into an age string.

    If show_short_version is True, then it will generate a not so accurate but shorter string,

    example: 2days ago, instead of 2 days and 23 hours ago.

    :param prevdate: datetime object

    :param show_short_version: if it should aproximate the date and return a shorter string

    :rtype: unicode

    :returns: unicode words describing age

    """

    order = ['year', 'month', 'day', 'hour', 'minute', 'second']

    deltas = {}

    future = False

    if prevdate > now:

        now, prevdate = prevdate, now

        future = True

    # Get date parts deltas

    for part in order:

    # Fix negative offsets (there is 1 second between 10:59:59 and 11:00:00,

    # not 1 hour, -59 minutes and -59 seconds)

    for num, length in [(5, 60), (4, 60), (3, 24)]:  # seconds, minutes, hours

        part = order[num]

        carry_part = order[num - 1]

        if deltas[part] < 0:

            deltas[part] += length

            deltas[carry_part] -= 1

    # Same thing for days except that the increment depends on the (variable)

    # number of days in the month

    month_lengths = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]

    if deltas['day'] < 0:

        if prevdate.month == 2 and (prevdate.year % 4 == 0 and

            (prevdate.year % 100 != 0 or prevdate.year % 400 == 0)):

            deltas['day'] += 29

        else:

            deltas['day'] += month_lengths[prevdate.month - 1]

        deltas['month'] -= 1

    if deltas['month'] < 0:

        deltas['month'] += 12

        deltas['year'] -= 1

    # Format the result

    fmt_funcs = {

        'year': lambda d: ungettext(u'%d year', '%d years', d) % d,

        'month': lambda d: ungettext(u'%d month', '%d months', d) % d,

        'day': lambda d: ungettext(u'%d day', '%d days', d) % d,

        'hour': lambda d: ungettext(u'%d hour', '%d hours', d) % d,

        'minute': lambda d: ungettext(u'%d minute', '%d minutes', d) % d,

        'second': lambda d: ungettext(u'%d second', '%d seconds', d) % d,

    }

    for i, part in enumerate(order):

        value = deltas[part]

        if value == 0:

            continue

        if i < 5:

            sub_part = order[i + 1]

            sub_value = deltas[sub_part]

        else:

            sub_value = 0

        if sub_value == 0 or show_short_version:

            if future:

                return _(u'in %s') % fmt_funcs[part](value)

            else:

                return _(u'%s ago') % fmt_funcs[part](value)

        if future:

            return _(u'in %s and %s') % (fmt_funcs[part](value),

                fmt_funcs[sub_part](sub_value))

        else:

            return _(u'%s and %s ago') % (fmt_funcs[part](value),

                fmt_funcs[sub_part](sub_value))

    return _(u'just now')

def uri_filter(uri):

    """

    Removes user:password from given url string

    :param uri:

    :rtype: unicode

    :returns: filtered list of strings

    """

    if not uri:

        return ''

    proto = ''

    for pat in ('https://', 'http://'):

        if uri.startswith(pat):

            uri = uri[len(pat):]

            proto = pat

            break

    # remove passwords and username

    uri = uri[uri.find('@') + 1:]

    # get the port

    cred_pos = uri.find(':')

    if cred_pos == -1:

        host, port = uri, None

    else:

        host, port = uri[:cred_pos], uri[cred_pos + 1:]

    return filter(None, [proto, host, port])

def credentials_filter(uri):

# 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/>.

from __future__ import with_statement

import unittest

import datetime

import hashlib

import mock

from rhodecode.tests import *

proto = 'http'

TEST_URLS = [

    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

     '%s://127.0.0.1:8080' % proto),

    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],

     '%s://domain.org' % proto),

    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',

                                                '8080'],

     '%s://domain.org:8080' % proto),

]

proto = 'https'

TEST_URLS += [

    ('%s://127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://marcink:pass@127.0.0.1' % proto, ['%s://' % proto, '127.0.0.1'],

     '%s://127.0.0.1' % proto),

    ('%s://127.0.0.1:8080' % proto, ['%s://' % proto, '127.0.0.1', '8080'],

     '%s://127.0.0.1:8080' % proto),

    ('%s://domain.org' % proto, ['%s://' % proto, 'domain.org'],

     '%s://domain.org' % proto),

    ('%s://user:pass@domain.org:8080' % proto, ['%s://' % proto, 'domain.org',

                                                '8080'],

     '%s://domain.org:8080' % proto),

]

class TestLibs(unittest.TestCase):

    @parameterized.expand(TEST_URLS)

    def test_uri_filter(self, test_url, expected, expected_creds):

        from rhodecode.lib.utils2 import uri_filter

        self.assertEqual(uri_filter(test_url), expected)

    @parameterized.expand(TEST_URLS)

    def test_credentials_filter(self, test_url, expected, expected_creds):

        from rhodecode.lib.utils2 import credentials_filter

        self.assertEqual(credentials_filter(test_url), expected_creds)

    @parameterized.expand([('t', True),

                           ('true', True),

                           ('y', True),

                           ('yes', True),

                           ('on', True),

                           ('1', True),

                           ('Y', True),

                           ('yeS', True),

                           ('Y', True),

                           ('TRUE', True),

                           ('T', True),

                           ('False', False),

                           ('F', False),

                           ('FALSE', False),

                           ('0', False),

                           ('-1', False),

                           ('', False)

    ])

    def test_str2bool(self, str_bool, expected):

        from rhodecode.lib.utils2 import str2bool

        self.assertEqual(str2bool(str_bool), expected)

    def test_mention_extractor(self):

        from rhodecode.lib.utils2 import extract_mentioned_users

        sample = (

            "@first hi there @marcink here's my email marcin@email.com "

            "@lukaszb check @one_more22 it pls @ ttwelve @D[] @one@two@three "

            "@MARCIN    @maRCiN @2one_more22 @john please see this http://org.pl "

            "@marian.user just do it @marco-polo and next extract @marco_polo "

            "user.dot  hej ! not-needed maril@domain.org"

        )

        s = sorted([

        'first', 'marcink', 'lukaszb', 'one_more22', 'MARCIN', 'maRCiN', 'john',

        'marian.user', 'marco-polo', 'marco_polo'

        ], key=lambda k: k.lower())

        self.assertEqual(s, extract_mentioned_users(sample))

        from rhodecode.lib.utils2 import age

        from dateutil import relativedelta

        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

        from rhodecode.lib.utils2 import age

        from dateutil import relativedelta

        delt = lambda *args, **kwargs: relativedelta.relativedelta(*args, **kwargs)

    def test_tag_exctrator(self):

        sample = (

            "hello pta[tag] gog [[]] [[] sda ero[or]d [me =>>< sa]"

            "[requires] [stale] [see<>=>] [see => http://url.com]"

            "[requires => url] [lang => python] [just a tag]"

            "[,d] [ => ULR ] [obsolete] [desc]]"

        )

        from rhodecode.lib.helpers import desc_stylize

        res = desc_stylize(sample)

        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)

        self.assertTrue('<div class="metatag" tag="obsolete">obsolete</div>' in res)

        self.assertTrue('<div class="metatag" tag="stale">stale</div>' in res)

        self.assertTrue('<div class="metatag" tag="lang">python</div>' in res)

        self.assertTrue('<div class="metatag" tag="requires">requires =&gt; <a href="/url">url</a></div>' in res)

        self.assertTrue('<div class="metatag" tag="tag">tag</div>' in res)

    def test_alternative_gravatar(self):

        from rhodecode.lib.helpers import gravatar_url

        _md5 = lambda s: hashlib.md5(s).hexdigest()

        def fake_conf(**kwargs):

            from pylons import config

            config['app_conf'] = {}

            config['app_conf']['use_gravatar'] = True

            config['app_conf'].update(kwargs)

            return config

        class fake_url():

            @classmethod

            def current(cls, *args, **kwargs):

                return 'https://server.com'

        with mock.patch('pylons.url', fake_url):

            fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')

            with mock.patch('pylons.config', fake):

                    from pylons import url

                    assert url.current() == 'https://server.com'

                    grav = gravatar_url(email_address='test@foo.com', size=24)

                    assert grav == 'http://test.com/test@foo.com'

            fake = fake_conf(alternative_gravatar_url='http://test.com/{email}')

            with mock.patch('pylons.config', fake):

                grav = gravatar_url(email_address='test@foo.com', size=24)

                assert grav == 'http://test.com/test@foo.com'

            fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}')

            with mock.patch('pylons.config', fake):

                em = 'test@foo.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'http://test.com/%s' % (_md5(em))

            fake = fake_conf(alternative_gravatar_url='http://test.com/{md5email}/{size}')

            with mock.patch('pylons.config', fake):

                em = 'test@foo.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'http://test.com/%s/%s' % (_md5(em), 24)

            fake = fake_conf(alternative_gravatar_url='{scheme}://{netloc}/{md5email}/{size}')

            with mock.patch('pylons.config', fake):

                em = 'test@foo.com'

                grav = gravatar_url(email_address=em, size=24)

                assert grav == 'https://server.com/%s/%s' % (_md5(em), 24)

    def _quick_url(self, text, tmpl="""<a class="revision-link" href="%s">%s</a>""", url_=None):

        """

        Changes `some text url[foo]` => `some text <a href="/">foo</a>

        :param text:

        """

        import re

        URL_PAT = re.compile(r'(?:url\[)(.+?)(?:\])')

        def url_func(match_obj):

            _url = match_obj.groups()[0]

            return tmpl % (url_ or '/some-url', _url)

        return URL_PAT.sub(url_func, text)

    @parameterized.expand([

      ("",

       ""),

      ("git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68",

       "git-svn-id: https://svn.apache.org/repos/asf/libcloud/trunk@1441655 13f79535-47bb-0310-9956-ffa450edef68"),

      ("from rev 000000000000",

       "from rev url[000000000000]"),

      ("from rev 000000000000123123 also rev 000000000000",

       "from rev url[000000000000123123] also rev url[000000000000]"),

      ("this should-000 00",

       "this should-000 00"),

      ("longtextffffffffff rev 123123123123",

       "longtextffffffffff rev url[123123123123]"),

      ("rev ffffffffffffffffffffffffffffffffffffffffffffffffff",

       "rev ffffffffffffffffffffffffffffffffffffffffffffffffff"),

      ("ffffffffffff some text traalaa",

       "url[ffffffffffff] some text traalaa"),

       ("""Multi line

       123123123123

       some text 123123123123

       sometimes !

       """,

       """Multi line

       url[123123123123]

       some text url[123123123123]

       sometimes !

       """)

    ])

    def test_urlify_changesets(self, sample, expected):

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

            return '/some-url'

        expected = self._quick_url(expected)

        with mock.patch('pylons.url', fake_url):

            from rhodecode.lib.helpers import urlify_changesets

            self.assertEqual(urlify_changesets(sample, 'repo_name'), expected)

    @parameterized.expand([

      ("",

       "",

       ""),

      ("https://svn.apache.org/repos",

       "url[https://svn.apache.org/repos]",

       "https://svn.apache.org/repos"),

      ("http://svn.apache.org/repos",

       "url[http://svn.apache.org/repos]",

       "http://svn.apache.org/repos"),

      ("from rev a also rev http://google.com",

       "from rev a also rev url[http://google.com]",

       "http://google.com"),

       ("""Multi line

       https://foo.bar.com

       some text lalala""",

       """Multi line

       url[https://foo.bar.com]

       some text lalala""",

       "https://foo.bar.com")

    ])

    def test_urlify_test(self, sample, expected, url_):

        from rhodecode.lib.helpers import urlify_text

        expected = self._quick_url(expected,

                                   tmpl="""<a href="%s">%s</a>""", url_=url_)

        self.assertEqual(urlify_text(sample), expected)