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

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

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

        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)

        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)

        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

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

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

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

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

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

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

        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)