tests: fixed tests for random dict order could sometimes break the tests

rhodecode/apps/ssh_support/tests/test_server_git.py

@@ -1,195 +1,196 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 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/
 
 import json
 
 import pytest
 from mock import Mock, patch, call
 
 from rhodecode.apps.ssh_support.lib.ssh_wrapper import GitServer
 
 
 @pytest.fixture
 def git_server():
     return GitServerCreator()
 
 
 class GitServerCreator(object):
     root = '/tmp/repo/path/'
     git_path = '/usr/local/bin/'
     config_data = {
         'app:main': {
             'ssh.executable.git': git_path
         }
     }
     repo_name = 'test_git'
     repo_mode = 'receive-pack'
     user = 'vcs'
 
     def __init__(self):
         def config_get(part, key):
             return self.config_data.get(part, {}).get(key)
         self.config_mock = Mock()
         self.config_mock.get = Mock(side_effect=config_get)
 
     def create(self, **kwargs):
         parameters = {
             'store': {'path': self.root},
             'ini_path': '',
             'user': self.user,
             'repo_name': self.repo_name,
             'repo_mode': self.repo_mode,
             'user_permissions': {
                 self.repo_name: 'repo_admin'
             },
             'config': self.config_mock,
         }
         parameters.update(kwargs)
         server = GitServer(**parameters)
         return server
 
 
 class TestGitServer(object):
     def test_command(self, git_server):
         server = git_server.create()
         server.read_only = False
         expected_command = (
             'cd {root}; {git_path}-{repo_mode}'
             ' \'{root}{repo_name}\''.format(
                 root=git_server.root, git_path=git_server.git_path,
                 repo_mode=git_server.repo_mode, repo_name=git_server.repo_name)
         )
         assert expected_command == server.command
 
     def test_run_returns_exit_code_2_when_no_permissions(self, git_server, caplog):
         server = git_server.create()
         with patch.object(server, '_check_permissions') as permissions_mock:
             with patch.object(server, '_update_environment'):
                 permissions_mock.return_value = 2
                 exit_code = server.run()
 
         assert exit_code == (2, False)
 
     def test_run_returns_executes_command(self, git_server, caplog):
         server = git_server.create()
         with patch.object(server, '_check_permissions') as permissions_mock:
             with patch('os.system') as system_mock:
                 with patch.object(server, '_update_environment') as (
                         update_mock):
                     permissions_mock.return_value = 0
                     system_mock.return_value = 0
                     exit_code = server.run()
 
         system_mock.assert_called_once_with(server.command)
         update_mock.assert_called_once_with()
 
         assert exit_code == (0, True)
 
     @pytest.mark.parametrize(
         'repo_mode, action', [
             ['receive-pack', 'push'],
             ['upload-pack', 'pull']
         ])
     def test_update_environment(self, git_server, repo_mode, action):
         server = git_server.create(repo_mode=repo_mode)
         with patch('os.environ', {'SSH_CLIENT': '10.10.10.10 b'}):
             with patch('os.putenv') as putenv_mock:
                 server._update_environment()
+
         expected_data = {
             "username": git_server.user,
             "scm": "git",
             "repository": git_server.repo_name,
             "make_lock": None,
             "action": [action],
             "ip": "10.10.10.10",
             "locked_by": [None, None],
             "config": ""
         }
-        putenv_mock.assert_called_once_with(
-            'RC_SCM_DATA', json.dumps(expected_data))
+        args, kwargs = putenv_mock.call_args
+        assert json.loads(args[1]) == expected_data
 
 
 class TestGitServerCheckPermissions(object):
     def test_returns_2_when_no_permissions_found(self, git_server, caplog):
         user_permissions = {}
         server = git_server.create(user_permissions=user_permissions)
         result = server._check_permissions()
         assert result == 2
 
         log_msg = 'permission for vcs on test_git are: None'
         assert log_msg in [t[2] for t in caplog.record_tuples]
 
     def test_returns_2_when_no_permissions(self, git_server, caplog):
         user_permissions = {git_server.repo_name: 'repository.none'}
         server = git_server.create(user_permissions=user_permissions)
         result = server._check_permissions()
         assert result == 2
 
         log_msg = 'repo not found or no permissions'
         assert log_msg in [t[2] for t in caplog.record_tuples]
 
     @pytest.mark.parametrize(
         'permission', ['repository.admin', 'repository.write'])
     def test_access_allowed_when_user_has_write_permissions(
             self, git_server, permission, caplog):
         user_permissions = {git_server.repo_name: permission}
         server = git_server.create(user_permissions=user_permissions)
         result = server._check_permissions()
         assert result is None
 
         log_msg = 'Write Permissions for User "%s" granted to repo "%s"!' % (
             git_server.user, git_server.repo_name)
         assert log_msg in [t[2] for t in caplog.record_tuples]
 
     def test_write_access_is_not_allowed_when_user_has_read_permission(
             self, git_server, caplog):
         user_permissions = {git_server.repo_name: 'repository.read'}
         server = git_server.create(
             user_permissions=user_permissions, repo_mode='receive-pack')
         result = server._check_permissions()
         assert result == -3
 
         log_msg = 'Only Read Only access for User "%s" granted to repo "%s"! Failing!' % (
             git_server.user, git_server.repo_name)
         assert log_msg in [t[2] for t in caplog.record_tuples]
 
     def test_read_access_allowed_when_user_has_read_permission(
             self, git_server, caplog):
         user_permissions = {git_server.repo_name: 'repository.read'}
         server = git_server.create(
             user_permissions=user_permissions, repo_mode='upload-pack')
         result = server._check_permissions()
         assert result is None
 
         log_msg = 'Only Read Only access for User "%s" granted to repo "%s"!' % (
             git_server.user, git_server.repo_name)
         assert log_msg in [t[2] for t in caplog.record_tuples]
 
     def test_returns_error_when_permission_not_recognised(
             self, git_server, caplog):
         user_permissions = {git_server.repo_name: 'repository.whatever'}
         server = git_server.create(
             user_permissions=user_permissions, repo_mode='upload-pack')
         result = server._check_permissions()
         assert result == -2
 
         log_msg = 'Cannot properly fetch user permission. ' \
                   'Return value is: repository.whatever'
         assert log_msg in [t[2] for t in caplog.record_tuples]
  No newline at end of file

rhodecode/tests/lib/middleware/test_simplehg.py

@@ -1,128 +1,128 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 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/
 
 import urlparse
 
 import mock
 import pytest
 import simplejson as json
 
 from rhodecode.lib.vcs.backends.base import Config
 from rhodecode.tests.lib.middleware import mock_scm_app
 import rhodecode.lib.middleware.simplehg as simplehg
 
 
 def get_environ(url):
     """Construct a minimum WSGI environ based on the URL."""
     parsed_url = urlparse.urlparse(url)
     environ = {
         'PATH_INFO': parsed_url.path,
         'QUERY_STRING': parsed_url.query,
     }
 
     return environ
 
 
 @pytest.mark.parametrize(
     'url, expected_action',
     [
         ('/foo/bar?cmd=unbundle&key=tip', 'push'),
         ('/foo/bar?cmd=pushkey&key=tip', 'push'),
         ('/foo/bar?cmd=listkeys&key=tip', 'pull'),
         ('/foo/bar?cmd=changegroup&key=tip', 'pull'),
         # Edge case: unknown argument: assume pull
         ('/foo/bar?cmd=unknown&key=tip', 'pull'),
         ('/foo/bar?cmd=&key=tip', 'pull'),
         # Edge case: not cmd argument
         ('/foo/bar?key=tip', 'pull'),
     ])
 def test_get_action(url, expected_action):
     app = simplehg.SimpleHg(application=None,
                             config={'auth_ret_code': '', 'base_path': ''},
                             registry=None)
     assert expected_action == app._get_action(get_environ(url))
 
 
 @pytest.mark.parametrize(
     'url, expected_repo_name',
     [
         ('/foo?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz?cmd=listkeys&key=tip', 'foo/bar/baz'),
         # Repos with trailing slashes.
         ('/foo/?cmd=unbundle&key=tip', 'foo'),
         ('/foo/bar/?cmd=pushkey&key=tip', 'foo/bar'),
         ('/foo/bar/baz/?cmd=listkeys&key=tip', 'foo/bar/baz'),
     ])
 def test_get_repository_name(url, expected_repo_name):
     app = simplehg.SimpleHg(application=None,
                             config={'auth_ret_code': '', 'base_path': ''},
                             registry=None)
     assert expected_repo_name == app._get_repository_name(get_environ(url))
 
 
 def test_get_config(pylonsapp, user_util):
     repo = user_util.create_repo(repo_type='git')
     app = simplehg.SimpleHg(application=None,
                             config={'auth_ret_code': '', 'base_path': ''},
                             registry=None)
-    extras = {'foo': 'FOO', 'bar': 'BAR'}
+    extras = [('foo', 'FOO', 'bar', 'BAR')]
 
     hg_config = app._create_config(extras, repo_name=repo.repo_name)
 
     config = simplehg.utils.make_db_config(repo=repo.repo_name)
     config.set('rhodecode', 'RC_SCM_DATA', json.dumps(extras))
     hg_config_org = config
 
     expected_config = [
         ('vcs_svn_tag', 'ff89f8c714d135d865f44b90e5413b88de19a55f', '/tags/*'),
         ('web', 'push_ssl', 'False'),
         ('web', 'allow_push', '*'),
         ('web', 'allow_archive', 'gz zip bz2'),
         ('web', 'baseurl', '/'),
         ('vcs_git_lfs', 'store_location', hg_config_org.get('vcs_git_lfs', 'store_location')),
         ('vcs_svn_branch', '9aac1a38c3b8a0cdc4ae0f960a5f83332bc4fa5e', '/branches/*'),
         ('vcs_svn_branch', 'c7e6a611c87da06529fd0dd733308481d67c71a8', '/trunk'),
         ('largefiles', 'usercache', hg_config_org.get('largefiles', 'usercache')),
         ('hooks', 'preoutgoing.pre_pull', 'python:vcsserver.hooks.pre_pull'),
         ('hooks', 'prechangegroup.pre_push', 'python:vcsserver.hooks.pre_push'),
         ('hooks', 'outgoing.pull_logger', 'python:vcsserver.hooks.log_pull_action'),
         ('hooks', 'pretxnchangegroup.pre_push', 'python:vcsserver.hooks.pre_push'),
         ('hooks', 'changegroup.push_logger', 'python:vcsserver.hooks.log_push_action'),
         ('hooks', 'changegroup.repo_size', 'python:vcsserver.hooks.repo_size'),
         ('phases', 'publish', 'True'),
         ('extensions', 'largefiles', ''),
         ('paths', '/', hg_config_org.get('paths', '/')),
-        ('rhodecode', 'RC_SCM_DATA', '{"foo": "FOO", "bar": "BAR"}')
+        ('rhodecode', 'RC_SCM_DATA', '[["foo", "FOO", "bar", "BAR"]]')
     ]
     for entry in expected_config:
         assert entry in hg_config
 
 
 def test_create_wsgi_app_uses_scm_app_from_simplevcs():
     config = {
         'auth_ret_code': '',
         'base_path': '',
         'vcs.scm_app_implementation':
             'rhodecode.tests.lib.middleware.mock_scm_app',
     }
     app = simplehg.SimpleHg(application=None, config=config, registry=None)
     wsgi_app = app._create_wsgi_app('/tmp/test', 'test_repo', {})
     assert wsgi_app is mock_scm_app.mock_hg_wsgi