##// END OF EJS Templates
ip-restritions: fix issue with loading empty IP rules for non-logged not-anonymous users....
ip-restritions: fix issue with loading empty IP rules for non-logged not-anonymous users. This case won't have any IP rules set. And we accidentally loaded them becuase of no empty user filter was present. This caused some odd display warnings for users.

File last commit:

r2592:0e0508de default
r2838:d1dcf373 default
Show More
fixture.py
349 lines | 11.6 KiB | text/x-python | PythonLexer
project: added all source files and assets
r1 # -*- coding: utf-8 -*-
release: update copyright year to 2018
r2487 # Copyright (C) 2010-2018 RhodeCode GmbH
project: added all source files and assets
r1 #
# 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/
"""
Helpers for fixture generation
"""
import os
import time
import tempfile
import shutil
import configobj
from rhodecode.tests import *
Bartłomiej Wołyńczyk
my-account: security change, added select filed with email from extra emails while editing user profile, now adding extra emails required type password. Task #5386
r2592 from rhodecode.model.db import Repository, User, RepoGroup, UserGroup, Gist, UserEmailMap
project: added all source files and assets
r1 from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.user_group import UserGroupModel
from rhodecode.model.gist import GistModel
auth-tokens: fixed tests
r1482 from rhodecode.model.auth_token import AuthTokenModel
project: added all source files and assets
r1
dn = os.path.dirname
FIXTURES = os.path.join(dn(dn(os.path.abspath(__file__))), 'tests', 'fixtures')
def error_function(*args, **kwargs):
raise Exception('Total Crash !')
class TestINI(object):
"""
Allows to create a new test.ini file as a copy of existing one with edited
data. Example usage::
with TestINI('test.ini', [{'section':{'key':val'}]) as new_test_ini_path:
print 'paster server %s' % new_test_ini
"""
def __init__(self, ini_file_path, ini_params, new_file_prefix='DEFAULT',
destroy=True, dir=None):
self.ini_file_path = ini_file_path
self.ini_params = ini_params
self.new_path = None
self.new_path_prefix = new_file_prefix
self._destroy = destroy
self._dir = dir
def __enter__(self):
return self.create()
def __exit__(self, exc_type, exc_val, exc_tb):
self.destroy()
def create(self):
config = configobj.ConfigObj(
self.ini_file_path, file_error=True, write_empty_values=True)
for data in self.ini_params:
section, ini_params = data.items()[0]
tests: Fix --without-vcsserver parameter...
r816 for key, val in ini_params.items():
config[section][key] = val
project: added all source files and assets
r1 with tempfile.NamedTemporaryFile(
prefix=self.new_path_prefix, suffix='.ini', dir=self._dir,
delete=False) as new_ini_file:
config.write(new_ini_file)
self.new_path = new_ini_file.name
return self.new_path
def destroy(self):
if self._destroy:
os.remove(self.new_path)
class Fixture(object):
def anon_access(self, status):
"""
Context process for disabling anonymous access. use like:
fixture = Fixture()
with fixture.anon_access(False):
#tests
after this block anon access will be set to `not status`
"""
class context(object):
def __enter__(self):
anon = User.get_default_user()
anon.active = status
Session().add(anon)
Session().commit()
time.sleep(1.5) # must sleep for cache (1s to expire)
def __exit__(self, exc_type, exc_val, exc_tb):
anon = User.get_default_user()
anon.active = not status
Session().add(anon)
Session().commit()
return context()
def _get_repo_create_params(self, **custom):
defs = {
'repo_name': None,
'repo_type': 'hg',
'clone_uri': '',
api: update pull method with possible specification of the url
r2563 'push_uri': '',
project: added all source files and assets
r1 'repo_group': '-1',
'repo_description': 'DESC',
'repo_private': False,
'repo_landing_rev': 'rev:tip',
'repo_copy_permissions': False,
'repo_state': Repository.STATE_CREATED,
}
defs.update(custom)
if 'repo_name_full' not in custom:
defs.update({'repo_name_full': defs['repo_name']})
# fix the repo name if passed as repo_name_full
if defs['repo_name']:
defs['repo_name'] = defs['repo_name'].split('/')[-1]
return defs
def _get_group_create_params(self, **custom):
defs = {
'group_name': None,
'group_description': 'DESC',
'perm_updates': [],
'perm_additions': [],
'perm_deletions': [],
'group_parent_id': -1,
'enable_locking': False,
'recursive': False,
}
defs.update(custom)
return defs
def _get_user_create_params(self, name, **custom):
defs = {
'username': name,
'password': 'qweqwe',
'email': '%s+test@rhodecode.org' % name,
'firstname': 'TestUser',
'lastname': 'Test',
'active': True,
'admin': False,
'extern_type': 'rhodecode',
'extern_name': None,
}
defs.update(custom)
return defs
def _get_user_group_create_params(self, name, **custom):
defs = {
'users_group_name': name,
'user_group_description': 'DESC',
'users_group_active': True,
'user_group_data': {},
}
defs.update(custom)
return defs
def create_repo(self, name, **kwargs):
repo_group = kwargs.get('repo_group')
if isinstance(repo_group, RepoGroup):
kwargs['repo_group'] = repo_group.group_id
name = name.split(Repository.NAME_SEP)[-1]
name = Repository.NAME_SEP.join((repo_group.group_name, name))
if 'skip_if_exists' in kwargs:
del kwargs['skip_if_exists']
r = Repository.get_by_repo_name(name)
if r:
return r
form_data = self._get_repo_create_params(repo_name=name, **kwargs)
cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
RepoModel().create(form_data, cur_user)
Session().commit()
repo = Repository.get_by_repo_name(name)
assert repo
return repo
def create_fork(self, repo_to_fork, fork_name, **kwargs):
repo_to_fork = Repository.get_by_repo_name(repo_to_fork)
form_data = self._get_repo_create_params(repo_name=fork_name,
fork_parent_id=repo_to_fork.repo_id,
repo_type=repo_to_fork.repo_type,
**kwargs)
#TODO: fix it !!
form_data['description'] = form_data['repo_description']
form_data['private'] = form_data['repo_private']
form_data['landing_rev'] = form_data['repo_landing_rev']
owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
RepoModel().create_fork(form_data, cur_user=owner)
Session().commit()
r = Repository.get_by_repo_name(fork_name)
assert r
return r
def destroy_repo(self, repo_name, **kwargs):
RepoModel().delete(repo_name, **kwargs)
Session().commit()
def destroy_repo_on_filesystem(self, repo_name):
rm_path = os.path.join(RepoModel().repos_path, repo_name)
if os.path.isdir(rm_path):
shutil.rmtree(rm_path)
def create_repo_group(self, name, **kwargs):
if 'skip_if_exists' in kwargs:
del kwargs['skip_if_exists']
gr = RepoGroup.get_by_group_name(group_name=name)
if gr:
return gr
form_data = self._get_group_create_params(group_name=name, **kwargs)
owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
gr = RepoGroupModel().create(
group_name=form_data['group_name'],
group_description=form_data['group_name'],
owner=owner)
Session().commit()
gr = RepoGroup.get_by_group_name(gr.group_name)
return gr
def destroy_repo_group(self, repogroupid):
RepoGroupModel().delete(repogroupid)
Session().commit()
def create_user(self, name, **kwargs):
if 'skip_if_exists' in kwargs:
del kwargs['skip_if_exists']
user = User.get_by_username(name)
if user:
return user
form_data = self._get_user_create_params(name, **kwargs)
user = UserModel().create(form_data)
auth-tokens: fixed tests
r1482
# create token for user
AuthTokenModel().create(
db: use unicode for some of the defaults to reduce sqlalchemy warnings.
r1967 user=user, description=u'TEST_USER_TOKEN')
auth-tokens: fixed tests
r1482
project: added all source files and assets
r1 Session().commit()
user = User.get_by_username(user.username)
return user
def destroy_user(self, userid):
UserModel().delete(userid)
Session().commit()
Bartłomiej Wołyńczyk
my-account: security change, added select filed with email from extra emails while editing user profile, now adding extra emails required type password. Task #5386
r2592 def create_additional_user_email(self, user, email):
uem = UserEmailMap()
uem.user = user
uem.email = email
Session().add(uem)
return uem
project: added all source files and assets
r1 def destroy_users(self, userid_iter):
for user_id in userid_iter:
if User.get_by_username(user_id):
UserModel().delete(user_id)
Session().commit()
def create_user_group(self, name, **kwargs):
if 'skip_if_exists' in kwargs:
del kwargs['skip_if_exists']
gr = UserGroup.get_by_group_name(group_name=name)
if gr:
return gr
core: moved users and user groups autocomplete into pyramid....
r1666 # map active flag to the real attribute. For API consistency of fixtures
if 'active' in kwargs:
kwargs['users_group_active'] = kwargs['active']
del kwargs['active']
project: added all source files and assets
r1 form_data = self._get_user_group_create_params(name, **kwargs)
owner = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)
user_group = UserGroupModel().create(
name=form_data['users_group_name'],
description=form_data['user_group_description'],
owner=owner, active=form_data['users_group_active'],
group_data=form_data['user_group_data'])
Session().commit()
user_group = UserGroup.get_by_group_name(user_group.users_group_name)
return user_group
def destroy_user_group(self, usergroupid):
UserGroupModel().delete(user_group=usergroupid, force=True)
Session().commit()
def create_gist(self, **kwargs):
form_data = {
'description': 'new-gist',
'owner': TEST_USER_ADMIN_LOGIN,
'gist_type': GistModel.cls.GIST_PUBLIC,
'lifetime': -1,
'acl_level': Gist.ACL_LEVEL_PUBLIC,
'gist_mapping': {'filename1.txt': {'content': 'hello world'},}
}
form_data.update(kwargs)
gist = GistModel().create(
description=form_data['description'], owner=form_data['owner'],
gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],
lifetime=form_data['lifetime'], gist_acl_level=form_data['acl_level']
)
Session().commit()
return gist
def destroy_gists(self, gistid=None):
for g in GistModel.cls.get_all():
if gistid:
if gistid == g.gist_access_id:
GistModel().delete(g)
else:
GistModel().delete(g)
Session().commit()
def load_resource(self, resource_name, strip=False):
with open(os.path.join(FIXTURES, resource_name)) as f:
source = f.read()
if strip:
source = source.strip()
return source