312 lines
11 KiB
Python
312 lines
11 KiB
Python
# Copyright (C) 2010-2024 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 pytest
|
|
import mock
|
|
|
|
from rhodecode.model.db import true, User, UserGroup, UserGroupMember, UserEmailMap, Permission, UserIpMap
|
|
from rhodecode.model.meta import Session
|
|
from rhodecode.model.user import UserModel
|
|
from rhodecode.model.user_group import UserGroupModel
|
|
from rhodecode.model.repo import RepoModel
|
|
from rhodecode.model.repo_group import RepoGroupModel
|
|
from rhodecode.tests.fixtures.rc_fixture import Fixture
|
|
from rhodecode.lib.str_utils import safe_str
|
|
|
|
|
|
fixture = Fixture()
|
|
|
|
|
|
class TestGetUsers(object):
|
|
def test_returns_active_users(self, backend, user_util):
|
|
for i in range(4):
|
|
is_active = i % 2 == 0
|
|
user_util.create_user(active=is_active, lastname="Fake user")
|
|
|
|
with mock.patch("rhodecode.lib.helpers.gravatar_url"):
|
|
with mock.patch("rhodecode.lib.helpers.link_to_user"):
|
|
users = UserModel().get_users()
|
|
fake_users = [u for u in users if u["last_name"] == "Fake user"]
|
|
assert len(fake_users) == 2
|
|
|
|
expected_keys = (
|
|
"id",
|
|
"first_name",
|
|
"last_name",
|
|
"username",
|
|
"icon_link",
|
|
"value_display",
|
|
"value",
|
|
"value_type",
|
|
)
|
|
for user in users:
|
|
assert user["value_type"] == "user"
|
|
for key in expected_keys:
|
|
assert key in user
|
|
|
|
def test_returns_user_filtered_by_last_name(self, backend, user_util):
|
|
keywords = ("aBc", "ünicode")
|
|
for keyword in keywords:
|
|
for i in range(2):
|
|
user_util.create_user(active=True, lastname="Fake {} user".format(keyword))
|
|
|
|
with mock.patch("rhodecode.lib.helpers.gravatar_url"):
|
|
with mock.patch("rhodecode.lib.helpers.link_to_user"):
|
|
keyword = keywords[1].lower()
|
|
users = UserModel().get_users(name_contains=keyword)
|
|
|
|
fake_users = [u for u in users if u["last_name"].startswith("Fake")]
|
|
assert len(fake_users) == 2
|
|
for user in fake_users:
|
|
assert user["last_name"] == safe_str("Fake ünicode user")
|
|
|
|
def test_returns_user_filtered_by_first_name(self, backend, user_util):
|
|
created_users = []
|
|
keywords = ("aBc", "ünicode")
|
|
for keyword in keywords:
|
|
for i in range(2):
|
|
created_users.append(
|
|
user_util.create_user(active=True, lastname="Fake user", firstname="Fake {} user".format(keyword))
|
|
)
|
|
|
|
keyword = keywords[1].lower()
|
|
with mock.patch("rhodecode.lib.helpers.gravatar_url"):
|
|
with mock.patch("rhodecode.lib.helpers.link_to_user"):
|
|
users = UserModel().get_users(name_contains=keyword)
|
|
|
|
fake_users = [u for u in users if u["last_name"].startswith("Fake")]
|
|
assert len(fake_users) == 2
|
|
for user in fake_users:
|
|
assert user["first_name"] == safe_str("Fake ünicode user")
|
|
|
|
def test_returns_user_filtered_by_username(self, backend, user_util):
|
|
created_users = []
|
|
for i in range(5):
|
|
created_users.append(user_util.create_user(active=True, lastname="Fake user"))
|
|
|
|
user_filter = created_users[-1].username[-2:]
|
|
with mock.patch("rhodecode.lib.helpers.gravatar_url"):
|
|
with mock.patch("rhodecode.lib.helpers.link_to_user"):
|
|
users = UserModel().get_users(name_contains=user_filter)
|
|
|
|
fake_users = [u for u in users if u["last_name"].startswith("Fake")]
|
|
assert len(fake_users) == 1
|
|
assert fake_users[0]["username"] == created_users[-1].username
|
|
|
|
def test_returns_limited_user_list(self, backend, user_util):
|
|
created_users = []
|
|
for i in range(5):
|
|
created_users.append(user_util.create_user(active=True, lastname="Fake user"))
|
|
|
|
with mock.patch("rhodecode.lib.helpers.gravatar_url"):
|
|
with mock.patch("rhodecode.lib.helpers.link_to_user"):
|
|
users = UserModel().get_users(name_contains="Fake", limit=3)
|
|
|
|
fake_users = [u for u in users if u["last_name"].startswith("Fake")]
|
|
assert len(fake_users) == 3
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_user(request, baseapp):
|
|
usr = UserModel().create_or_update(
|
|
username="test_user", password="qweqwe", email="main_email@rhodecode.org", firstname="u1", lastname="u1"
|
|
)
|
|
Session().commit()
|
|
assert User.get_by_username("test_user") == usr
|
|
|
|
@request.addfinalizer
|
|
def cleanup():
|
|
if UserModel().get_user(usr.user_id) is None:
|
|
return
|
|
|
|
perm = Permission.query().all()
|
|
for p in perm:
|
|
UserModel().revoke_perm(usr, p)
|
|
|
|
UserModel().delete(usr.user_id)
|
|
Session().commit()
|
|
|
|
return usr
|
|
|
|
|
|
def test_create_and_remove(test_user):
|
|
usr = test_user
|
|
|
|
# make user group
|
|
user_group = fixture.create_user_group("some_example_group")
|
|
Session().commit()
|
|
|
|
UserGroupModel().add_user_to_group(user_group, usr)
|
|
Session().commit()
|
|
|
|
assert UserGroup.get(user_group.users_group_id) == user_group
|
|
assert UserGroupMember.query().count() == 1
|
|
UserModel().delete(usr.user_id)
|
|
Session().commit()
|
|
|
|
assert UserGroupMember.query().all() == []
|
|
|
|
|
|
def test_additonal_email_as_main(test_user):
|
|
with pytest.raises(AttributeError):
|
|
m = UserEmailMap()
|
|
m.email = test_user.email
|
|
m.user = test_user
|
|
Session().add(m)
|
|
Session().commit()
|
|
|
|
|
|
def test_extra_email_map(test_user):
|
|
m = UserEmailMap()
|
|
m.email = "main_email2@rhodecode.org"
|
|
m.user = test_user
|
|
Session().add(m)
|
|
Session().commit()
|
|
|
|
u = User.get_by_email(email="main_email@rhodecode.org")
|
|
assert test_user.user_id == u.user_id
|
|
assert test_user.username == u.username
|
|
|
|
u = User.get_by_email(email="main_email2@rhodecode.org")
|
|
assert test_user.user_id == u.user_id
|
|
assert test_user.username == u.username
|
|
u = User.get_by_email(email="main_email3@rhodecode.org")
|
|
assert u is None
|
|
|
|
|
|
def test_get_api_data_replaces_secret_data_by_default(test_user):
|
|
api_data = test_user.get_api_data()
|
|
api_key_length = 40
|
|
expected_replacement = "*" * api_key_length
|
|
|
|
for key in api_data["auth_tokens"]:
|
|
assert key == expected_replacement
|
|
|
|
|
|
def test_get_api_data_includes_secret_data_if_activated(test_user):
|
|
api_data = test_user.get_api_data(include_secrets=True)
|
|
assert api_data["auth_tokens"] == test_user.auth_tokens
|
|
|
|
|
|
def test_add_perm(test_user):
|
|
perm = Permission.query().all()[0]
|
|
UserModel().grant_perm(test_user, perm)
|
|
Session().commit()
|
|
assert UserModel().has_perm(test_user, perm)
|
|
|
|
|
|
def test_has_perm(test_user):
|
|
perm = Permission.query().all()
|
|
for p in perm:
|
|
assert not UserModel().has_perm(test_user, p)
|
|
|
|
|
|
def test_revoke_perm(test_user):
|
|
perm = Permission.query().all()[0]
|
|
UserModel().grant_perm(test_user, perm)
|
|
Session().commit()
|
|
assert UserModel().has_perm(test_user, perm)
|
|
|
|
# revoke
|
|
UserModel().revoke_perm(test_user, perm)
|
|
Session().commit()
|
|
assert not UserModel().has_perm(test_user, perm)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"ip_range, expected, expect_errors",
|
|
[
|
|
("", [], False),
|
|
("127.0.0.1", ["127.0.0.1"], False),
|
|
("127.0.0.1,127.0.0.2", ["127.0.0.1", "127.0.0.2"], False),
|
|
("127.0.0.1 , 127.0.0.2", ["127.0.0.1", "127.0.0.2"], False),
|
|
("127.0.0.1,172.172.172.0,127.0.0.2", ["127.0.0.1", "172.172.172.0", "127.0.0.2"], False),
|
|
("127.0.0.1-127.0.0.5", ["127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"], False),
|
|
("127.0.0.1 - 127.0.0.5", ["127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5"], False),
|
|
("-", [], True),
|
|
("127.0.0.1-32", [], True),
|
|
("127.0.0.1,127.0.0.1,127.0.0.1,127.0.0.1-127.0.0.2,127.0.0.2", ["127.0.0.1", "127.0.0.2"], False),
|
|
(
|
|
"127.0.0.1-127.0.0.2,127.0.0.4-127.0.0.6,",
|
|
["127.0.0.1", "127.0.0.2", "127.0.0.4", "127.0.0.5", "127.0.0.6"],
|
|
False,
|
|
),
|
|
(
|
|
"127.0.0.1-127.0.0.2,127.0.0.1-127.0.0.6,",
|
|
["127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4", "127.0.0.5", "127.0.0.6"],
|
|
False,
|
|
),
|
|
],
|
|
)
|
|
def test_ip_range_generator(ip_range, expected, expect_errors):
|
|
func = UserModel().parse_ip_range
|
|
if expect_errors:
|
|
pytest.raises(ValueError, func, ip_range)
|
|
else:
|
|
parsed_list = func(ip_range)
|
|
assert parsed_list == expected
|
|
|
|
|
|
def test_user_delete_cascades_ip_whitelist(test_user):
|
|
sample_ip = "1.1.1.1"
|
|
uid_map = UserIpMap(user_id=test_user.user_id, ip_addr=sample_ip)
|
|
Session().add(uid_map)
|
|
Session().delete(test_user)
|
|
try:
|
|
Session().flush()
|
|
finally:
|
|
Session().rollback()
|
|
|
|
|
|
def test_account_for_deactivation_generation(test_user):
|
|
accounts = UserModel().get_accounts_in_creation_order(current_user=test_user)
|
|
# current user should be #1 in the list
|
|
assert accounts[0] == test_user.user_id
|
|
active_users = User.query().filter(User.active == true()).count()
|
|
assert active_users == len(accounts)
|
|
|
|
|
|
def test_user_delete_cascades_permissions_on_repo(backend, test_user):
|
|
test_repo = backend.create_repo()
|
|
RepoModel().grant_user_permission(test_repo, test_user, "repository.write")
|
|
Session().commit()
|
|
|
|
assert test_user.repo_to_perm
|
|
|
|
UserModel().delete(test_user)
|
|
Session().commit()
|
|
|
|
|
|
def test_user_delete_cascades_permissions_on_repo_group(test_repo_group, test_user):
|
|
RepoGroupModel().grant_user_permission(test_repo_group, test_user, "group.write")
|
|
Session().commit()
|
|
|
|
assert test_user.repo_group_to_perm
|
|
|
|
Session().delete(test_user)
|
|
Session().commit()
|
|
|
|
|
|
def test_user_delete_cascades_permissions_on_user_group(test_user_group, test_user):
|
|
UserGroupModel().grant_user_permission(test_user_group, test_user, "usergroup.write")
|
|
Session().commit()
|
|
|
|
assert test_user.user_group_to_perm
|
|
|
|
Session().delete(test_user)
|
|
Session().commit()
|