clone_url = vt.repo_url_param(webserver, vt.repo_name, password='bad!')

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'fatal: Authentication failed' in stderr

        elif vt.repo_type == 'hg':

            assert 'abort: authorization failed' in stderr

    def test_clone_git_dir_as_hg(self, webserver):

        clone_url = HgHttpVcsTest.repo_url_param(webserver, base.GIT_REPO)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'HTTP Error 404: Not Found' in stderr or "not a valid repository" in stdout and 'abort:' in stderr

    def test_clone_hg_repo_as_git(self, webserver):

        clone_url = GitHttpVcsTest.repo_url_param(webserver, base.HG_REPO)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'not found' in stderr

    @parametrize_vcs_test

    def test_clone_non_existing_path(self, webserver, vt):

        clone_url = vt.repo_url_param(webserver, 'trololo')

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'not found' in stderr or 'abort: Access to %r denied' % 'trololo' in stderr

        elif vt.repo_type == 'hg':

    @parametrize_vcs_test

    def test_push_new_repo(self, webserver, vt):

        # Clear the log so we know what is added

        UserLog.query().delete()

        Session().commit()

        # Create an empty server repo using the API

        repo_name = 'new_%s_%s' % (vt.repo_type, next(_RandomNameSequence()))

        usr = User.get_by_username(base.TEST_USER_ADMIN_LOGIN)

        params = {

            "id": 7,

            "api_key": usr.api_key,

            "method": 'create_repo',

            "args": dict(repo_name=repo_name,

                         owner=base.TEST_USER_ADMIN_LOGIN,

                         repo_type=vt.repo_type),

        }

        req = urllib.request.Request(

            'http://%s:%s/_admin/api' % webserver.server_address,

            data=ascii_bytes(json.dumps(params)),

            headers={'content-type': 'application/json'})

        response = urllib.request.urlopen(req)

        result = json.loads(response.read())

        Session.close()  # make sure SA fetches all new log entries (apparently only needed for MariaDB/MySQL ...)

        if vt.repo_type == 'git':

            assert 'FETCH_HEAD' in stderr

        elif vt.repo_type == 'hg':

            assert 'new changesets' in stdout

        action_parts = [ul.action for ul in UserLog.query().order_by(UserLog.user_log_id)]

        assert action_parts == ['pull']

        # Test handling of URLs with extra '/' around repo_name

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/./%s/' % vt.repo_name), ignoreReturnCode=True)

        if issubclass(vt, HttpVcsTest):

            if vt.repo_type == 'git':

                # NOTE: when pulling from http://hostname/./vcs_test_git/ , the git client will normalize that and issue an HTTP request to /vcs_test_git/info/refs

                assert 'Already up to date.' in stdout

            else:

                assert vt.repo_type == 'hg'

                assert "abort: HTTP Error 404: Not Found" in stderr

        else:

            assert issubclass(vt, SshVcsTest)

            if vt.repo_type == 'git':

                assert "abort: Access to './%s' denied" % vt.repo_name in stderr

            else:

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/%s/' % vt.repo_name), ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'Already up to date.' in stdout

        else:

            assert vt.repo_type == 'hg'

            assert "no changes found" in stdout

        assert "denied" not in stderr

        assert "denied" not in stdout

        assert "404" not in stdout

    @parametrize_vcs_test

    def test_push_invalidates_cache(self, webserver, testfork, vt):

        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)

        if vt.repo_type == 'git':

            _check_proper_git_push(stdout, stderr)

            webserver, vt, dest_dir, clone_url='http://%s:%s/tmp' % (

                webserver.server_address[0], webserver.server_address[1]),

            ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'not found' in stderr

        elif vt.repo_type == 'hg':

            assert 'HTTP Error 404: Not Found' in stderr

    @parametrize_vcs_test

    def test_ip_restriction(self, webserver, vt):

        user_model = UserModel()

        try:

            # Add IP constraint that excludes the test context:

            user_model.add_extra_ip(base.TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

            clone_url = vt.repo_url_param(webserver, vt.repo_name)

            stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

            if vt.repo_type == 'git':

                # The message apparently changed in Git 1.8.3, so match it loosely.

                assert re.search(r'\b403\b', stderr) or 'abort: User test_admin from 127.0.0.127 cannot be authorized' in stderr

            elif vt.repo_type == 'hg':

        finally:

            # release IP restrictions

            for ip in UserIpMap.query():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

            # IP permissions are cached, need to wait for the cache in the server process to expire

            time.sleep(1.5)

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, _get_tmp_dir())

        if vt.repo_type == 'git':

            assert 'Cloning into' in stdout + stderr

            assert stderr == '' or stdout == ''

        elif vt.repo_type == 'hg':

            assert 'requesting all changes' in stdout

            assert 'adding changesets' in stdout

            assert 'adding manifests' in stdout

            assert 'adding file changes' in stdout

            assert stderr == ''

    @parametrize_vcs_test_hg # git hooks doesn't work like hg hooks

    def test_custom_hooks_preoutgoing(self, testhook_cleanup, webserver, testfork, vt):

requirements = [

    "alembic >= 1.0.10, < 1.5",

    "gearbox >= 0.1.0, < 1",

    "waitress >= 0.8.8, < 1.5",

    "WebOb >= 1.8, < 1.9",

    "backlash >= 0.1.2, < 1",

    "TurboGears2 >= 2.4, < 2.5",

    "tgext.routes >= 0.2.0, < 1",

    "Beaker >= 1.10.1, < 2",

    "WebHelpers2 >= 2.0, < 2.1",

    "FormEncode >= 1.3.1, < 1.4",

    "SQLAlchemy >= 1.2.9, < 1.4",

    "Mako >= 0.9.1, < 1.2",

    "Pygments >= 2.2.0, < 2.7",

    "Whoosh >= 2.7.1, < 2.8",

    "celery >= 4.3, < 4.5, != 4.4.4", # 4.4.4 is broken due to unexpressed dependency on 'future', see https://github.com/celery/celery/pull/6146

    "Babel >= 1.3, < 2.9",

    "python-dateutil >= 2.1.0, < 2.9",

    "Markdown >= 2.2.1, < 3.2",

    "docutils >= 0.11, < 0.17",

    "URLObject >= 2.3.4, < 2.5",

    "Routes >= 2.0, < 2.5",

    "dulwich >= 0.19.0, < 0.20",

    "decorator >= 4.2.1, < 4.5",

    "Paste >= 2.0.3, < 3.5",

    "bleach >= 3.0, < 3.1.4",

    "Click >= 7.0, < 8",

    "ipaddr >= 2.2.0, < 2.3",

    "paginate >= 0.5, < 0.6",

    "paginate_sqlalchemy >= 0.3.0, < 0.4",

    "bcrypt >= 3.1.0, < 3.2",

    "pip >= 20.0, < 999",

]

dependency_links = [

]

classifiers = [

    'Development Status :: 4 - Beta',

    'Environment :: Web Environment',

    'Framework :: Pylons',

    'Intended Audience :: Developers',

    'License :: OSI Approved :: GNU General Public License (GPL)',

    'Operating System :: OS Independent',

    'Programming Language :: Python :: 3.6',

    'Programming Language :: Python :: 3.7',

    'Programming Language :: Python :: 3.8',