tests: fixes existing tests and add new assertion

This commit is contained in:
ievgenii vdovenko 2025-07-26 10:36:17 +02:00
parent 3ebb39e681
commit 6e8b6c4975
8 changed files with 55 additions and 117 deletions

View file

@ -31,7 +31,12 @@ class TestGetUsers(object):
)
response = api_call(self.app, params)
ret_all = []
_users = User.query().filter(User.username != User.DEFAULT_USER).order_by(User.username).all()
_users = (
User.query()
.filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER]))
.order_by(User.username)
.all()
)
for usr in _users:
ret = usr.get_api_data(include_secrets=True)
ret_all.append(jsonify(ret))

View file

@ -21,7 +21,7 @@ from sqlalchemy.orm.exc import NoResultFound
from rhodecode.lib import auth
from rhodecode.lib import helpers as h
from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository
from rhodecode.model.db import User, UserApiKeys, UserEmailMap, Repository, RepoGroup, UserGroup
from rhodecode.model.meta import Session
from rhodecode.model.user import UserModel
@ -41,7 +41,7 @@ class TestAdminUsersView(TestController):
self.log_user()
response = self.app.get(route_path("users_data"), extra_environ=xhr_header)
all_users = User.query().filter(User.username != User.DEFAULT_USER).count()
all_users = User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).count()
assert response.json["recordsTotal"] == all_users
def test_show_users_data_filtered(self, xhr_header):
@ -50,7 +50,7 @@ class TestAdminUsersView(TestController):
route_path("users_data", params={"search[value]": "empty_search"}), extra_environ=xhr_header
)
all_users = User.query().filter(User.username != User.DEFAULT_USER).count()
all_users = User.query().filter(User.username.notin_([User.DEFAULT_USER, User.GHOST_USER])).count()
assert response.json["recordsTotal"] == all_users
assert response.json["recordsFiltered"] == 0
@ -448,25 +448,6 @@ class TestAdminUsersView(TestController):
assert_session_flash(response, "Successfully deleted user `{}`".format(username))
def test_delete_owner_of_repository(self, request, user_util):
self.log_user()
obj_name = "test_repo"
usr = user_util.create_user()
username = usr.username
fixture.create_repo(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 repositories and cannot be removed. '
"Switch owners or remove those repositories:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_repo(obj_name)
def test_delete_owner_of_repository_detaching(self, request, user_util):
self.log_user()
obj_name = "test_repo"
@ -478,48 +459,18 @@ class TestAdminUsersView(TestController):
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repos": "detach", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Detached 1 repositories"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
repo = Repository.get_by_repo_name(obj_name)
assert repo.user_id == ghost_user.user_id, "Ghost user should be a new owner of the repo."
fixture.destroy_repo(obj_name)
def test_delete_owner_of_repository_deleting(self, request, user_util):
self.log_user()
obj_name = "test_repo"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_repo(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repos": "delete", "csrf_token": self.csrf_token},
)
msg = "Deleted 1 repositories"
assert_session_flash(response, msg)
def test_delete_owner_of_repository_group(self, request, user_util):
self.log_user()
obj_name = "test_group"
usr = user_util.create_user()
username = usr.username
fixture.create_repo_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 repository groups and cannot be removed. '
"Switch owners or remove those repository groups:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_repo_group(obj_name)
def test_delete_owner_of_repository_group_detaching(self, request, user_util):
self.log_user()
obj_name = "test_group"
@ -530,48 +481,18 @@ class TestAdminUsersView(TestController):
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repo_groups": "delete", "csrf_token": self.csrf_token},
)
msg = "Deleted 1 repository groups"
assert_session_flash(response, msg)
def test_delete_owner_of_repository_group_deleting(self, request, user_util):
self.log_user()
obj_name = "test_group"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_repo_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_repo_groups": "detach", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Detached 1 repository groups"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
repo_group = RepoGroup.get_by_group_name(obj_name)
assert repo_group.user_id == ghost_user.user_id, "Ghost user should be a new owner of the repo_group."
fixture.destroy_repo_group(obj_name)
def test_delete_owner_of_user_group(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
usr = user_util.create_user()
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id), params={"csrf_token": self.csrf_token}
)
msg = (
'user "%s" still owns 1 user groups and cannot be removed. '
"Switch owners or remove those user groups:%s" % (username, obj_name)
)
assert_session_flash(response, msg)
fixture.destroy_user_group(obj_name)
def test_delete_owner_of_user_group_detaching(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
@ -579,34 +500,21 @@ class TestAdminUsersView(TestController):
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
try:
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_user_groups": "detach", "csrf_token": self.csrf_token},
)
msg = "Detached 1 user groups"
assert_session_flash(response, msg)
finally:
fixture.destroy_user_group(obj_name)
def test_delete_owner_of_user_group_deleting(self, request, user_util):
self.log_user()
obj_name = "test_user_group"
usr = user_util.create_user(auto_cleanup=False)
username = usr.username
fixture.create_user_group(obj_name, cur_user=usr.username)
new_user = Session().query(User).filter(User.username == username).one()
response = self.app.post(
route_path("user_delete", user_id=new_user.user_id),
params={"user_user_groups": "delete", "csrf_token": self.csrf_token},
params={"csrf_token": self.csrf_token},
)
msg = "Deleted 1 user groups"
msg = "Detached 1 user groups"
assert_session_flash(response, msg)
ghost_user = User.get_ghost_user()
user_group = UserGroup.get_by_group_name(obj_name)
assert user_group.user_id == ghost_user.user_id, "Ghost user should be a new owner of the user_group."
fixture.destroy_user_group(obj_name)
def test_edit(self, user_util):
self.log_user()
user = user_util.create_user()

View file

@ -67,5 +67,8 @@ class TestUsersController(TestController):
# default user, not allowed to show it
self.app.get(route_path("user_profile", username=User.DEFAULT_USER), status=404)
# ghost user, not allowed to show it
self.app.get(route_path("user_profile", username=User.GHOST_USER), status=404)
# actual 404
self.app.get(route_path("user_profile", username="unknown"), status=404)

View file

@ -605,6 +605,21 @@ class DbManage(object):
user=user, lifetime_minutes=-1, role=UserModel.auth_token_role.ROLE_ALL, description="BUILTIN TOKEN"
)
def create_ghost_user(self):
log.info("creating ghost user")
if not User.get_by_username(User.GHOST_USER):
ghost_user = User()
ghost_user.name = User.GHOST_USER.capitalize()
ghost_user.username = User.GHOST_USER
ghost_user.lastname = "User"
ghost_user.strict_creation_check = False
ghost_user.email = User.GHOST_USER_EMAIL
ghost_user.active = False
ghost_user.description = "System user to own assets of deleted users (does not consume license slot)"
ghost_user.is_new_user = False
Session().add(ghost_user)
Session().commit()
def create_default_user(self):
log.info("creating default user")
# create default user for handling default permissions.

View file

@ -96,6 +96,7 @@ def command(
opts = dbmanage.config_prompt(None)
dbmanage.create_settings(opts)
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_admin_and_prompt()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()

View file

@ -841,6 +841,7 @@ def create_test_database(test_path, config):
# for tests dynamically set new root paths based on generated content
dbmanage.create_settings(dbmanage.config_prompt(test_path))
dbmanage.create_default_user()
dbmanage.create_ghost_user()
dbmanage.create_test_admin_and_users()
dbmanage.create_permissions()
dbmanage.populate_default_permissions()

View file

@ -128,6 +128,9 @@ class UserModel(BaseModel):
user = user.options(FromCache("sql_cache_short", f"get_user_{name_key}"))
return user.scalar()
def get_by_id(self, id_):
return self.sa.query(User).filter(User.user_id == id_).scalar()
def get_by_email(self, email, cache=False, case_insensitive=False):
return User.get_by_email(email, case_insensitive, cache)

View file

@ -1297,8 +1297,10 @@ class UserUtility(object):
self.fixture.destroy_user_group(user_group_id)
def _cleanup_users(self):
user_model = UserModel()
for user_id in self.user_ids:
self.fixture.destroy_user(user_id)
if user_model.get_by_id(user_id):
self.fixture.destroy_user(user_id)
@pytest.fixture(scope="session")