692429879

Build Type

github

Committed by Saurav-Shrivastav

Commit Message

[change] Additional validation for checking if the email already exists (with a different case) #227

Pull Request Pull Request #234: [change] Make email lowercase when adding new user #227

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

2 existing lines in 1 file now uncovered.

888 of 913 relevant lines covered (97.26%)

3.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.57

/openwisp_users/base/models.py

import logging
import uuid

import django
from allauth.account.models import EmailAddress
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AbstractUser as BaseUser
from django.contrib.auth.models import UserManager as BaseUserManager
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.db import models
from django.utils.functional import cached_property
from django.utils.translation import ugettext_lazy as _
from phonenumber_field.modelfields import PhoneNumberField
from swapper import load_model

logger = logging.getLogger(__name__)


class UserManager(BaseUserManager):
    def _create_user(self, *args, **kwargs):
        """
        adds automatic email address object creation to django
        management commands "create_user" and "create_superuser"
        """
        user = super()._create_user(*args, **kwargs)
        self._create_email(user)
        return user

    def _create_email(self, user):
        """
        creates verified and primary email address objects
        """
        if user.email:
            set_primary = (
                EmailAddress.objects.filter(user=user, primary=True).count() == 0
            )
            email = EmailAddress.objects.create(
                user=user, email=user.email, verified=True
            )
            if set_primary:
                email.set_as_primary()


class AbstractUser(BaseUser):
    """
    OpenWISP User model
    """

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    email = models.EmailField(_('email address'), unique=True, blank=True, null=True)
    bio = models.TextField(_('bio'), blank=True)
    url = models.URLField(_('URL'), blank=True)
    company = models.CharField(_('company'), max_length=30, blank=True)
    location = models.CharField(_('location'), max_length=256, blank=True)
    phone_number = PhoneNumberField(
        _('phone number'), unique=True, blank=True, null=True
    )
    birth_date = models.DateField(_('birth date'), blank=True, null=True)
    notes = models.TextField(
        _('notes'), help_text=_('notes for internal usage'), blank=True
    )

    objects = UserManager()

    class Meta(BaseUser.Meta):
        abstract = True
        index_together = ('id', 'email')

    @staticmethod
    def _get_pk(obj):
        """ meant for internal usage only """
        if isinstance(obj, (uuid.UUID, str)):
            pk = obj
        elif isinstance(obj, BaseOrganization):
            pk = obj.pk
        elif obj is None:
            return None
        else:
            raise ValueError('expected UUID, str or Organization instance')
        return str(pk)

    def is_member(self, organization):
        return self._get_pk(organization) in self.organizations_dict

    def is_manager(self, organization):
        org_dict = self.organizations_dict.get(self._get_pk(organization))
        return org_dict is not None and (org_dict['is_admin'] or org_dict['is_owner'])

    def is_owner(self, organization):
        org_dict = self.organizations_dict.get(self._get_pk(organization))
        return org_dict is not None and org_dict['is_owner']

    @cached_property
    def is_owner_of_any_organization(self):
        for value in self.organizations_dict.values():
            if value['is_owner']:
                return True
        return False

    @property
    def organizations_dict(self):
        """
        Returns a dictionary which represents the organizations which
        the user is member of, or which the user manages or owns.
        """
        cache_key = 'user_{}_organizations'.format(self.pk)
        organizations = cache.get(cache_key)
        if organizations is not None:
            return organizations

        manager = load_model('openwisp_users', 'OrganizationUser').objects
        org_users = manager.filter(
            user=self, organization__is_active=True
        ).select_related('organization', 'organizationowner')

        organizations = {}
        for org_user in org_users:
            org = org_user.organization
            org_id = str(org.pk)
            organizations[org_id] = {
                'is_admin': org_user.is_admin,
                'is_owner': hasattr(org_user, 'organizationowner'),
            }

        cache.set(cache_key, organizations, 86400 * 2)  # Cache for two days
        return organizations

    def __get_orgs(self, attribute):
        org_list = []
        for org_pk, options in self.organizations_dict.items():
            if options[attribute]:
                org_list.append(org_pk)
        return org_list

    @cached_property
    def organizations_managed(self):
        return self.__get_orgs('is_admin')

    @cached_property
    def organizations_owned(self):
        return self.__get_orgs('is_owner')

    def clean(self):
        email = self.email
        if email == '':
            self.email = None
        if self.phone_number == '':
            self.phone_number = None
        if (
            email
            and get_user_model()
            .objects.filter(email__iexact=email)
            .exclude(pk=self.pk)
            .exists()
        ):
            raise ValidationError(
                {'email': _('User with this Email address already exists.')}
            )

    @property
    def permissions(self):
        """
        Returns the user permissions from the cache, if the cache is
        empty it will call self.get_all_permissions() and cache the result
        """
        cache_key = f'user_{self.pk}_permissions'
        permissions = cache.get(cache_key)
        if permissions is not None:
            return permissions
        permissions = self.get_all_permissions()
        cache.set(cache_key, permissions)
        return permissions

    def has_permission(self, permission):
        if self.is_superuser:
            return True
        return permission in self.permissions


# on django 3.1, the max length of first_name has been changed, we need to
# backport this change to the previous versions to avoid migration issues
# TODO: remove this when support to django < 3.1 is dropped
if django.VERSION < (3, 1):
    first_name = AbstractUser._meta.get_field('first_name')
    first_name.max_length = 150


class BaseGroup(object):
    """
    Proxy model used to move ``GroupAdmin``
    under the same app label as the other models
    """

    class Meta:
        proxy = True
        verbose_name = _('group')
        verbose_name_plural = _('groups')


class BaseOrganization(models.Model):
    """
    OpenWISP Organization model
    """

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    description = models.TextField(_('description'), blank=True)
    email = models.EmailField(_('email'), blank=True)
    url = models.URLField(_('URL'), blank=True)

    def __str__(self):
        value = self.name
        if not self.is_active:
            value = '{0} ({1})'.format(value, _('disabled'))
        return value

    class Meta:
        abstract = True


class BaseOrganizationUser(models.Model):
    """
    OpenWISP OrganizationUser model
    """

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

    class Meta:
        abstract = True

    def clean(self):
        if self.user.is_owner(self.organization_id) and not self.is_admin:
            raise ValidationError(
                _(
                    f'{self.user.username} is the owner of the organization: '
                    f'{self.organization}, and cannot be downgraded'
                )
            )


class BaseOrganizationOwner(models.Model):
    """
    OpenWISP OrganizationOwner model
    """

    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

    def clean(self):
        if self.organization_user.organization.pk != self.organization.pk:
            raise ValidationError(
                {
                    'organization_user': _(
                        'The selected user is not member of this organization.'
                    )
                }
            )

    class Meta:
        abstract = True

1	import logging	4✔
2	import uuid	4✔
3
4	import django	4✔
5	from allauth.account.models import EmailAddress	4✔
6	from django.contrib.auth import get_user_model	4✔
7	from django.contrib.auth.models import AbstractUser as BaseUser	4✔
8	from django.contrib.auth.models import UserManager as BaseUserManager	4✔
9	from django.core.cache import cache	4✔
10	from django.core.exceptions import ValidationError	4✔
11	from django.db import models	4✔
12	from django.utils.functional import cached_property	4✔
13	from django.utils.translation import ugettext_lazy as _	4✔
14	from phonenumber_field.modelfields import PhoneNumberField	4✔
15	from swapper import load_model	4✔
16
17	logger = logging.getLogger(__name__)	4✔
18
19
20	class UserManager(BaseUserManager):	4✔
21	def _create_user(self, args, *kwargs):	4✔
22	"""
23	adds automatic email address object creation to django
24	management commands "create_user" and "create_superuser"
25	"""
26	user = super()._create_user(args, *kwargs)	4✔
27	self._create_email(user)	4✔
28	return user	4✔
29
30	def _create_email(self, user):	4✔
31	"""
32	creates verified and primary email address objects
33	"""
34	if user.email:	4✔
35	set_primary = (	4✔
36	EmailAddress.objects.filter(user=user, primary=True).count() == 0
37	)
38	email = EmailAddress.objects.create(	4✔
39	user=user, email=user.email, verified=True
40	)
41	if set_primary:	4✔
42	email.set_as_primary()	4✔
43
44
45	class AbstractUser(BaseUser):	4✔
46	"""
47	OpenWISP User model
48	"""
49
50	id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)	4✔
51	email = models.EmailField(_('email address'), unique=True, blank=True, null=True)	4✔
52	bio = models.TextField(_('bio'), blank=True)	4✔
53	url = models.URLField(_('URL'), blank=True)	4✔
54	company = models.CharField(_('company'), max_length=30, blank=True)	4✔
55	location = models.CharField(_('location'), max_length=256, blank=True)	4✔
56	phone_number = PhoneNumberField(	4✔
57	_('phone number'), unique=True, blank=True, null=True
58	)
59	birth_date = models.DateField(_('birth date'), blank=True, null=True)	4✔
60	notes = models.TextField(	4✔
61	_('notes'), help_text=_('notes for internal usage'), blank=True
62	)
63
64	objects = UserManager()	4✔
65
66	class Meta(BaseUser.Meta):	4✔
67	abstract = True	4✔
68	index_together = ('id', 'email')	4✔
69
70	@staticmethod	4✔
71	def _get_pk(obj):
72	""" meant for internal usage only """
73	if isinstance(obj, (uuid.UUID, str)):	4✔
74	pk = obj	4✔
75	elif isinstance(obj, BaseOrganization):	4✔
76	pk = obj.pk	4✔
77	elif obj is None:	4✔
78	return None	4✔
79	else:
80	raise ValueError('expected UUID, str or Organization instance')	4✔
81	return str(pk)	4✔
82
83	def is_member(self, organization):	4✔
84	return self._get_pk(organization) in self.organizations_dict	4✔
85
86	def is_manager(self, organization):	4✔
87	org_dict = self.organizations_dict.get(self._get_pk(organization))	4✔
88	return org_dict is not None and (org_dict['is_admin'] or org_dict['is_owner'])	4✔
89
90	def is_owner(self, organization):	4✔
91	org_dict = self.organizations_dict.get(self._get_pk(organization))	4✔
92	return org_dict is not None and org_dict['is_owner']	4✔
93
94	@cached_property	4✔
95	def is_owner_of_any_organization(self):
96	for value in self.organizations_dict.values():	4✔
97	if value['is_owner']:	4✔
98	return True	4✔
99	return False	4✔
100
101	@property	4✔
102	def organizations_dict(self):
103	"""
104	Returns a dictionary which represents the organizations which
105	the user is member of, or which the user manages or owns.
106	"""
107	cache_key = 'user_{}_organizations'.format(self.pk)	4✔
108	organizations = cache.get(cache_key)	4✔
109	if organizations is not None:	4✔
110	return organizations	4✔
111
112	manager = load_model('openwisp_users', 'OrganizationUser').objects	4✔
113	org_users = manager.filter(	4✔
114	user=self, organization__is_active=True
115	).select_related('organization', 'organizationowner')
116
117	organizations = {}	4✔
118	for org_user in org_users:	4✔
119	org = org_user.organization	4✔
120	org_id = str(org.pk)	4✔
121	organizations[org_id] = {	4✔
122	'is_admin': org_user.is_admin,
123	'is_owner': hasattr(org_user, 'organizationowner'),
124	}
125
126	cache.set(cache_key, organizations, 86400 * 2) # Cache for two days	4✔
127	return organizations	4✔
128
129	def __get_orgs(self, attribute):	4✔
130	org_list = []	4✔
131	for org_pk, options in self.organizations_dict.items():	4✔
132	if options[attribute]:	4✔
133	org_list.append(org_pk)	4✔
134	return org_list	4✔
135
136	@cached_property	4✔
137	def organizations_managed(self):
138	return self.__get_orgs('is_admin')	4✔
139
140	@cached_property	4✔
141	def organizations_owned(self):
142	return self.__get_orgs('is_owner')	4✔
143
144	def clean(self):	4✔
145	email = self.email	4✔
146	if email == '':	4✔
147	self.email = None	4✔
148	if self.phone_number == '':	4✔
149	self.phone_number = None	4✔
150	if (	4✔
151	email
152	and get_user_model()
153	.objects.filter(email__iexact=email)
154	.exclude(pk=self.pk)
155	.exists()
156	):
157	raise ValidationError(	4✔
158	{'email': _('User with this Email address already exists.')}
159	)
160
161	@property	4✔
162	def permissions(self):
163	"""
164	Returns the user permissions from the cache, if the cache is
165	empty it will call self.get_all_permissions() and cache the result
166	"""
167	cache_key = f'user_{self.pk}_permissions'	4✔
168	permissions = cache.get(cache_key)	4✔
169	if permissions is not None:	4✔
170	return permissions	4✔
171	permissions = self.get_all_permissions()	4✔
172	cache.set(cache_key, permissions)	4✔
173	return permissions	4✔
174
175	def has_permission(self, permission):	4✔
176	if self.is_superuser:	4✔
177	return True	4✔
178	return permission in self.permissions	4✔
179
180
181	# on django 3.1, the max length of first_name has been changed, we need to
182	# backport this change to the previous versions to avoid migration issues
183	# TODO: remove this when support to django < 3.1 is dropped
184	if django.VERSION < (3, 1):	4✔
UNCOV 185	first_name = AbstractUser._meta.get_field('first_name')	×
UNCOV 186	first_name.max_length = 150	×
187
188
189	class BaseGroup(object):	4✔
190	"""
191	Proxy model used to move ``GroupAdmin``
192	under the same app label as the other models
193	"""
194
195	class Meta:	4✔
196	proxy = True	4✔
197	verbose_name = _('group')	4✔
198	verbose_name_plural = _('groups')	4✔
199
200
201	class BaseOrganization(models.Model):	4✔
202	"""
203	OpenWISP Organization model
204	"""
205
206	id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)	4✔
207	description = models.TextField(_('description'), blank=True)	4✔
208	email = models.EmailField(_('email'), blank=True)	4✔
209	url = models.URLField(_('URL'), blank=True)	4✔
210
211	def __str__(self):	4✔
212	value = self.name	4✔
213	if not self.is_active:	4✔
214	value = '{0} ({1})'.format(value, _('disabled'))	4✔
215	return value	4✔
216
217	class Meta:	4✔
218	abstract = True	4✔
219
220
221	class BaseOrganizationUser(models.Model):	4✔
222	"""
223	OpenWISP OrganizationUser model
224	"""
225
226	id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)	4✔
227
228	class Meta:	4✔
229	abstract = True	4✔
230
231	def clean(self):	4✔
232	if self.user.is_owner(self.organization_id) and not self.is_admin:	4✔
233	raise ValidationError(	4✔
234	_(
235	f'{self.user.username} is the owner of the organization: '
236	f'{self.organization}, and cannot be downgraded'
237	)
238	)
239
240
241	class BaseOrganizationOwner(models.Model):	4✔
242	"""
243	OpenWISP OrganizationOwner model
244	"""
245
246	id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)	4✔
247
248	def clean(self):	4✔
249	if self.organization_user.organization.pk != self.organization.pk:	4✔
250	raise ValidationError(	4✔
251	{
252	'organization_user': _(
253	'The selected user is not member of this organization.'
254	)
255	}
256	)
257
258	class Meta:	4✔
259	abstract = True	4✔

openwisp / openwisp-users / 692429879

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous