Merge pull request #589 from taigaio/max-projects-per-user

Max projects per user
remotes/origin/logger
David Barragán Merino 2016-02-04 18:13:10 +01:00
commit 04a2cc71d1
15 changed files with 979 additions and 51 deletions

View File

@ -524,6 +524,11 @@ FRONT_SITEMAP_CACHE_TIMEOUT = 24*60*60 # In second
EXTRA_BLOCKING_CODES = [] EXTRA_BLOCKING_CODES = []
MAX_PRIVATE_PROJECTS_PER_USER = None # None == no limit
MAX_PUBLIC_PROJECTS_PER_USER = None # None == no limit
MAX_MEMBERS_PRIVATE_PROJECTS = None # None == no limit
MAX_MEMBERS_PUBLIC_PROJECTS = None # None == no limit
from .sr import * from .sr import *

View File

@ -36,6 +36,7 @@ from taiga.projects.models import Project, Membership
from taiga.projects.issues.models import Issue from taiga.projects.issues.models import Issue
from taiga.projects.tasks.models import Task from taiga.projects.tasks.models import Task
from taiga.projects.serializers import ProjectSerializer from taiga.projects.serializers import ProjectSerializer
from taiga.users import services as users_service
from . import mixins from . import mixins
from . import serializers from . import serializers
@ -90,6 +91,14 @@ class ProjectImporterViewSet(mixins.ImportThrottlingPolicyMixin, CreateModelMixi
data = request.DATA.copy() data = request.DATA.copy()
data['owner'] = data.get('owner', request.user.email) data['owner'] = data.get('owner', request.user.email)
is_private = data.get('is_private', False)
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
self.request.user,
project=Project(is_private=is_private, id=None)
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
# Create Project # Create Project
project_serialized = service.store_project(data) project_serialized = service.store_project(data)
@ -106,6 +115,14 @@ class ProjectImporterViewSet(mixins.ImportThrottlingPolicyMixin, CreateModelMixi
# Create memberships # Create memberships
if "memberships" in data: if "memberships" in data:
members = len(data['memberships'])
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
self.request.user,
project=Project(is_private=is_private, id=None),
members=max(members, 1)
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
service.store_memberships(project_serialized.object, data) service.store_memberships(project_serialized.object, data)
try: try:
@ -202,17 +219,35 @@ class ProjectImporterViewSet(mixins.ImportThrottlingPolicyMixin, CreateModelMixi
try: try:
dump = json.load(reader(dump)) dump = json.load(reader(dump))
is_private = dump["is_private"]
except Exception: except Exception:
raise exc.WrongArguments(_("Invalid dump format")) raise exc.WrongArguments(_("Invalid dump format"))
user = request.user
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
user,
project=Project(is_private=is_private, id=None)
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
if Project.objects.filter(slug=dump['slug']).exists(): if Project.objects.filter(slug=dump['slug']).exists():
del dump['slug'] del dump['slug']
members = len(dump.get("memberships", []))
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
user,
project=Project(is_private=is_private, id=None),
members=max(members, 1)
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
if settings.CELERY_ENABLED: if settings.CELERY_ENABLED:
task = tasks.load_project_dump.delay(request.user, dump) task = tasks.load_project_dump.delay(user, dump)
return response.Accepted({"import_id": task.id}) return response.Accepted({"import_id": task.id})
project = dump_service.dict_to_project(dump, request.user.email) project = dump_service.dict_to_project(dump, request.user)
response_data = ProjectSerializer(project).data response_data = ProjectSerializer(project).data
return response.Created(response_data) return response.Created(response_data)

View File

@ -17,7 +17,8 @@
from django.utils.translation import ugettext as _ from django.utils.translation import ugettext as _
from taiga.projects.models import Membership from taiga.projects.models import Membership, Project
from taiga.users import services as users_service
from . import serializers from . import serializers
from . import service from . import service
@ -89,7 +90,15 @@ def store_tags_colors(project, data):
def dict_to_project(data, owner=None): def dict_to_project(data, owner=None):
if owner: if owner:
data["owner"] = owner data["owner"] = owner.email
members = len(data.get("memberships", []))
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
owner,
project=Project(is_private=data["is_private"], id=None),
members=members
)
if not enough_slots:
raise TaigaImportError(not_enough_slots_error)
project_serialized = service.store_project(data) project_serialized = service.store_project(data)

View File

@ -25,6 +25,7 @@ from taiga.projects.models import Project
from taiga.export_import.renderers import ExportRenderer from taiga.export_import.renderers import ExportRenderer
from taiga.export_import.dump_service import dict_to_project, TaigaImportError from taiga.export_import.dump_service import dict_to_project, TaigaImportError
from taiga.export_import.service import get_errors from taiga.export_import.service import get_errors
from taiga.users.models import User
class Command(BaseCommand): class Command(BaseCommand):
@ -58,7 +59,9 @@ class Command(BaseCommand):
except Project.DoesNotExist: except Project.DoesNotExist:
pass pass
signals.post_delete.receivers = receivers_back signals.post_delete.receivers = receivers_back
dict_to_project(data, args[1])
user = User.objects.get(email=args[1])
dict_to_project(data, user)
except TaigaImportError as e: except TaigaImportError as e:
print("ERROR:", end=" ") print("ERROR:", end=" ")
print(e.message) print(e.message)

View File

@ -79,7 +79,7 @@ def delete_project_dump(project_id, project_slug, task_id):
@app.task @app.task
def load_project_dump(user, dump): def load_project_dump(user, dump):
try: try:
project = dict_to_project(dump, user.email) project = dict_to_project(dump, user)
except Exception: except Exception:
ctx = { ctx = {
"user": user, "user": user,

View File

@ -51,6 +51,7 @@ from taiga.projects.tasks.models import Task
from taiga.projects.issues.models import Issue from taiga.projects.issues.models import Issue
from taiga.projects.likes.mixins.viewsets import LikedResourceMixin, FansViewSetMixin from taiga.projects.likes.mixins.viewsets import LikedResourceMixin, FansViewSetMixin
from taiga.permissions import service as permissions_service from taiga.permissions import service as permissions_service
from taiga.users import services as users_service
from . import filters as project_filters from . import filters as project_filters
from . import models from . import models
@ -342,9 +343,13 @@ class ProjectViewSet(LikedResourceMixin, HistoryResourceMixin,
permissions_service.set_base_permissions_for_project(obj) permissions_service.set_base_permissions_for_project(obj)
def pre_save(self, obj): def pre_save(self, obj):
user = self.request.user
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(user, project=obj)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
if not obj.id: if not obj.id:
obj.owner = self.request.user obj.owner = user
# TODO REFACTOR THIS
obj.template = self.request.QUERY_PARAMS.get('template', None) obj.template = self.request.QUERY_PARAMS.get('template', None)
self._set_base_permissions(obj) self._set_base_permissions(obj)
@ -550,6 +555,15 @@ class MembershipViewSet(BlockedByProjectMixin, ModelCrudViewSet):
# TODO: this should be moved to main exception handler instead # TODO: this should be moved to main exception handler instead
# of handling explicit exception catchin here. # of handling explicit exception catchin here.
if "bulk_memberships" in data and isinstance(data["bulk_memberships"], list):
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
request.user,
project=project,
members=len(data["bulk_memberships"])
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
try: try:
members = services.create_members_in_bulk(data["bulk_memberships"], members = services.create_members_in_bulk(data["bulk_memberships"],
project=project, project=project,
@ -577,6 +591,15 @@ class MembershipViewSet(BlockedByProjectMixin, ModelCrudViewSet):
raise exc.BadRequest(_("The project must have an owner and at least one of the users must be an active admin")) raise exc.BadRequest(_("The project must have an owner and at least one of the users must be an active admin"))
def pre_save(self, obj): def pre_save(self, obj):
if not obj.id:
(enough_slots, not_enough_slots_error) = users_service.has_available_slot_for_project(
self.request.user,
project=obj.project,
members=1
)
if not enough_slots:
raise exc.BadRequest(not_enough_slots_error)
if not obj.token: if not obj.token:
obj.token = str(uuid.uuid1()) obj.token = str(uuid.uuid1())

View File

@ -30,14 +30,14 @@ admin.site.unregister(Group)
class RoleAdmin(admin.ModelAdmin): class RoleAdmin(admin.ModelAdmin):
list_display = ["name"] list_display = ["name"]
filter_horizontal = ('permissions',) filter_horizontal = ("permissions",)
def formfield_for_manytomany(self, db_field, request=None, **kwargs): def formfield_for_manytomany(self, db_field, request=None, **kwargs):
if db_field.name == 'permissions': if db_field.name == "permissions":
qs = kwargs.get('queryset', db_field.rel.to.objects) qs = kwargs.get("queryset", db_field.rel.to.objects)
# Avoid a major performance hit resolving permission names which # Avoid a major performance hit resolving permission names which
# triggers a content_type load: # triggers a content_type load:
kwargs['queryset'] = qs.select_related('content_type') kwargs["queryset"] = qs.select_related("content_type")
return super().formfield_for_manytomany( return super().formfield_for_manytomany(
db_field, request=request, **kwargs) db_field, request=request, **kwargs)
@ -47,18 +47,21 @@ class RoleAdmin(admin.ModelAdmin):
class UserAdmin(DjangoUserAdmin): class UserAdmin(DjangoUserAdmin):
fieldsets = ( fieldsets = (
(None, {'fields': ('username', 'password')}), (None, {"fields": ("username", "password")}),
(_('Personal info'), {'fields': ('full_name', 'email', 'bio', 'photo')}), (_("Personal info"), {"fields": ("full_name", "email", "bio", "photo")}),
(_('Extra info'), {'fields': ('color', 'lang', 'timezone', 'token', 'colorize_tags', 'email_token', 'new_email')}), (_("Extra info"), {"fields": ("color", "lang", "timezone", "token", "colorize_tags",
(_('Permissions'), {'fields': ('is_active', 'is_superuser',)}), "email_token", "new_email")}),
(_('Important dates'), {'fields': ('last_login', 'date_joined')}), (_("Permissions"), {"fields": ("is_active", "is_superuser")}),
(_("Restrictions"), {"fields": (("max_private_projects", "max_members_private_projects"),
("max_public_projects", "max_members_public_projects"))}),
(_("Important dates"), {"fields": ("last_login", "date_joined")}),
) )
form = UserChangeForm form = UserChangeForm
add_form = UserCreationForm add_form = UserCreationForm
list_display = ('username', 'email', 'full_name') list_display = ("username", "email", "full_name")
list_filter = ('is_superuser', 'is_active') list_filter = ("is_superuser", "is_active")
search_fields = ('username', 'full_name', 'email') search_fields = ("username", "full_name", "email")
ordering = ('username',) ordering = ("username",)
filter_horizontal = () filter_horizontal = ()
class RoleInline(admin.TabularInline): class RoleInline(admin.TabularInline):

View File

@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
from django.conf import settings
class Migration(migrations.Migration):
dependencies = [
('users', '0014_auto_20151005_1357'),
]
operations = [
migrations.AddField(
model_name='user',
name='max_private_projects',
field=models.IntegerField(null=True, verbose_name='max number of private projects owned', default=settings.MAX_PRIVATE_PROJECTS_PER_USER, blank=True),
),
migrations.AddField(
model_name='user',
name='max_public_projects',
field=models.IntegerField(null=True, verbose_name='max number of public projects owned', default=settings.MAX_PUBLIC_PROJECTS_PER_USER, blank=True),
),
]

View File

@ -0,0 +1,25 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.db import migrations, models
from django.conf import settings
class Migration(migrations.Migration):
dependencies = [
('users', '0015_auto_20160120_1409'),
]
operations = [
migrations.AddField(
model_name='user',
name='max_members_private_projects',
field=models.IntegerField(default=settings.MAX_MEMBERS_PRIVATE_PROJECTS, blank=True, verbose_name='max number of memberships for each owned private project', null=True),
),
migrations.AddField(
model_name='user',
name='max_members_public_projects',
field=models.IntegerField(default=settings.MAX_MEMBERS_PUBLIC_PROJECTS, blank=True, verbose_name='max number of memberships for each owned public project', null=True),
),
]

View File

@ -25,6 +25,7 @@ import uuid
from unidecode import unidecode from unidecode import unidecode
from django.apps import apps from django.apps import apps
from django.conf import settings
from django.contrib.contenttypes.models import ContentType from django.contrib.contenttypes.models import ContentType
from django.db import models from django.db import models
from django.dispatch import receiver from django.dispatch import receiver
@ -71,11 +72,11 @@ def get_user_file_path(instance, filename):
class PermissionsMixin(models.Model): class PermissionsMixin(models.Model):
""" """
A mixin class that adds the fields and methods necessary to support A mixin class that adds the fields and methods necessary to support
Django's Permission model using the ModelBackend. Django"s Permission model using the ModelBackend.
""" """
is_superuser = models.BooleanField(_('superuser status'), default=False, is_superuser = models.BooleanField(_("superuser status"), default=False,
help_text=_('Designates that this user has all permissions without ' help_text=_("Designates that this user has all permissions without "
'explicitly assigning them.')) "explicitly assigning them."))
class Meta: class Meta:
abstract = True abstract = True
@ -104,25 +105,25 @@ class PermissionsMixin(models.Model):
class User(AbstractBaseUser, PermissionsMixin): class User(AbstractBaseUser, PermissionsMixin):
username = models.CharField(_('username'), max_length=255, unique=True, username = models.CharField(_("username"), max_length=255, unique=True,
help_text=_('Required. 30 characters or fewer. Letters, numbers and ' help_text=_("Required. 30 characters or fewer. Letters, numbers and "
'/./-/_ characters'), "/./-/_ characters"),
validators=[ validators=[
validators.RegexValidator(re.compile('^[\w.-]+$'), _('Enter a valid username.'), 'invalid') validators.RegexValidator(re.compile("^[\w.-]+$"), _("Enter a valid username."), "invalid")
]) ])
email = models.EmailField(_('email address'), max_length=255, blank=True, unique=True) email = models.EmailField(_("email address"), max_length=255, blank=True, unique=True)
is_active = models.BooleanField(_('active'), default=True, is_active = models.BooleanField(_("active"), default=True,
help_text=_('Designates whether this user should be treated as ' help_text=_("Designates whether this user should be treated as "
'active. Unselect this instead of deleting accounts.')) "active. Unselect this instead of deleting accounts."))
full_name = models.CharField(_('full name'), max_length=256, blank=True) full_name = models.CharField(_("full name"), max_length=256, blank=True)
color = models.CharField(max_length=9, null=False, blank=True, default=generate_random_hex_color, color = models.CharField(max_length=9, null=False, blank=True, default=generate_random_hex_color,
verbose_name=_("color")) verbose_name=_("color"))
bio = models.TextField(null=False, blank=True, default="", verbose_name=_("biography")) bio = models.TextField(null=False, blank=True, default="", verbose_name=_("biography"))
photo = models.FileField(upload_to=get_user_file_path, photo = models.FileField(upload_to=get_user_file_path,
max_length=500, null=True, blank=True, max_length=500, null=True, blank=True,
verbose_name=_("photo")) verbose_name=_("photo"))
date_joined = models.DateTimeField(_('date joined'), default=timezone.now) date_joined = models.DateTimeField(_("date joined"), default=timezone.now)
lang = models.CharField(max_length=20, null=True, blank=True, default="", lang = models.CharField(max_length=20, null=True, blank=True, default="",
verbose_name=_("default language")) verbose_name=_("default language"))
theme = models.CharField(max_length=100, null=True, blank=True, default="", theme = models.CharField(max_length=100, null=True, blank=True, default="",
@ -137,16 +138,33 @@ class User(AbstractBaseUser, PermissionsMixin):
email_token = models.CharField(max_length=200, null=True, blank=True, default=None, email_token = models.CharField(max_length=200, null=True, blank=True, default=None,
verbose_name=_("email token")) verbose_name=_("email token"))
new_email = models.EmailField(_('new email address'), null=True, blank=True) new_email = models.EmailField(_("new email address"), null=True, blank=True)
is_system = models.BooleanField(null=False, blank=False, default=False) is_system = models.BooleanField(null=False, blank=False, default=False)
max_private_projects = models.IntegerField(null=True, blank=True,
default=settings.MAX_PRIVATE_PROJECTS_PER_USER,
verbose_name=_("max number of private projects owned"))
max_public_projects = models.IntegerField(null=True, blank=True,
default=settings.MAX_PUBLIC_PROJECTS_PER_USER,
verbose_name=_("max number of public projects owned"))
max_members_private_projects = models.IntegerField(null=True, blank=True,
default=settings.MAX_MEMBERS_PRIVATE_PROJECTS,
verbose_name=_("max number of memberships for "
"each owned private project"))
max_members_public_projects = models.IntegerField(null=True, blank=True,
default=settings.MAX_MEMBERS_PUBLIC_PROJECTS,
verbose_name=_("max number of memberships for "
"each owned public project"))
_cached_memberships = None _cached_memberships = None
_cached_liked_ids = None _cached_liked_ids = None
_cached_watched_ids = None _cached_watched_ids = None
_cached_notify_levels = None _cached_notify_levels = None
USERNAME_FIELD = 'username' USERNAME_FIELD = "username"
REQUIRED_FIELDS = ['email'] REQUIRED_FIELDS = ["email"]
objects = UserManager() objects = UserManager()
@ -154,9 +172,6 @@ class User(AbstractBaseUser, PermissionsMixin):
verbose_name = "user" verbose_name = "user"
verbose_name_plural = "users" verbose_name_plural = "users"
ordering = ["username"] ordering = ["username"]
permissions = (
("view_user", "Can view user"),
)
def __str__(self): def __str__(self):
return self.get_full_name() return self.get_full_name()
@ -279,16 +294,13 @@ class Role(models.Model):
verbose_name_plural = "roles" verbose_name_plural = "roles"
ordering = ["order", "slug"] ordering = ["order", "slug"]
unique_together = (("slug", "project"),) unique_together = (("slug", "project"),)
permissions = (
("view_role", "Can view role"),
)
def __str__(self): def __str__(self):
return self.name return self.name
class AuthData(models.Model): class AuthData(models.Model):
user = models.ForeignKey('users.User', related_name="auth_data") user = models.ForeignKey("users.User", related_name="auth_data")
key = models.SlugField(max_length=50) key = models.SlugField(max_length=50)
value = models.CharField(max_length=300) value = models.CharField(max_length=300)
extra = JsonField() extra = JsonField()

View File

@ -104,14 +104,30 @@ class UserSerializer(serializers.ModelSerializer):
return ContactProjectDetailSerializer(projects, many=True).data return ContactProjectDetailSerializer(projects, many=True).data
class UserAdminSerializer(UserSerializer): class UserAdminSerializer(UserSerializer):
total_private_projects = serializers.SerializerMethodField("get_total_private_projects")
total_public_projects = serializers.SerializerMethodField("get_total_public_projects")
class Meta: class Meta:
model = User model = User
# IMPORTANT: Maintain the UserSerializer Meta up to date # IMPORTANT: Maintain the UserSerializer Meta up to date
# with this info (including here the email) # with this info (including here the email)
fields = ("id", "username", "full_name", "full_name_display", "email", fields = ("id", "username", "full_name", "full_name_display", "email",
"color", "bio", "lang", "theme", "timezone", "is_active", "photo", "color", "bio", "lang", "theme", "timezone", "is_active", "photo",
"big_photo") "big_photo",
read_only_fields = ("id", "email") "max_private_projects", "max_public_projects",
"max_members_private_projects", "max_members_public_projects",
"total_private_projects", "total_public_projects")
read_only_fields = ("id", "email",
"max_private_projects", "max_public_projects",
"max_members_private_projects",
"max_members_public_projects")
def get_total_private_projects(self, user):
return user.owned_projects.filter(is_private=True).count()
def get_total_public_projects(self, user):
return user.owned_projects.filter(is_private=False).count()
class UserBasicInfoSerializer(UserSerializer): class UserBasicInfoSerializer(UserSerializer):

View File

@ -572,3 +572,43 @@ def get_voted_list(for_user, from_user, type=None, q=None):
dict(zip([col[0] for col in desc], row)) dict(zip([col[0] for col in desc], row))
for row in cursor.fetchall() for row in cursor.fetchall()
] ]
def has_available_slot_for_project(user, project, members=1):
(enough, error) = _has_available_slot_for_project_type(user, project)
if not enough:
return (enough, error)
return _has_available_slot_for_project_members(user, project, members)
def _has_available_slot_for_project_type(user, project):
if project.is_private:
if user.max_private_projects is None:
return (True, None)
elif user.owned_projects.filter(is_private=True).exclude(id=project.id).count() < user.max_private_projects:
return (True, None)
return (False, _("You can't have more private projects"))
else:
if user.max_public_projects is None:
return (True, None)
elif user.owned_projects.filter(is_private=False).exclude(id=project.id).count() < user.max_public_projects:
return (True, None)
return (False, _("You can't have more public projects"))
def _has_available_slot_for_project_members(user, project, members):
current_memberships = project.memberships.count()
if project.is_private:
if user.max_members_private_projects is None:
return (True, None)
elif current_memberships + members <= user.max_members_private_projects:
return (True, None)
return (False, _("You have reached the limit of memberships for private projects"))
else:
if user.max_members_public_projects is None:
return (True, None)
elif current_memberships + members <= user.max_members_public_projects:
return (True, None)
return (False, _("You have reached the limit of memberships for public projects"))

View File

@ -23,6 +23,7 @@ from django.core.urlresolvers import reverse
from django.core.files.base import ContentFile from django.core.files.base import ContentFile
from taiga.base.utils import json from taiga.base.utils import json
from taiga.export_import.dump_service import dict_to_project, TaigaImportError
from taiga.projects.models import Project, Membership from taiga.projects.models import Project, Membership
from taiga.projects.issues.models import Issue from taiga.projects.issues.models import Issue
from taiga.projects.userstories.models import UserStory from taiga.projects.userstories.models import UserStory
@ -72,6 +73,85 @@ def test_valid_project_import_without_extra_data(client):
assert response_data["watchers"] == [user.email, user_watching.email] assert response_data["watchers"] == [user.email, user_watching.email]
def test_valid_project_without_enough_public_projects_slots(client):
user = f.UserFactory.create(max_public_projects=0)
url = reverse("importer-list")
data = {
"slug": "public-project-without-slots",
"name": "Imported project",
"description": "Imported project",
"roles": [{"name": "Role"}],
"is_private": False
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more public projects" in response.data["_error_message"]
assert Project.objects.filter(slug="public-project-without-slots").count() == 0
def test_valid_project_without_enough_private_projects_slots(client):
user = f.UserFactory.create(max_private_projects=0)
url = reverse("importer-list")
data = {
"slug": "private-project-without-slots",
"name": "Imported project",
"description": "Imported project",
"roles": [{"name": "Role"}],
"is_private": True
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more private projects" in response.data["_error_message"]
assert Project.objects.filter(slug="private-project-without-slots").count() == 0
def test_valid_project_with_enough_public_projects_slots(client):
user = f.UserFactory.create(max_public_projects=1)
url = reverse("importer-list")
data = {
"slug": "public-project-with-slots",
"name": "Imported project",
"description": "Imported project",
"roles": [{"name": "Role"}],
"is_private": False
}
client.login(user)
response = client.json.post(url, json.dumps(data))
print(response.content)
assert response.status_code == 201
assert Project.objects.filter(slug="public-project-with-slots").count() == 1
def test_valid_project_with_enough_private_projects_slots(client):
user = f.UserFactory.create(max_private_projects=1)
url = reverse("importer-list")
data = {
"slug": "private-project-with-slots",
"name": "Imported project",
"description": "Imported project",
"roles": [{"name": "Role"}],
"is_private": True
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 201
assert Project.objects.filter(slug="private-project-with-slots").count() == 1
def test_valid_project_import_with_not_existing_memberships(client): def test_valid_project_import_with_not_existing_memberships(client):
user = f.UserFactory.create() user = f.UserFactory.create()
client.login(user) client.login(user)
@ -930,6 +1010,92 @@ def test_milestone_import_duplicated_milestone(client):
assert response_data["milestones"][0]["name"][0] == "Name duplicated for the project" assert response_data["milestones"][0]["name"][0] == "Name duplicated for the project"
def test_dict_to_project_with_no_projects_slots_available(client):
user = f.UserFactory.create(max_private_projects=0)
data = {
"slug": "valid-project",
"name": "Valid project",
"description": "Valid project desc",
"is_private": True
}
with pytest.raises(TaigaImportError) as excinfo:
project = dict_to_project(data, owner=user)
assert "can't have more private projects" in str(excinfo.value)
def test_dict_to_project_with_no_members_private_project_slots_available(client):
user = f.UserFactory.create(max_members_private_projects=2)
data = {
"slug": "valid-project",
"name": "Valid project",
"description": "Valid project desc",
"is_private": True,
"roles": [{"name": "Role"}],
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
}
]
}
with pytest.raises(TaigaImportError) as excinfo:
project = dict_to_project(data, owner=user)
assert "reached the limit of memberships for private" in str(excinfo.value)
def test_dict_to_project_with_no_members_public_project_slots_available(client):
user = f.UserFactory.create(max_members_public_projects=2)
data = {
"slug": "valid-project",
"name": "Valid project",
"description": "Valid project desc",
"is_private": False,
"roles": [{"name": "Role"}],
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
}
]
}
with pytest.raises(TaigaImportError) as excinfo:
project = dict_to_project(data, owner=user)
assert "reached the limit of memberships for public" in str(excinfo.value)
def test_invalid_dump_import(client): def test_invalid_dump_import(client):
user = f.UserFactory.create() user = f.UserFactory.create()
client.login(user) client.login(user)
@ -957,6 +1123,7 @@ def test_valid_dump_import_with_logo(client, settings):
"slug": "valid-project", "slug": "valid-project",
"name": "Valid project", "name": "Valid project",
"description": "Valid project desc", "description": "Valid project desc",
"is_private": False,
"logo": { "logo": {
"name": "logo.bmp", "name": "logo.bmp",
"data": base64.b64encode(DUMMY_BMP_DATA).decode("utf-8") "data": base64.b64encode(DUMMY_BMP_DATA).decode("utf-8")
@ -986,7 +1153,8 @@ def test_valid_dump_import_with_celery_disabled(client, settings):
data = ContentFile(bytes(json.dumps({ data = ContentFile(bytes(json.dumps({
"slug": "valid-project", "slug": "valid-project",
"name": "Valid project", "name": "Valid project",
"description": "Valid project desc" "description": "Valid project desc",
"is_private": True
}), "utf-8")) }), "utf-8"))
data.name = "test" data.name = "test"
@ -1008,7 +1176,8 @@ def test_valid_dump_import_with_celery_enabled(client, settings):
data = ContentFile(bytes(json.dumps({ data = ContentFile(bytes(json.dumps({
"slug": "valid-project", "slug": "valid-project",
"name": "Valid project", "name": "Valid project",
"description": "Valid project desc" "description": "Valid project desc",
"is_private": True
}), "utf-8")) }), "utf-8"))
data.name = "test" data.name = "test"
@ -1028,7 +1197,8 @@ def test_dump_import_duplicated_project(client):
data = ContentFile(bytes(json.dumps({ data = ContentFile(bytes(json.dumps({
"slug": project.slug, "slug": project.slug,
"name": "Test import", "name": "Test import",
"description": "Valid project desc" "description": "Valid project desc",
"is_private": True
}), "utf-8")) }), "utf-8"))
data.name = "test" data.name = "test"
@ -1051,7 +1221,8 @@ def test_dump_import_throttling(client, settings):
data = ContentFile(bytes(json.dumps({ data = ContentFile(bytes(json.dumps({
"slug": project.slug, "slug": project.slug,
"name": "Test import", "name": "Test import",
"description": "Valid project desc" "description": "Valid project desc",
"is_private": True
}), "utf-8")) }), "utf-8"))
data.name = "test" data.name = "test"
@ -1059,3 +1230,249 @@ def test_dump_import_throttling(client, settings):
assert response.status_code == 201 assert response.status_code == 201
response = client.post(url, {'dump': data}) response = client.post(url, {'dump': data})
assert response.status_code == 429 assert response.status_code == 429
def test_valid_dump_import_without_enough_public_projects_slots(client):
user = f.UserFactory.create(max_public_projects=0)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "public-project-without-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": False
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 400
assert "can't have more public projects" in response.data["_error_message"]
assert Project.objects.filter(slug="public-project-without-slots").count() == 0
def test_valid_dump_import_without_enough_private_projects_slots(client):
user = f.UserFactory.create(max_private_projects=0)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "private-project-without-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": True
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 400
assert "can't have more private projects" in response.data["_error_message"]
assert Project.objects.filter(slug="private-project-without-slots").count() == 0
def test_valid_dump_import_without_enough_membership_private_project_slots_one_project(client):
user = f.UserFactory.create(max_members_private_projects=5)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "project-without-memberships-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": True,
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
},
{
"email": "test5@test.com",
"role": "Role",
},
{
"email": "test6@test.com",
"role": "Role",
},
],
"roles": [{"name": "Role"}]
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 400
assert "reached the limit of memberships for private" in response.data["_error_message"]
assert Project.objects.filter(slug="project-without-memberships-slots").count() == 0
def test_valid_dump_import_without_enough_membership_public_project_slots_one_project(client):
user = f.UserFactory.create(max_members_public_projects=5)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "project-without-memberships-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": False,
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
},
{
"email": "test5@test.com",
"role": "Role",
},
{
"email": "test6@test.com",
"role": "Role",
},
],
"roles": [{"name": "Role"}]
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 400
assert "reached the limit of memberships for public" in response.data["_error_message"]
assert Project.objects.filter(slug="project-without-memberships-slots").count() == 0
def test_valid_dump_import_with_enough_membership_private_project_slots_multiple_projects(client, settings):
settings.CELERY_ENABLED = False
user = f.UserFactory.create(max_members_private_projects=10)
project = f.ProjectFactory.create(owner=user)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "project-without-memberships-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": True,
"roles": [{"name": "Role"}],
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
},
{
"email": "test5@test.com",
"role": "Role",
},
{
"email": "test6@test.com",
"role": "Role",
}
]
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 201
response_data = response.data
assert "id" in response_data
assert response_data["name"] == "Valid project"
def test_valid_dump_import_with_enough_membership_public_project_slots_multiple_projects(client, settings):
settings.CELERY_ENABLED = False
user = f.UserFactory.create(max_members_public_projects=10)
project = f.ProjectFactory.create(owner=user)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
f.MembershipFactory.create(project=project)
client.login(user)
url = reverse("importer-load-dump")
data = ContentFile(bytes(json.dumps({
"slug": "project-without-memberships-slots",
"name": "Valid project",
"description": "Valid project desc",
"is_private": False,
"roles": [{"name": "Role"}],
"memberships": [
{
"email": "test1@test.com",
"role": "Role",
},
{
"email": "test2@test.com",
"role": "Role",
},
{
"email": "test3@test.com",
"role": "Role",
},
{
"email": "test4@test.com",
"role": "Role",
},
{
"email": "test5@test.com",
"role": "Role",
},
{
"email": "test6@test.com",
"role": "Role",
}
]
}), "utf-8"))
data.name = "test"
response = client.post(url, {'dump': data})
assert response.status_code == 201
response_data = response.data
assert "id" in response_data
assert response_data["name"] == "Valid project"

View File

@ -53,6 +53,112 @@ def test_api_create_bulk_members(client):
assert response.data[1]["email"] == joseph.email assert response.data[1]["email"] == joseph.email
def test_api_create_bulk_members_without_enough_memberships_private_project_slots_one_project(client):
user = f.UserFactory.create(max_members_private_projects=3)
project = f.ProjectFactory(owner=user, is_private=True)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
url = reverse("memberships-bulk-create")
data = {
"project_id": project.id,
"bulk_memberships": [
{"role_id": role.pk, "email": "test1@test.com"},
{"role_id": role.pk, "email": "test2@test.com"},
{"role_id": role.pk, "email": "test3@test.com"},
{"role_id": role.pk, "email": "test4@test.com"},
]
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "reached the limit of memberships for private" in response.data["_error_message"]
def test_api_create_bulk_members_with_enough_memberships_private_project_slots_multiple_projects(client):
user = f.UserFactory.create(max_members_private_projects=6)
project = f.ProjectFactory(owner=user, is_private=True)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
other_project = f.ProjectFactory(owner=user)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
url = reverse("memberships-bulk-create")
data = {
"project_id": project.id,
"bulk_memberships": [
{"role_id": role.pk, "email": "test1@test.com"},
{"role_id": role.pk, "email": "test2@test.com"},
{"role_id": role.pk, "email": "test3@test.com"},
{"role_id": role.pk, "email": "test4@test.com"},
]
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 200
def test_api_create_bulk_members_without_enough_memberships_public_project_slots_one_project(client):
user = f.UserFactory.create(max_members_public_projects=3)
project = f.ProjectFactory(owner=user, is_private=False)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
url = reverse("memberships-bulk-create")
data = {
"project_id": project.id,
"bulk_memberships": [
{"role_id": role.pk, "email": "test1@test.com"},
{"role_id": role.pk, "email": "test2@test.com"},
{"role_id": role.pk, "email": "test3@test.com"},
{"role_id": role.pk, "email": "test4@test.com"},
]
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "reached the limit of memberships for public" in response.data["_error_message"]
def test_api_create_bulk_members_with_enough_memberships_public_project_slots_multiple_projects(client):
user = f.UserFactory.create(max_members_public_projects=6)
project = f.ProjectFactory(owner=user, is_private=False)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
other_project = f.ProjectFactory(owner=user)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
url = reverse("memberships-bulk-create")
data = {
"project_id": project.id,
"bulk_memberships": [
{"role_id": role.pk, "email": "test1@test.com"},
{"role_id": role.pk, "email": "test2@test.com"},
{"role_id": role.pk, "email": "test3@test.com"},
{"role_id": role.pk, "email": "test4@test.com"},
]
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 200
def test_api_create_bulk_members_with_extra_text(client, outbox): def test_api_create_bulk_members_with_extra_text(client, outbox):
project = f.ProjectFactory() project = f.ProjectFactory()
tester = f.RoleFactory(project=project, name="Tester") tester = f.RoleFactory(project=project, name="Tester")
@ -162,6 +268,76 @@ def test_api_create_membership(client):
assert response.data["user_email"] == user.email assert response.data["user_email"] == user.email
def test_api_create_membership_without_enough_memberships_private_project_slots_one_projects(client):
user = f.UserFactory.create(max_members_private_projects=1)
project = f.ProjectFactory(owner=user, is_private=True)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
client.login(user)
url = reverse("memberships-list")
data = {"role": role.pk, "project": project.pk, "email": "test@test.com"}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "reached the limit of memberships for private" in response.data["_error_message"]
def test_api_create_membership_with_enough_memberships_private_project_slots_multiple_projects(client):
user = f.UserFactory.create(max_members_private_projects=5)
project = f.ProjectFactory(owner=user, is_private=True)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
other_project = f.ProjectFactory(owner=user)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
client.login(user)
url = reverse("memberships-list")
data = {"role": role.pk, "project": project.pk, "email": "test@test.com"}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 201
def test_api_create_membership_without_enough_memberships_public_project_slots_one_projects(client):
user = f.UserFactory.create(max_members_public_projects=1)
project = f.ProjectFactory(owner=user, is_private=False)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
client.login(user)
url = reverse("memberships-list")
data = {"role": role.pk, "project": project.pk, "email": "test@test.com"}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "reached the limit of memberships for public" in response.data["_error_message"]
def test_api_create_membership_with_enough_memberships_public_project_slots_multiple_projects(client):
user = f.UserFactory.create(max_members_public_projects=5)
project = f.ProjectFactory(owner=user, is_private=False)
role = f.RoleFactory(project=project, name="Test")
f.MembershipFactory(project=project, user=user, is_owner=True)
other_project = f.ProjectFactory(owner=user)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
f.MembershipFactory.create(project=other_project)
client.login(user)
url = reverse("memberships-list")
data = {"role": role.pk, "project": project.pk, "email": "test@test.com"}
response = client.json.post(url, json.dumps(data))
assert response.status_code == 201
def test_api_edit_membership(client): def test_api_edit_membership(client):
membership = f.MembershipFactory(is_owner=True) membership = f.MembershipFactory(is_owner=True)
client.login(membership.user) client.login(membership.user)

View File

@ -43,6 +43,145 @@ def test_create_project(client):
assert response.status_code == 201 assert response.status_code == 201
def test_create_private_project_without_enough_private_projects_slots(client):
user = f.create_user(max_private_projects=0)
url = reverse("projects-list")
data = {
"name": "project name",
"description": "project description",
"is_private": True
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more private projects" in response.data["_error_message"]
def test_create_public_project_without_enough_public_projects_slots(client):
user = f.create_user(max_public_projects=0)
url = reverse("projects-list")
data = {
"name": "project name",
"description": "project description",
"is_private": False
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more public projects" in response.data["_error_message"]
def test_change_project_from_private_to_public_without_enough_public_projects_slots(client):
project = f.create_project(is_private=True, owner__max_public_projects=0)
f.MembershipFactory(user=project.owner, project=project, is_owner=True)
url = reverse("projects-detail", kwargs={"pk": project.pk})
data = {
"is_private": False
}
client.login(project.owner)
response = client.json.patch(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more public projects" in response.data["_error_message"]
def test_change_project_from_public_to_private_without_enough_private_projects_slots(client):
project = f.create_project(is_private=False, owner__max_private_projects=0)
f.MembershipFactory(user=project.owner, project=project, is_owner=True)
url = reverse("projects-detail", kwargs={"pk": project.pk})
data = {
"is_private": True
}
client.login(project.owner)
response = client.json.patch(url, json.dumps(data))
assert response.status_code == 400
assert "can't have more private projects" in response.data["_error_message"]
def test_create_private_project_with_enough_private_projects_slots(client):
user = f.create_user(max_private_projects=1)
url = reverse("projects-list")
data = {
"name": "project name",
"description": "project description",
"is_private": True
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 201
def test_create_public_project_with_enough_public_projects_slots(client):
user = f.create_user(max_public_projects=1)
url = reverse("projects-list")
data = {
"name": "project name",
"description": "project description",
"is_private": False
}
client.login(user)
response = client.json.post(url, json.dumps(data))
assert response.status_code == 201
def test_change_project_from_private_to_public_with_enough_public_projects_slots(client):
project = f.create_project(is_private=True, owner__max_public_projects=1)
f.MembershipFactory(user=project.owner, project=project, is_owner=True)
url = reverse("projects-detail", kwargs={"pk": project.pk})
data = {
"is_private": False
}
client.login(project.owner)
response = client.json.patch(url, json.dumps(data))
assert response.status_code == 200
def test_change_project_from_public_to_private_with_enough_private_projects_slots(client):
project = f.create_project(is_private=False, owner__max_private_projects=1)
f.MembershipFactory(user=project.owner, project=project, is_owner=True)
url = reverse("projects-detail", kwargs={"pk": project.pk})
data = {
"is_private": True
}
client.login(project.owner)
response = client.json.patch(url, json.dumps(data))
assert response.status_code == 200
def test_change_project_other_data_with_enough_private_projects_slots(client):
project = f.create_project(is_private=True, owner__max_private_projects=1)
f.MembershipFactory(user=project.owner, project=project, is_owner=True)
url = reverse("projects-detail", kwargs={"pk": project.pk})
data = {
"name": "test-project-change"
}
client.login(project.owner)
response = client.json.patch(url, json.dumps(data))
assert response.status_code == 200
def test_partially_update_project(client): def test_partially_update_project(client):
project = f.create_project() project = f.create_project()
f.MembershipFactory(user=project.owner, project=project, is_owner=True) f.MembershipFactory(user=project.owner, project=project, is_owner=True)