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

import os

import mock

import urllib

from kallithea.lib import vcs

from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User, \

    Permission

from kallithea.model.user import UserModel

from kallithea.tests import *

from kallithea.model.repo_group import RepoGroupModel

from kallithea.model.repo import RepoModel

from kallithea.model.meta import Session

from kallithea.tests.fixture import Fixture, error_function

fixture = Fixture()

            .filter(Repository.repo_name == repo_name).one()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            self.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name)

        Session().commit()

    def test_create_non_ascii(self):

        self.log_user()

        non_ascii = "ąęł"

        repo_name = "%s%s" % (self.NEW_REPO, non_ascii)

        repo_name_unicode = repo_name.decode('utf8')

        description = 'description for newly created repo' + non_ascii

            .filter(Repository.repo_name == repo_name_unicode).one()

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            self.fail('no repo %s in filesystem' % repo_name)

    def test_create_in_group(self):

        self.log_user()

        ## create GROUP

        group_name = 'sometest_%s' % self.REPO_TYPE

        gr = RepoGroupModel().create(group_name=group_name,

                                     group_description='test',

                                     owner=TEST_USER_ADMIN_LOGIN)

        Session().commit()

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            self.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        Session().commit()

    def test_create_in_group_without_needed_permissions(self):

        usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)

        # avoid spurious RepoGroup DetachedInstanceError ...

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 1)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            self.fail('no repo %s in filesystem' % repo_name)

        RepoModel().delete(repo_name_full)

        RepoGroupModel().delete(group_name)

        RepoGroupModel().delete(group_name_allowed)

        Session().commit()

    def test_create_in_group_inherit_permissions(self):

        self.log_user()

        new_repo_id = new_repo.repo_id

        self.assertEqual(new_repo.repo_name, repo_name_full)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name_full))

        response.mustcontain(repo_name_full)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            RepoGroupModel().delete(group_name)

            Session().commit()

            self.fail('no repo %s in filesystem' % repo_name)

        #check if inherited permissiona are applied

        inherited_perms = UserRepoToPerm.query() \

            .filter(UserRepoToPerm.repository_id == new_repo_id).all()

        self.assertEqual(len(inherited_perms), 2)

        self.assertTrue(TEST_USER_REGULAR_LOGIN in [x.user.username

                                                    for x in inherited_perms])

            .filter(Repository.repo_name == repo_name).one()

        self.assertEqual(new_repo.repo_name, repo_name)

        self.assertEqual(new_repo.description, description)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            self.fail('no repo %s in filesystem' % repo_name)

        response = self.app.post(url('delete_repo', repo_name=repo_name),

            params={'_method': 'delete', '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))

        response.follow()

        #check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_unicode).one()

        self.assertEqual(new_repo.repo_name, repo_name_unicode)

        self.assertEqual(new_repo.description, description_unicode)

        # test if the repository is visible in the list ?

        response = self.app.get(url('summary_home', repo_name=repo_name))

        response.mustcontain(repo_name)

        response.mustcontain(self.REPO_TYPE)

        # test if the repository was created on filesystem

        try:

        except vcs.exceptions.VCSError:

            self.fail('no repo %s in filesystem' % repo_name)

        response = self.app.post(url('delete_repo', repo_name=repo_name),

            params={'_method': 'delete', '_authentication_token': self.authentication_token()})

        self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))

        response.follow()

        #check if repo was deleted from db

        deleted_repo = Session().query(Repository) \

            .filter(Repository.repo_name == repo_name_unicode).scalar()

import os

from kallithea.lib.vcs.backends.hg import MercurialRepository, MercurialChangeset

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError, NodeDoesNotExistError

from kallithea.lib.vcs.nodes import NodeKind, NodeState

from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_HG_REPO_CLONE, \

    TEST_HG_REPO_PULL

from kallithea.lib.vcs.utils.compat import unittest

# Use only clean mercurial's ui

from kallithea.lib.vcs.utils.hgcompat import mercurial

mercurial.scmutil.rcpath()

if mercurial.scmutil._rcpath:

    mercurial.scmutil._rcpath = mercurial.scmutil._rcpath[:1]

class MercurialRepositoryTest(unittest.TestCase):

    def __check_for_existing_repo(self):

        if os.path.exists(TEST_HG_REPO_CLONE):

            self.fail('Cannot test mercurial clone repo as location %s already '

                      'exists. You should manually remove it first.'

                      % TEST_HG_REPO_CLONE)

    def setUp(self):

    def test_wrong_repo_path(self):

        wrong_repo_path = '/tmp/errorrepo'

        self.assertRaises(RepositoryError, MercurialRepository, wrong_repo_path)

    def test_unicode_path_repo(self):

        self.assertRaises(VCSError,lambda:MercurialRepository(u'iShouldFail'))

    def test_repo_clone(self):

        self.__check_for_existing_repo()

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE,

            src_url=TEST_HG_REPO, update_after_clone=True)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        # Checking hashes of changesets should be enough

        for changeset in repo.get_changesets():

            raw_id = changeset.raw_id

            self.assertEqual(raw_id, repo_clone.get_changeset(raw_id).raw_id)

    def test_repo_clone_with_update(self):

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_w_update',

            src_url=TEST_HG_REPO, update_after_clone=True)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        #check if current workdir was updated

        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \

                                                    + '_w_update',

                                                    'MANIFEST.in')), True,)

    def test_repo_clone_without_update(self):

        repo_clone = MercurialRepository(TEST_HG_REPO_CLONE + '_wo_update',

            src_url=TEST_HG_REPO, update_after_clone=False)

        self.assertEqual(len(repo.revisions), len(repo_clone.revisions))

        self.assertEqual(os.path.isfile(os.path.join(TEST_HG_REPO_CLONE \

                                                    + '_wo_update',

                                                    'MANIFEST.in')), False,)

    def test_pull(self):

        if os.path.exists(TEST_HG_REPO_PULL):

            self.fail('Cannot test mercurial pull command as location %s '

                      'already exists. You should manually remove it first'

                      % TEST_HG_REPO_PULL)

TODO: To be written...

"""

        node = chset10.get_node('README.rst')

        self.assertEqual(node.kind, NodeKind.FILE)

        self.assertEqual(node.content, README)

class MercurialChangesetTest(unittest.TestCase):

    def setUp(self):

    def _test_equality(self, changeset):

        revision = changeset.revision

        self.assertEqual(changeset, self.repo.get_changeset(revision))

    def test_equality(self):

        self.setUp()

        revs = [0, 10, 20]

        changesets = [self.repo.get_changeset(rev) for rev in revs]

        for changeset in changesets:

            self._test_equality(changeset)

import os

import shutil

from kallithea.lib.vcs import VCSError, get_repo, get_backend

from kallithea.lib.vcs.backends.hg import MercurialRepository

from kallithea.lib.vcs.utils.compat import unittest

from kallithea.tests.vcs.conf import TEST_HG_REPO, TEST_GIT_REPO, TEST_TMP_PATH

class VCSTest(unittest.TestCase):

    """

    Tests for main module's methods.

    """

    def test_get_backend(self):

        hg = get_backend('hg')

        self.assertEqual(hg, MercurialRepository)

    def test_alias_detect_hg(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

        self.assertEqual('hg',repo.alias)

    def test_alias_detect_git(self):

        alias = 'git'

        path = TEST_GIT_REPO

        backend = get_backend(alias)

        self.assertEqual('git',repo.alias)

    def test_wrong_alias(self):

        alias = 'wrong_alias'

        self.assertRaises(VCSError, get_backend, alias)

    def test_get_repo(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

    def test_get_repo_autoalias_hg(self):

        alias = 'hg'

        path = TEST_HG_REPO

        backend = get_backend(alias)

    def test_get_repo_autoalias_git(self):

        alias = 'git'

        path = TEST_GIT_REPO

        backend = get_backend(alias)

    def test_get_repo_err(self):

        blank_repo_path = os.path.join(TEST_TMP_PATH, 'blank-error-repo')

        if os.path.isdir(blank_repo_path):

            shutil.rmtree(blank_repo_path)

        os.mkdir(blank_repo_path)

        self.assertRaises(VCSError, get_repo, blank_repo_path)

        self.assertRaises(VCSError, get_repo, blank_repo_path + 'non_existing')

    def test_get_repo_multialias(self):