fix(tests): fixed svn tests cases when interactive prompt was displayed

rhodecode/tests/vcs_operations/__init__.py

@@ -1,214 +1,220 b''
 
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Base for test suite for making push/pull operations.
 
 .. important::
 
    You must have git >= 1.8.5 for tests to work fine. With 68b939b git started
    to redirect things to stderr instead of stdout.
 """
 
 
 import logging
 import os
 import tempfile
 import subprocess
 
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.tests import GIT_REPO, HG_REPO, SVN_REPO
 
 DEBUG = True
 RC_LOG = os.path.join(tempfile.gettempdir(), 'rc.log')
 REPO_GROUP = 'a_repo_group'
 HG_REPO_WITH_GROUP = f'{REPO_GROUP}/{HG_REPO}'
 GIT_REPO_WITH_GROUP = f'{REPO_GROUP}/{GIT_REPO}'
 SVN_REPO_WITH_GROUP = f'{REPO_GROUP}/{SVN_REPO}'
 
 log = logging.getLogger(__name__)
 
 
 class Command(object):
 
     def __init__(self, cwd):
         self.cwd = cwd
         self.process = None
 
     def execute(self, cmd, *args):
         """
         Runs command on the system with given ``args``.
         """
 
         command = cmd + ' ' + ' '.join(args)
         if DEBUG:
             log.debug('*** CMD %s ***', command)
 
         env = dict(os.environ)
         # Delete coverage variables, as they make the test fail for Mercurial
         for key in env.keys():
             if key.startswith('COV_CORE_'):
                 del env[key]
 
         self.process = subprocess.Popen(
             command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
             cwd=self.cwd, env=env)
         stdout, stderr = self.process.communicate()
 
         stdout = safe_str(stdout)
         stderr = safe_str(stderr)
 
         if DEBUG:
             log.debug('STDOUT:%s', stdout)
             log.debug('STDERR:%s', stderr)
         return stdout, stderr
 
     def assert_returncode_success(self):
         assert self.process.returncode == 0
 
 
 def _add_files(vcs, dest, clone_url=None, tags=None, target_branch=None, new_branch=False, **kwargs):
     full_name = 'Marcin Kuźminski'
     email = '[e-mail]'
     git_ident = f"git config user.name {full_name} && git config user.email {email}"
     cwd = path = os.path.join(dest)
 
     tags = tags or []
     name_sequence = next(tempfile._RandomNameSequence())
     added_file = os.path.join(path, f'{name_sequence}_setup.py')
 
     Command(cwd).execute(f'touch {added_file}')
     Command(cwd).execute(f'{vcs} add {added_file}')
     author_str = 'Marcin Kuźminski <[e-mail]>'
 
     for i in range(kwargs.get('files_no', 3)):
         cmd = f"""echo 'added_line{i}' >> {added_file}"""
         Command(cwd).execute(cmd)
 
         if vcs == 'hg':
             cmd = f"""hg commit -m 'committed new {i}' -u '{author_str}' {added_file} """
         elif vcs == 'git':
             cmd = f"""{git_ident} && git commit -m 'committed new {i}' {added_file}"""
         Command(cwd).execute(cmd)
 
     for tag in tags:
         if vcs == 'hg':
             Command(cwd).execute(
                 f"""hg tag -m "{tag['commit']}" -u "{author_str}" """,
                 tag['name'])
         elif vcs == 'git':
             if tag['commit']:
                 # annotated tag
                 _stdout, _stderr = Command(cwd).execute(
                     f"""{git_ident} && git tag -a {tag['name']} -m "{tag['commit']}" """
                 )
             else:
                 # lightweight tag
                 _stdout, _stderr = Command(cwd).execute(
                     f"""{git_ident} && git tag {tag['name']}"""
                 )
 
 
 def _add_files_and_push(vcs, dest, clone_url=None, tags=None, target_branch=None,
                         new_branch=False, **kwargs):
     """
     Generate some files, add it to DEST repo and push back
     vcs is git or hg and defines what VCS we want to make those files for
     """
     git_ident = "git config user.name Marcin Kuźminski && git config user.email [e-mail]"
     cwd = os.path.join(dest)
 
     # commit some stuff into this repo
     _add_files(vcs, dest, clone_url, tags, target_branch, new_branch, **kwargs)
 
     default_target_branch = {
         'git': 'master',
         'hg': 'default'
     }.get(vcs)
 
     target_branch = target_branch or default_target_branch
 
     # PUSH it back
     stdout = stderr = None
     if vcs == 'hg':
         maybe_new_branch = ''
         if new_branch:
             maybe_new_branch = '--new-branch'
         stdout, stderr = Command(cwd).execute(
             f'hg push --traceback --verbose {maybe_new_branch} -r {target_branch} {clone_url}'
         )
     elif vcs == 'git':
         stdout, stderr = Command(cwd).execute(
             f'{git_ident} && git push --verbose --tags {clone_url} {target_branch}'
         )
     elif vcs == 'svn':
+        username = kwargs.pop('username', '')
+        password = kwargs.pop('password', '')
+        auth = ''
+        if username and password:
+            auth = f'--username {username} --password {password}'
+
         stdout, stderr = Command(cwd).execute(
-            f'svn ci -m "pushing to {target_branch}"'
+            f'svn commit --no-auth-cache --non-interactive {auth} -m "pushing to {target_branch}"'
         )
 
     return stdout, stderr
 
 
 def _check_proper_git_push(
         stdout, stderr, branch='master', should_set_default_branch=False):
     # Note: Git is writing most information to stderr intentionally
     assert 'fatal' not in stderr
     assert 'rejected' not in stderr
     assert 'Pushing to' in stderr
     assert '%s -> %s' % (branch, branch) in stderr
 
     if should_set_default_branch:
         assert "Setting default branch to %s" % branch in stderr
     else:
         assert "Setting default branch" not in stderr
 
 
 def _check_proper_hg_push(stdout, stderr, branch='default'):
     assert 'pushing to' in stdout
     assert 'searching for changes' in stdout
 
     assert 'abort:' not in stderr
 
 
 def _check_proper_svn_push(stdout, stderr):
     assert 'pushing to' in stdout
     assert 'searching for changes' in stdout
 
     assert 'abort:' not in stderr
 
 
 def _check_proper_clone(stdout, stderr, vcs):
     if vcs == '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 == ''
 
     if vcs == 'git':
         assert '' == stdout
         assert 'Cloning into' in stderr
         assert 'abort:' not in stderr
         assert 'fatal:' not in stderr
 
     if vcs == 'svn':
         assert 'dupa' in stdout
 
 

rhodecode/tests/vcs_operations/test_vcs_operations_svn.py

@@ -1,197 +1,224 b''
 # Copyright (C) 2010-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Test suite for making push/pull operations, on specially modified INI files
 
 .. important::
 
    You must have git >= 1.8.5 for tests to work fine. With 68b939b git started
    to redirect things to stderr instead of stdout.
 """
 
 
 import time
 import pytest
 
 from rhodecode.model.db import Repository, UserIpMap
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (SVN_REPO, TEST_USER_ADMIN_LOGIN)
 
 
 from rhodecode.tests.vcs_operations import (
     Command, _check_proper_clone, _check_proper_svn_push,
     _add_files_and_push, SVN_REPO_WITH_GROUP)
 
 
+def get_cli_flags(username, password):
+    flags = '--no-auth-cache --non-interactive'
+    auth = ''
+    if username and password:
+        auth = f'--username {username} --password {password}'
+    return flags, auth
+
+
 @pytest.mark.usefixtures("disable_locking", "disable_anonymous_user")
 class TestVCSOperations(object):
 
     def test_clone_svn_repo_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO)
         username, password = rc_web_server.repo_clone_credentials()
 
         cmd = Command('/tmp')
 
-        auth = f'--non-interactive --username={username} --password={password}'
-        stdout, stderr = cmd.execute(f'svn checkout {auth}', clone_url, tmpdir.strpath)
+        flags, auth = get_cli_flags(username, password)
+
+        stdout, stderr = Command('/tmp').execute(
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
+
         _check_proper_clone(stdout, stderr, 'svn')
         cmd.assert_returncode_success()
 
     def test_clone_svn_repo_by_id_by_admin(self, rc_web_server, tmpdir):
         repo_id = Repository.get_by_repo_name(SVN_REPO).repo_id
         username, password = rc_web_server.repo_clone_credentials()
 
         clone_url = rc_web_server.repo_clone_url('_%s' % repo_id)
         cmd = Command('/tmp')
-        auth = f'--non-interactive --username={username} --password={password}'
-        stdout, stderr = cmd.execute(f'svn checkout {auth}', clone_url, tmpdir.strpath)
+
+        flags, auth = get_cli_flags(username, password)
+
+        stdout, stderr = Command('/tmp').execute(
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
+
         _check_proper_clone(stdout, stderr, 'svn')
         cmd.assert_returncode_success()
 
     def test_clone_svn_repo_with_group_by_admin(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO_WITH_GROUP)
         username, password = rc_web_server.repo_clone_credentials()
 
-        cmd = Command('/tmp')
-        auth = f'--non-interactive --username={username} --password={password}'
-        stdout, stderr = cmd.execute(f'svn checkout {auth}', clone_url, tmpdir.strpath)
+        flags, auth = get_cli_flags(username, password)
+
+        stdout, stderr = Command('/tmp').execute(
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
+
         _check_proper_clone(stdout, stderr, 'svn')
         cmd.assert_returncode_success()
 
     def test_clone_wrong_credentials_svn(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO)
         username, password = rc_web_server.repo_clone_credentials()
         password = 'bad-password'
 
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
+
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
         assert 'fatal: Authentication failed' in stderr
 
     def test_clone_svn_with_slashes(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url('//' + SVN_REPO)
-        stdout, stderr = Command('/tmp').execute('svn checkout', clone_url)
+        username, password = '', ''
+        flags, auth = get_cli_flags(username, password)
+
+        stdout, stderr = Command('/tmp').execute(
+            f'svn checkout {flags} {auth}', clone_url)
+
         assert 'not found' in stderr
 
     def test_clone_existing_path_svn_not_in_database(
             self, rc_web_server, tmpdir, fs_repo_only):
         db_name = fs_repo_only('not-in-db-git', repo_type='git')
         clone_url = rc_web_server.repo_clone_url(db_name)
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
 
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_clone_existing_path_svn_not_in_database_different_scm(
             self, rc_web_server, tmpdir, fs_repo_only):
         db_name = fs_repo_only('not-in-db-hg', repo_type='hg')
         clone_url = rc_web_server.repo_clone_url(db_name)
 
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
+
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_clone_non_existing_store_path_svn(self, rc_web_server, tmpdir, user_util):
         repo = user_util.create_repo(repo_type='git')
         clone_url = rc_web_server.repo_clone_url(repo.repo_name)
 
         # Damage repo by removing it's folder
         RepoModel()._delete_filesystem_repo(repo)
 
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
+
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
         assert 'not found' in stderr
 
     def test_push_new_file_svn(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO)
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
 
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
 
         # commit some stuff into this repo
         stdout, stderr = _add_files_and_push(
-            'svn', tmpdir.strpath, clone_url=clone_url)
+            'svn', tmpdir.strpath, clone_url=clone_url, username=username, password=password)
 
         _check_proper_svn_push(stdout, stderr)
 
     def test_push_wrong_credentials_svn(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO)
 
-        username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        username, password = rc_web_server.repo_clone_credentials()
+        flags, auth = get_cli_flags(username, password)
+
         stdout, stderr = Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
 
         push_url = rc_web_server.repo_clone_url(
             SVN_REPO, user='bad', passwd='name')
         stdout, stderr = _add_files_and_push(
-            'svn', tmpdir.strpath, clone_url=push_url)
+            'svn', tmpdir.strpath, clone_url=push_url, username=username, password=password)
 
         assert 'fatal: Authentication failed' in stderr
 
     def test_push_back_to_wrong_url_svn(self, rc_web_server, tmpdir):
         clone_url = rc_web_server.repo_clone_url(SVN_REPO)
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
-        Command('/tmp').execute(
-            f'svn checkout {auth}', clone_url, tmpdir.strpath)
+        flags, auth = get_cli_flags(username, password)
+
+        stdout, stderr = Command('/tmp').execute(
+            f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
 
         stdout, stderr = _add_files_and_push(
             'svn', tmpdir.strpath,
-            clone_url=rc_web_server.repo_clone_url('not-existing'))
+            clone_url=rc_web_server.repo_clone_url('not-existing'), username=username, password=password)
 
         assert 'not found' in stderr
 
     def test_ip_restriction_svn(self, rc_web_server, tmpdir):
         user_model = UserModel()
         username, password = '', ''
-        auth = f'--non-interactive --username={username} --password={password}'
+        flags, auth = get_cli_flags(username, password)
 
         try:
             user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')
             Session().commit()
             time.sleep(2)
             clone_url = rc_web_server.repo_clone_url(SVN_REPO)
 
             stdout, stderr = Command('/tmp').execute(
-                f'svn checkout {auth}', clone_url, tmpdir.strpath)
+                f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
             msg = "The requested URL returned error: 403"
             assert msg in stderr
         finally:
             # release IP restrictions
             for ip in UserIpMap.getAll():
                 UserIpMap.delete(ip.ip_id)
             Session().commit()
 
         time.sleep(2)
 
         cmd = Command('/tmp')
-        stdout, stderr = cmd.execute(f'svn checkout {auth}', clone_url, tmpdir.strpath)
+        stdout, stderr = cmd.execute(f'svn checkout {flags} {auth}', clone_url, tmpdir.strpath)
         cmd.assert_returncode_success()
         _check_proper_clone(stdout, stderr, 'svn')