685572099

Build # 685572099

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%)

6 existing lines in 2 files now uncovered.

890 of 915 relevant lines covered (97.27%)

3.89 hits per line

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

98.53

/openwisp_users/base/models.py

import logging
import uuid

import django
from allauth.account.models import EmailAddress
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):
        if self.email == '':
            self.email = None
        if self.phone_number == '':
            self.phone_number = None

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

openwisp / openwisp-users / 685572099

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