# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

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

# Sequences

RESET_SEQ = "\033[0m"

COLOR_SEQ = "\033[0;%dm"

BOLD_SEQ = "\033[1m"

COLORS = {

    'CRITICAL': MAGENTA,

    'ERROR': RED,

    'WARNING': CYAN,

    'INFO': GREEN,

    'DEBUG': BLUE,

    'SQL': YELLOW

}

def one_space_trim(s):

    if s.find("  ") == -1:

        return s

    else:

        s = s.replace('  ', ' ')

        return one_space_trim(s)

    if i == 0:

        return v, t, p

    if i == 1:

        return q, v, p

    if i == 2:

        return p, v, t

    if i == 3:

        return p, q, v

    if i == 4:

        return t, p, v

    if i == 5:

        return v, p, q

def gen_color(n=10000):

    """generator for getting n of evenly distributed colors using

    hsv color and golden ratio. It always return same order of colors

    :returns: RGB tuple

    """

    golden_ratio = 0.618033988749895

    h = 0.22717784590367374

        h += golden_ratio

        h %= 1

        HSV_tuple = [h, 0.95, 0.95]

        RGB_tuple = hsv_to_rgb(*HSV_tuple)

        yield [str(int(x * 256)) for x in RGB_tuple]

def pygmentize_annotation(repo_name, filenode, **kwargs):

    """

    pygmentize function for annotation

    :param filenode:

    """

    cgenerator = gen_color()

    color_dict = {}

    def get_color_string(cs):

        if cs in color_dict:

            col = color_dict[cs]

        else:

            col = color_dict[cs] = next(cgenerator)

        return "color: rgb(%s)! important;" % (', '.join(col))

    def url_func(changeset):

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

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

import time

from sqlalchemy.interfaces import ConnectionProxy

log = logging.getLogger('timerproxy')

def color_sql(sql):

    COLOR_SEQ = "\033[1;%dm"

    COLOR_SQL = YELLOW

    normal = '\x1b[0m'

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

class TimerProxy(ConnectionProxy):

    def __init__(self):

        super(TimerProxy, self).__init__()

    def cursor_execute(self, execute, cursor, statement, parameters,

                       context, executemany):

        now = time.time()

        try:

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

            return execute(cursor, statement, parameters, context)

        finally:

            total = time.time() - now

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

    default_elements = ['percentage', 'bar', 'steps']

    def __init__(self, steps=100, stream=None, elements=None):

        self.step = 0

        self.steps = steps

        self.stream = stream or sys.stderr

        self.bar_char = '='

        self.width = 50

        self.separator = ' | '

        self.elements = elements or self.default_elements

        self.started = None

        self.finished = False

        self.steps_label = 'Step'

        self.time_label = 'Time'

        self.eta_label = 'ETA'

        self.speed_label = 'Speed'

        self.transfer_label = 'Transfer'

    def __str__(self):

        return self.get_line()

    def __iter__(self):

        start = self.step

        end = self.steps + 1

            self.render(x)

            yield x

    def get_separator(self):

        return self.separator

    def get_bar_char(self):

        return self.bar_char

    def get_bar(self):

        char = self.get_bar_char()

        perc = self.get_percentage()

        length = int(self.width * perc / 100)

        bar = char * length

        bar = bar.ljust(self.width)

        return bar

    def get_elements(self):

        return self.elements

    def get_template(self):

        separator = self.get_separator()

        elements = self.get_elements()

        return string.Template(separator.join((('$%s' % e) for e in elements)))

        perc = self.get_percentage()

        perc_text = '%s%%' % int(perc)

        text = (' %s%% ' % (perc_text)).center(self.width, '=')

        L = text.find(' ')

        R = text.rfind(' ')

        bar = ' '.join((bar[:L], perc_text, bar[R:]))

        return bar

class AnimatedColoredProgressBar(AnimatedProgressBar,

                                 ColoredProgressBar):

    pass

class BarOnlyColoredProgressBar(ColoredProgressBar,

                                BarOnlyProgressBar):

    pass

def main():

    import time

    print("Standard progress bar...")

    bar = ProgressBar(30)

        bar.render(x)

        time.sleep(0.02)

    bar.stream.write('\n')

    print()

    print("Empty bar...")

    bar = ProgressBar(50)

    bar.render(0)

    print()

    print()

    print("Colored bar...")

    bar = ColoredProgressBar(20)

    for x in bar:

        time.sleep(0.01)

    print()

    print("Animated char bar...")

    bar = AnimatedProgressBar(20)

    for x in bar:

        time.sleep(0.01)

    print()

    print("Animated + colored char bar...")

        time.sleep(0.01)

    print()

    print("Bar only ...")

    bar = BarOnlyProgressBar(20)

    for x in bar:

        time.sleep(0.01)

    print()

    print("Colored, longer bar-only, eta, total time ...")

    bar = BarOnlyColoredProgressBar(40)

    bar.width = 60

    bar.elements += ['time', 'eta']

    for x in bar:

        time.sleep(0.01)

    print()

    print()

    print("File transfer bar, breaks after 2 seconds ...")

    total_bytes = 1024 * 1024 * 2

    bar = ProgressBar(total_bytes)

    bar.width = 50

    bar.elements.remove('steps')

    bar.elements += ['transfer', 'time', 'eta', 'speed']

        bar.render(x)

        time.sleep(0.01)

        now = datetime.datetime.now()

        if now - bar.started >= datetime.timedelta(seconds=2):

            break

    print()

    print()

if __name__ == '__main__':

    main()

import os

import random

import shutil

import time

import traceback

from kallithea.lib import ext_json

from kallithea.lib.utils2 import AttributeDict, ascii_bytes, safe_int, safe_unicode, time_to_datetime

from kallithea.model.db import Gist, Session, User

from kallithea.model.repo import RepoModel

from kallithea.model.scm import ScmModel

log = logging.getLogger(__name__)

GIST_STORE_LOC = '.rc_gist_store'

GIST_METADATA_FILE = '.rc_gist_metadata'

def make_gist_access_id():

    """Generate a random, URL safe, almost certainly unique gist identifier."""

    rnd = random.SystemRandom() # use cryptographically secure system PRNG

    alphabet = '23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghjklmnpqrstuvwxyz'

    length = 20

class GistModel(object):

    def __delete_gist(self, gist):

        """

        removes gist from filesystem

        :param gist: gist object

        """

        root_path = RepoModel().repos_path

        rm_path = os.path.join(root_path, GIST_STORE_LOC, gist.gist_access_id)

        log.info("Removing %s", rm_path)

        shutil.rmtree(rm_path)

    def _store_metadata(self, repo, gist_id, gist_access_id, user_id, gist_type,

                        gist_expires):

        """

        store metadata inside the gist, this can be later used for imports

        or gist identification

        """

        metadata = {

            'metadata_version': '1',

            'gist_db_id': gist_id,

        return stdout, stderr

def _get_tmp_dir(prefix='vcs_operations-', suffix=''):

    return tempfile.mkdtemp(dir=base.TESTS_TMP_PATH, prefix=prefix, suffix=suffix)

def _add_files(vcs, dest_dir, files_no=3):

    """

    Generate some files, add it to dest_dir repo and push back

    vcs is git or hg and defines what VCS we want to make those files for

    :param vcs:

    :param dest_dir:

    """

    added_file = '%ssetup.py' % next(_RandomNameSequence())

    open(os.path.join(dest_dir, added_file), 'a').close()

    Command(dest_dir).execute(vcs, 'add', added_file)

    email = 'me@example.com'

    if os.name == 'nt':

        author_str = 'User <%s>' % email

    else:

        author_str = 'User ǝɯɐᴎ <%s>' % email

        cmd = """echo "added_line%s" >> %s""" % (i, added_file)

        Command(dest_dir).execute(cmd)

        if vcs == 'hg':

            cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (

                i, author_str, added_file

            )

        elif vcs == 'git':

            cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (

                i, author_str, added_file

            )

        # git commit needs EMAIL on some machines

        Command(dest_dir).execute(cmd, EMAIL=email)

def _add_files_and_push(webserver, vt, dest_dir, clone_url, ignoreReturnCode=False, files_no=3):

    _add_files(vt.repo_type, dest_dir, files_no=files_no)

    # PUSH it back

    stdout = stderr = None

    if vt.repo_type == 'hg':

        stdout, stderr = Command(dest_dir).execute('hg push -f --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)

    elif vt.repo_type == 'git':

        stdout, stderr = Command(dest_dir).execute('git push -f --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)

    return stdout, stderr

import datetime

import io

import os

import tarfile

import tempfile

import zipfile

import pytest

from kallithea.lib.vcs.exceptions import VCSError

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

from kallithea.tests.vcs.conf import TESTS_TMP_PATH

class ArchivesTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

            yield {

                'message': 'Commit %d' % x,

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('%d/file_%d.txt' % (x, x),

                        content='Foobar %d' % x),

                ],

            }

    def test_archive_zip(self):

        path = tempfile.mkstemp(dir=TESTS_TMP_PATH, prefix='test_archive_zip-')[1]

        with open(path, 'wb') as f:

            self.tip.fill_archive(stream=f, kind='zip', prefix='repo')

        out = zipfile.ZipFile(path)

            node_path = '%d/file_%d.txt' % (x, x)

            decompressed = out.read('repo/' + node_path)

            assert decompressed == self.tip.get_node(node_path).content

    def test_archive_tgz(self):

        path = tempfile.mkstemp(dir=TESTS_TMP_PATH, prefix='test_archive_tgz-')[1]

        with open(path, 'wb') as f:

            self.tip.fill_archive(stream=f, kind='tgz', prefix='repo')

        outdir = tempfile.mkdtemp(dir=TESTS_TMP_PATH, prefix='test_archive_tgz-', suffix='-outdir')

        outfile = tarfile.open(path, 'r|gz')

        outfile.extractall(outdir)

            node_path = '%d/file_%d.txt' % (x, x)

            assert open(os.path.join(outdir, 'repo/' + node_path), 'rb').read() == self.tip.get_node(node_path).content

    def test_archive_tbz2(self):

        path = tempfile.mkstemp(dir=TESTS_TMP_PATH, prefix='test_archive_tbz2-')[1]

        with open(path, 'w+b') as f:

            self.tip.fill_archive(stream=f, kind='tbz2', prefix='repo')

        outdir = tempfile.mkdtemp(dir=TESTS_TMP_PATH, prefix='test_archive_tbz2-', suffix='-outdir')

        outfile = tarfile.open(path, 'r|bz2')

        outfile.extractall(outdir)

            node_path = '%d/file_%d.txt' % (x, x)

            assert open(os.path.join(outdir, 'repo/' + node_path), 'rb').read() == self.tip.get_node(node_path).content

    def test_archive_default_stream(self):

        tmppath = tempfile.mkstemp(dir=TESTS_TMP_PATH, prefix='test_archive_default_stream-')[1]

        with open(tmppath, 'wb') as stream:

            self.tip.fill_archive(stream=stream)

        mystream = io.BytesIO()

        self.tip.fill_archive(stream=mystream)

        mystream.seek(0)

        with open(tmppath, 'rb') as f:

            file_content = f.read()

            stringio_content = mystream.read()

            # the gzip header contains a MTIME header

            # because is takes a little bit of time from one fill_archive call to the next

            # this part may differ so don't include that part in the comparison

            assert file_content[:4] == stringio_content[:4]

            assert file_content[8:] == stringio_content[8:]

    def test_archive_wrong_kind(self):

        with pytest.raises(VCSError):

            self.tip.fill_archive(kind='wrong kind')

    def test_archive_empty_prefix(self):

        changeset.added = [FileNode('foo/bar/baz'), FileNode(u'foobar'), FileNode(u'blåbærgrød')]

        changeset.changed = []

        changeset.removed = []

        assert changeset.as_dict() == {

            'raw_id': 'RAW_ID',

            'short_id': 'SHORT_ID',

            'revision': 1009,

            'date': datetime.datetime(2011, 1, 30, 1, 45),

            'message': 'Message of a commit',

            'author': {

                'name': 'Joe Doe',

                'email': 'joe.doe@example.com',

            },

            'added': ['foo/bar/baz', 'foobar', u'bl\xe5b\xe6rgr\xf8d'],

            'changed': [],

            'removed': [],

        }

class _ChangesetsWithCommitsTestCaseixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

            yield {

                'message': 'Commit %d' % x,

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test_new_branch(self):

        self.imc.add(vcs.nodes.FileNode('docs/index.txt',

            content='Documentation\n'))

        foobar_tip = self.imc.commit(

            message=u'New branch: foobar',

            author=u'joe',

            branch='foobar',

        )

        assert 'foobar' in self.repo.branches

        assert foobar_tip.branch == 'foobar'

        assert foobar_tip.branches == ['foobar']

        # 'foobar' should be the only branch that contains the new commit

        branch_tips = list(self.repo.branches.values())

        assert branch_tips.count(str(foobar_tip.raw_id)) == 1

        for branch, sha in self.repo.branches.items():

            assert sha == self.repo.get_changeset(branch).raw_id

    def test_get_changeset_by_tag(self):

        for tag, sha in self.repo.tags.items():

            assert sha == self.repo.get_changeset(tag).raw_id

    def test_get_changeset_parents(self):

        for test_rev in [1, 2, 3]:

            sha = self.repo.get_changeset(test_rev-1)

            assert [sha] == self.repo.get_changeset(test_rev).parents

    def test_get_changeset_children(self):

        for test_rev in [1, 2, 3]:

            sha = self.repo.get_changeset(test_rev+1)

            assert [sha] == self.repo.get_changeset(test_rev).children

class _ChangesetsTestCaseMixin(_BackendTestMixin):

    recreate_repo_per_test = False

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

            yield {

                'message': u'Commit %d' % x,

                'author': u'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test_simple(self):

        tip = self.repo.get_changeset()

        assert tip.date == datetime.datetime(2010, 1, 3, 20)

    def test_get_changesets_is_ordered_by_date(self):

        changesets = list(self.repo.get_changesets())

        ordered_by_date = sorted(changesets,

            key=lambda cs: cs.date)

        assert changesets == ordered_by_date

    def test_get_changesets_respects_start(self):

        second_id = self.repo.revisions[1]

        changesets = list(self.repo.get_changesets(start=second_id))

        assert len(changesets) == 4

        for cs in self.repo.get_changesets(start_date=start_date):

            assert cs.date >= start_date

    def test_get_changesets_respects_end_date(self):

        start_date = datetime.datetime(2010, 1, 1)

        end_date = datetime.datetime(2010, 2, 1)

        for cs in self.repo.get_changesets(start_date=start_date,

                                           end_date=end_date):

            assert cs.date >= start_date

            assert cs.date <= end_date

    def test_get_changesets_respects_start_date_and_end_date(self):

        end_date = datetime.datetime(2010, 2, 1)

        for cs in self.repo.get_changesets(end_date=end_date):

            assert cs.date <= end_date

    def test_get_changesets_respects_reverse(self):

        changesets_id_list = [cs.raw_id for cs in

            self.repo.get_changesets(reverse=True)]

        assert changesets_id_list == list(reversed(self.repo.revisions))

    def test_get_filenodes_generator(self):

        tip = self.repo.get_changeset()

        filepaths = [node.path for node in tip.get_filenodes_generator()]

    def test_size(self):

        tip = self.repo.get_changeset()

        size = 5 * len('Foobar N') # Size of 5 files

        assert tip.size == size

    def test_author(self):

        tip = self.repo.get_changeset()

        assert tip.author == u'Joe Doe <joe.doe@example.com>'

    def test_author_name(self):

        tip = self.repo.get_changeset()

        assert tip.author_name == u'Joe Doe'

    def test_author_email(self):

        tip = self.repo.get_changeset()

        assert tip.author_email == u'joe.doe@example.com'

    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_start(self):

        with pytest.raises(ChangesetDoesNotExistError):

            list(self.repo.get_changesets(start='foobar'))

    def test_get_changesets_raise_changesetdoesnotexist_for_wrong_end(self):

        with pytest.raises(ChangesetDoesNotExistError):

import datetime

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

class GetitemTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

            yield {

                'message': 'Commit %d' % x,

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getitem__last_item_is_tip(self):

        assert self.repo[-1] == self.repo.get_changeset()

    def test__getitem__returns_correct_items(self):

        assert changesets == list(self.repo.get_changesets())

class TestGitGetitem(GetitemTestCaseMixin):

    backend_alias = 'git'

class TestHgGetitem(GetitemTestCaseMixin):

    backend_alias = 'hg'

import datetime

from kallithea.lib.vcs.nodes import FileNode

from kallithea.tests.vcs.base import _BackendTestMixin

class GetsliceTestCaseMixin(_BackendTestMixin):

    @classmethod

    def _get_commits(cls):

        start_date = datetime.datetime(2010, 1, 1, 20)

            yield {

                'message': 'Commit %d' % x,

                'author': 'Joe Doe <joe.doe@example.com>',

                'date': start_date + datetime.timedelta(hours=12 * x),

                'added': [

                    FileNode('file_%d.txt' % x, content='Foobar %d' % x),

                ],

            }

    def test__getslice__last_item_is_tip(self):

        assert list(self.repo[-1:])[0] == self.repo.get_changeset()

    def test__getslice__respects_start_index(self):

        assert list(self.repo[2:]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[2:]]

    def test__getslice__respects_negative_start_index(self):

        assert list(self.repo[-2:]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[-2:]]

    def test__getslice__respects_end_index(self):

        assert list(self.repo[:2]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[:2]]

    def test__getslice__respects_negative_end_index(self):

        assert list(self.repo[:-2]) == [self.repo.get_changeset(rev) for rev in self.repo.revisions[:-2]]

            self.imc.remove(node)

    def test_remove_raise_node_already_changed(self):

        self.test_add()  # Performs first commit

        node = FileNode(self.nodes[0].path, content='Bending time')

        self.imc.change(node)

        with pytest.raises(NodeAlreadyChangedError):

            self.imc.remove(node)

    def test_reset(self):

        self.imc.add(FileNode('foo', content='bar'))

        #self.imc.change(FileNode('baz', content='new'))

        #self.imc.remove(FileNode('qwe'))

        self.imc.reset()

        assert not any((

            self.imc.added,

            self.imc.changed,

            self.imc.removed

        ))

    def test_multiple_commits(self):

        N = 3  # number of commits to perform

        last = None

            fname = 'file%s' % str(x).rjust(5, '0')

            content = 'foobar\n' * x

            node = FileNode(fname, content=content)

            self.imc.add(node)

            commit = self.imc.commit(u"Commit no. %s" % (x + 1), author=u'vcs')

            assert last != commit

            last = commit

        # Check commit number for same repo

        assert len(self.repo.revisions) == N

        # Check commit number for recreated repo

        assert len(self.repo.revisions) == N

    def test_date_attr(self):

        node = FileNode('foobar.txt', content='Foobared!')

        self.imc.add(node)

        date = datetime.datetime(1985, 1, 30, 1, 45)

        commit = self.imc.commit(u"Committed at time when I was born ;-)",

            author=u'lb <lb@example.com>', date=date)

        assert commit.date == date