engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feature: initial implementation of qoutas for a user, repo count part

Browse source
This commit is contained in:
ievgenii vdovenko 2025-11-06 16:25:54 +01:00
parent a78d2157a2
commit 746c5b4699
11 changed files with 273 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

77
rhodecode/apps/admin/tests/test_admin_repos.py
View file

@ -19,6 +19,7 @@
import urllib.request
import urllib.parse
import urllib.error
from unittest.mock import PropertyMock
import mock
import pytest
@ -29,6 +30,7 @@ from rhodecode.lib.utils2 import safe_str
from rhodecode.lib import helpers as h
from rhodecode.model.db import Repository, RepoGroup, UserRepoToPerm, User, Permission
from rhodecode.model.meta import Session
from rhodecode.model.quota_model import UserQuotaModel
from rhodecode.model.repo import RepoModel
from rhodecode.model.repo_group import RepoGroupModel
from rhodecode.model.user import UserModel
@ -85,7 +87,7 @@ class TestAdminRepos(object):
description_unicode = "description for newly created repo" + suffix
description = description_unicode
response = self.app.post(
self.app.post(
route_path("repo_create"),
fixture._get_repo_create_params(
repo_private=False,
@ -99,6 +101,79 @@ class TestAdminRepos(object):
self.assert_repository_is_created_correctly(repo_name, description, backend)
def test_create_fails_due_to_quota(self, backend, monkeypatch):
import rhodecode as app
monkeypatch.setitem(app.CONFIG, "user.quotas.enabled", True)
session = login_user_session(self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
csrf_token = auth.get_csrf_token(session)
repo_name_unicode = backend.new_repo_name(suffix="no_quota")
repo_name = repo_name_unicode
user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
user.update_quotas(
max_disk_space_mb_allowance=200,
max_repository_count_allowance=2,
current_repos_count=2,
used_disk_space_bytes=0,
)
self.app.post(
route_path("repo_create"),
fixture._get_repo_create_params(
repo_name=repo_name,
repo_type=backend.alias,
repo_description="no_quota",
csrf_token=csrf_token,
),
status=302,
)
response = self.app.get(route_path("repo_creating_check", repo_name=repo_name))
assert response.json == {"result": False}
flash_msg = "User quotas exceeded. You can't create more repositories."
assert_session_flash(response, flash_msg)
new_repo = RepoModel().get_by_repo_name(repo_name)
assert new_repo is None
def test_create_count_user_repos(self, backend, monkeypatch):
import rhodecode as app
monkeypatch.setitem(app.CONFIG, "user.quotas.enabled", True)
session = login_user_session(self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
csrf_token = auth.get_csrf_token(session)
repo_name_unicode = backend.new_repo_name(suffix="no_quota")
repo_name = repo_name_unicode
user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
user.update_quotas(
max_disk_space_mb_allowance=200,
max_repository_count_allowance=2,
current_repos_count=0,
used_disk_space_bytes=0,
)
self.app.post(
route_path("repo_create"),
fixture._get_repo_create_params(
repo_name=repo_name,
repo_type=backend.alias,
repo_description="quota",
csrf_token=csrf_token,
),
status=302,
)
self.assert_repository_is_created_correctly(repo_name, "quota", backend)
user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
assert user.get_quotas()["current_repos_count"] == 1
def test_create_numeric_name(self, autologin_user, backend, csrf_token):
numeric_repo = "1234"
repo_name = numeric_repo

11
rhodecode/apps/admin/views/repositories.py
View file

@ -36,6 +36,7 @@ from rhodecode.lib.utils import repo_name_slug
from rhodecode.lib.utils2 import safe_int, safe_str
from rhodecode.model.forms import RepoForm
from rhodecode.model.permission import PermissionModel
from rhodecode.model.quota_model import UserQuotaModel
from rhodecode.model.repo import RepoModel
from rhodecode.model.scm import RepoList, RepoGroupList, ScmModel
from rhodecode.model.settings import SettingsModel
@ -220,6 +221,11 @@ class AdminReposView(BaseAppView, DataGridAppView):
copy_permissions = form_result.get("repo_copy_permissions")
# create is done sometimes async on celery, db transaction
# management is handled there.
quota_model = UserQuotaModel(self._rhodecode_db_user)
if not quota_model.is_repo_creation_allowed():
raise UserQuotaModel.QuotaError("User quotas exceeded. You can't create more repositories.")
task = RepoModel().create(form_result, self._rhodecode_user.user_id)
task_id = get_task_id(task)
except formencode.Invalid as errors:
@ -234,6 +240,11 @@ class AdminReposView(BaseAppView, DataGridAppView):
)
return Response(html)
except UserQuotaModel.QuotaError as e:
msg = str(e)
log.error(msg)
h.flash(msg, category="error")
raise HTTPFound(h.route_path("home"))
except Exception as e:
msg = self._log_creation_exception(e, form_result.get("repo_name"))
h.flash(msg, category="error")

42
rhodecode/apps/repository/tests/test_repo_settings_advanced.py
View file

@ -19,11 +19,13 @@ import pytest
from unittest.mock import patch, MagicMock
from rhodecode.apps.repository.views.repo_settings_advanced import RepoSettingsAdvancedView
from rhodecode.bootstrap import TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS
from rhodecode.lib import auth
from rhodecode.lib.str_utils import safe_str
from rhodecode.model.db import Repository
from rhodecode.model.db import Repository, User
from rhodecode.model.meta import Session
from rhodecode.model.repo import RepoModel, ForksAction
from rhodecode.tests import HG_REPO, GIT_REPO, assert_session_flash, no_newline_id_generator
from rhodecode.tests import HG_REPO, GIT_REPO, assert_session_flash, no_newline_id_generator, login_user_session
from rhodecode.tests.fixtures.rc_fixture import Fixture
from rhodecode.tests.utils import repo_on_filesystem
from rhodecode.tests.routes import route_path
@ -115,6 +117,42 @@ class TestAdminRepoSettingsAdvanced(object):
assert RepoModel().get_by_repo_name(repo_name) is None
assert not repo_on_filesystem(repo_name_str)
def test_advanced_repo_delete_decrement_repo_quota(self, backend, monkeypatch):
import rhodecode as app
monkeypatch.setitem(app.CONFIG, "user.quotas.enabled", True)
session = login_user_session(self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
csrf_token = auth.get_csrf_token(session)
user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
user.update_quotas(
max_disk_space_mb_allowance=200,
max_repository_count_allowance=2,
current_repos_count=1,
used_disk_space_bytes=0,
)
repo = backend.create_repo(name_suffix="quota")
repo.user = user
Session().commit()
repo_name = repo.repo_name
repo_name_str = safe_str(repo.repo_name)
response = self.app.post(
route_path("edit_repo_advanced_delete", repo_name=repo_name_str), params={"csrf_token": csrf_token}
)
assert_session_flash(response, "Deleted repository `{}`".format(repo_name))
response.follow()
user: User = User.get_by_username(TEST_USER_REGULAR_LOGIN)
# check if repo was deleted from db
assert RepoModel().get_by_repo_name(repo_name) is None
assert not repo_on_filesystem(repo_name_str)
assert user.get_quotas()["current_repos_count"] == 0
@pytest.mark.parametrize("suffix", ["", "ąęł", "123"], ids=no_newline_id_generator)
def test_delete_broken_repo(self, autologin_user, backend, suffix, csrf_token):
repo = backend.create_repo(name_suffix=suffix, private_repo=True)

4
rhodecode/apps/repository/views/repo_settings_advanced.py
View file

@ -32,6 +32,7 @@ from rhodecode.lib.utils2 import safe_int
from rhodecode.lib.vcs import RepositoryError
from rhodecode.model.db import Session, UserFollowing, User, Repository
from rhodecode.model.permission import PermissionModel
from rhodecode.model.quota_model import UserQuotaModel
from rhodecode.model.repo import RepoModel, ForksAction
from rhodecode.model.scm import ScmModel
@ -157,6 +158,9 @@ class RepoSettingsAdvancedView(RepoAppView):
self.request.POST.get("backup", "false")
) # if this option is absent, this means that on FE side checkbox was unchecked
delete_res = RepoModel().delete(self.db_repo, fs_backup=backup)
repo_owner = self.db_repo.user
quota_model = UserQuotaModel(repo_owner)
quota_model.decrement_repo_count()
fork_action = delete_res.fork_action_result.action
fork_cnt = delete_res.fork_action_result.fork_count