Entity.xml_types() - Code Metrics - GEANT/met - Measure and Improve Code Quality continuously with Scrutinizer

Entity.xml_types() A
last analyzed 2018-05-28 06:56 UTC

↳ Parent: Entity

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
c	1
b	0
f	0
dl	0
loc	6
rs	9.4285
cc	2

#################################################################
#
# This Software is Open Source. See License: https://github.com/TERENA/met/blob/master/LICENSE.md
# Copyright (c) 2012, TERENA All rights reserved.
#
# This Software is based on MET v1 developed for TERENA by Yaco Sistemas, http://www.yaco.es/
# MET v2 was developed for TERENA by Tamim Ziai, DAASI International GmbH, http://www.daasi.de
# Current version of MET has been revised for performance improvements by Andrea Biancini,
# Consortium GARR, http://www.garr.it
##########################################################################

import simplejson as json

from urlparse import urlparse
from urllib import quote_plus
from datetime import datetime

from django.conf import settings
from django.core.cache import cache
from django.core.urlresolvers import reverse
from django.db import models
from django.db.models import Count
from django.db.models.query import QuerySet
from django.utils.translation import ugettext_lazy as _

from met.metadataparser.templatetags import attributemap
from met.metadataparser.xmlparser import DESCRIPTOR_TYPES_DISPLAY

from met.metadataparser.models.base import JSONField, Base
from met.metadataparser.models.entity_type import EntityType
from met.metadataparser.models.entity_federations import Entity_Federations

TOP_LENGTH = getattr(settings, "TOP_LENGTH", 5)


def update_obj(mobj, obj, attrs=None):
    for_attrs = attrs or getattr(mobj, 'all_attrs', [])
    for attrb in attrs or for_attrs:
        if (getattr(mobj, attrb, None) and
            getattr(obj, attrb, None) and
                getattr(mobj, attrb) != getattr(obj, attrb)):
            setattr(obj, attrb, getattr(mobj, attrb))


class EntityQuerySet(QuerySet):
    """
    Entity queryset.
    """

    def iterator(self):
        cached_federations = {}
        for entity in super(EntityQuerySet, self).iterator():
            if entity.file:
                continue

            federations = entity.federations.all()
            if federations:
                federation = federations[0]
            else:
                raise ValueError("Can't find entity metadata")

            for federation in federations:
                if not federation.id in cached_federations:
                    cached_federations[federation.id] = federation

                cached_federation = cached_federations[federation.id]
                try:
                    entity.load_metadata(federation=cached_federation)
                except ValueError:
                    # Allow entity in federation but not in federation file
                    continue
                else:
                    break

            yield entity


class EntityManager(models.Manager):
    """
    Manager for entities.
    """

    def get_queryset(self):
        return EntityQuerySet(self.model, using=self._db)


class Entity(Base):
    """
    Model describin a federation entity (IdP, SP or AA).
    """

    READABLE_PROTOCOLS = {
        'urn:oasis:names:tc:SAML:1.1:protocol': 'SAML 1.1',
        'urn:oasis:names:tc:SAML:2.0:protocol': 'SAML 2.0',
        'urn:mace:shibboleth:1.0': 'Shiboleth 1.0',
    }

    entityid = models.CharField(blank=False, max_length=200, unique=True,
                                verbose_name=_(u'EntityID'), db_index=True)

    federations = models.ManyToManyField('Federation', through='Entity_Federations',
                                         verbose_name=_(u'Federations'))

    types = models.ManyToManyField('EntityType', verbose_name=_(u'Type'))

    name = JSONField(blank=True, null=True, max_length=2000,
                     verbose_name=_(u'Display Name'))

    certstats = models.CharField(blank=True, null=True, max_length=200,
                                 unique=False, verbose_name=_(u'Certificate Stats'))

    _display_protocols = models.CharField(blank=True, null=True, max_length=300,		
                                          unique=False, verbose_name=_(u'Display Protocols'))

    objects = models.Manager()

    longlist = EntityManager()

    curfed = None

    @property
    def certificates(self):
        return json.loads(self.certstats)

    @property
    def registration_authority_xml(self):
        return self._get_property('registration_authority')

    @property
    def registration_policy(self):
        return self._get_property('registration_policy')

    @property
    def registration_instant(self):
        reginstant = self._get_property('registration_instant')
        if reginstant is None:
            return None
        reginstant = "%sZ" % reginstant[0:19]
        return datetime.strptime(reginstant, '%Y-%m-%dT%H:%M:%SZ')

    @property
    def protocols(self):
        try:
            return ' '.join(self._get_property('protocols'))
        except Exception:
            return ''

    @property
    def languages(self):
        try:
            return ' '.join(self._get_property('languages'))
        except Exception:
            return ''

    @property
    def scopes(self):
        try:
            return ' '.join(self._get_property('scopes'))
        except Exception:
            return ''

    @property
    def attributes(self):
        try:
            attributes = self._get_property('attr_requested')
            if not attributes:
                return []
            return attributes['required']
        except Exception:
            return []

    @property
    def attributes_optional(self):
        try:
            attributes = self._get_property('attr_requested')
            if not attributes:
                return []
            return attributes['optional']
        except Exception:
            return []

    @property
    def organization(self):
        organization = self._get_property('organization')
        if not organization:
            return []

        vals = []
        for lang, data in organization.items():
            data['lang'] = lang
            vals.append(data)

        return vals

    @property
    def display_name(self):
        try:
            return self._get_property('displayName')
        except Exception:
            return ''

    @property
    def federations_count(self):
        try:
            return str(self.federations.all().count())
        except Exception:
            return ''

    @property
    def description(self):
        try:
            return self._get_property('description')
        except Exception:
            return ''

    @property
    def info_url(self):
        try:
            return self._get_property('infoUrl')
        except Exception:
            return ''

    @property
    def privacy_url(self):
        try:
            return self._get_property('privacyUrl')
        except Exception:
            return ''

    @property
    def xml(self):
        try:
            return self._get_property('xml')
        except Exception:
            return ''

    @property
    def xml_types(self):
        try:
            return self._get_property('entity_types')
        except Exception:
            return []

    @property
    def entity_categories(self):
        if self.curfed:
            efed = Entity_Federations.objects.get_or_create(federation=self.curfed, entity=self)[0]
            return efed.entity_categories
        return None

    @property
    def xml_categories(self):
        try:
            return self._get_property('entity_categories')
        except Exception:
            return []

    @property
    def display_protocols(self):
        protocols = []

        xml_protocols = self.protocols
        if xml_protocols:
            for proto in xml_protocols.split(' '):
                protocols.append(self.READABLE_PROTOCOLS.get(proto, proto))

        return protocols

    def display_attributes(self):
        attributes = {}
        for [attr, friendly] in self.attributes:
            if friendly:
                attributes[attr] = friendly
            elif attr in attributemap.MAP['fro']:
                attributes[attr] = attributemap.MAP['fro'][attr]
            else:
                attributes[attr] = '?'
        return attributes

    def display_attributes_optional(self):
        attributes = {}
        for [attr, friendly] in self.attributes_optional:
            if friendly:
                attributes[attr] = friendly
            elif attr in attributemap.MAP['fro']:
                attributes[attr] = attributemap.MAP['fro'][attr]
            else:
                attributes[attr] = '?'
        return attributes

    @property
    def contacts(self):
        contacts = []
        for cur_contact in self._get_property('contacts'):
            if cur_contact['name'] and cur_contact['surname']:
                contact_name = '%s %s' % (
                    cur_contact['name'], cur_contact['surname'])
            elif cur_contact['name']:
                contact_name = cur_contact['name']
            elif cur_contact['surname']:
                contact_name = cur_contact['surname']
            else:
                contact_name = urlparse(
                    cur_contact['email']).path.partition('?')[0]
            c_type = 'undefined'
            if cur_contact['type']:
                c_type = cur_contact['type']
            contacts.append(
                {'name': contact_name, 'email': cur_contact['email'], 'type': c_type})
        return contacts

    @property
    def logos(self):
        logos = []
        for cur_logo in self._get_property('logos'):
            cur_logo['external'] = True
            logos.append(cur_logo)

        return logos

    class Meta(object):
        verbose_name = _(u'Entity')
        verbose_name_plural = _(u'Entities')

    def __unicode__(self):
        return self.entityid

    def load_metadata(self, federation=None, entity_data=None):
        if hasattr(self, '_entity_cached'):
            return

        if self.file:
            self._entity_cached = self.load_file().get_entity(self.entityid)
        elif federation:
            self._entity_cached = federation.get_entity_metadata(self.entityid)
        elif entity_data:
            self._entity_cached = entity_data
        else:
            right_fed = None
            first_fed = None
            for fed in self.federations.all():
                if fed.registration_authority == self.registration_authority:
                    right_fed = fed
                if first_fed is None:
                    first_fed = fed

            if right_fed is not None:
                entity_cached = right_fed.get_entity_metadata(self.entityid)
                self._entity_cached = entity_cached
            else:
                entity_cached = first_fed.get_entity_metadata(self.entityid)
                self._entity_cached = entity_cached

        if not hasattr(self, '_entity_cached'):
            raise ValueError("Can't find entity metadata")

    def _get_property(self, prop, federation=None):
        try:
            self.load_metadata(federation or self.curfed)
        except ValueError:
            return None

        if hasattr(self, '_entity_cached'):
            return self._entity_cached.get(prop, None)
        else:
            raise ValueError("Not metadata loaded")

    def _get_or_create_etypes(self, cached_entity_types):

        entity_types = []
        cur_cached_types = [t.xmlname for t in self.types.all()]
        for etype in self.xml_types:
            if etype in cur_cached_types:
                break

            if cached_entity_types is None:
                entity_type, _ = EntityType.objects.get_or_create(xmlname=etype,
                                                                  name=DESCRIPTOR_TYPES_DISPLAY[etype])
            else:
                if etype in cached_entity_types:
                    entity_type = cached_entity_types[etype]
                else:
                    entity_type = EntityType.objects.create(xmlname=etype,
                                                            name=DESCRIPTOR_TYPES_DISPLAY[etype])
            entity_types.append(entity_type)
        return entity_types

    def process_metadata(self, auto_save=True, entity_data=None, cached_entity_types=None, federation=None):
        if not entity_data:
            self.load_metadata()

        if self.entityid.lower() != entity_data.get('entityid').lower():
            raise ValueError("EntityID is not the same: %s != %s" % (
                self.entityid.lower(), entity_data.get('entityid').lower()))

        self._entity_cached = entity_data

        if self.xml_types:
            entity_types = self._get_or_create_etypes(cached_entity_types)
            if len(entity_types) > 0:
                self.types.add(*entity_types)

        newname = self._get_property('displayName')
        if newname and newname != '':
            self.name = newname

        self.certstats = self._get_property('certstats')

        newprotocols = self.protocols		
        if newprotocols and newprotocols != "":		
            self._display_protocols = newprotocols

        if str(self._get_property('registration_authority')) != '':
            self.registration_authority = self._get_property(
                'registration_authority')

        if auto_save:
            self.save()

    def to_dict(self):
        self.load_metadata()

        entity = self._entity_cached.copy()
        entity["types"] = [unicode(f) for f in self.types.all()]
        entity["federations"] = [{u"name": unicode(f), u"url": f.get_absolute_url()}
                                 for f in self.federations.all()]

        if self.registration_authority:
            entity["registration_authority"] = self.registration_authority
        if self.registration_instant:
            entity["registration_instant"] = '%s' % self.registration_instant

        if "file_id" in entity.keys():
            del entity["file_id"]
        if "entity_types" in entity.keys():
            del entity["entity_types"]

        return entity

    def display_etype(value, separator=', '):
        return separator.join([unicode(item) for item in value.all()])

    @classmethod
    def get_most_federated_entities(self, maxlength=TOP_LENGTH, cache_expire=None):
        entities = None
        if cache_expire:
            entities = cache.get("most_federated_entities")

        if not entities or len(entities) < maxlength:
            # Entities with count how many federations belongs to, and sorted
            # by most first
            ob_entities = Entity.objects.all().annotate(
                federationslength=Count("federations")).order_by("-federationslength")
            ob_entities = ob_entities.prefetch_related('types', 'federations')
            ob_entities = ob_entities[:maxlength]

            entities = []
            for entity in ob_entities:
                entities.append({
                    'entityid': entity.entityid,
                    'name': entity.name,
                    'absolute_url': entity.get_absolute_url(),
                    'types': [unicode(item) for item in entity.types.all()],
                    'federations': [(unicode(item.name), item.get_absolute_url()) for item in entity.federations.all()],
                })

        if cache_expire:
            cache.set("most_federated_entities", entities, cache_expire)

        return entities[:maxlength]

    def get_absolute_url(self):
        return reverse('entity_view', args=[quote_plus(self.entityid.encode('utf-8'))])

    def can_edit(self, user, delete):
        permission = 'delete_entity' if delete else 'change_entity'
        if user.is_superuser or (user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all()):
            return True

        for federation in self.federations.all():
            if federation.can_edit(user, False):
                return True

        return False

    def has_changed(self, entityid, name, registration_authority, certstats, display_protocols):
        if self.entityid != entityid:
            return True
        if self.name != name:
            return True
        if self.registration_authority != registration_authority:
            return True
        if self.certstats != certstats:
            return True
        if self._display_protocols != display_protocols:
            return True

        return False


1		#################################################################
2		#
3		# This Software is Open Source. See License: https://github.com/TERENA/met/blob/master/LICENSE.md
4		# Copyright (c) 2012, TERENA All rights reserved.
5		#
6		# This Software is based on MET v1 developed for TERENA by Yaco Sistemas, http://www.yaco.es/
7		# MET v2 was developed for TERENA by Tamim Ziai, DAASI International GmbH, http://www.daasi.de
8		# Current version of MET has been revised for performance improvements by Andrea Biancini,
9		# Consortium GARR, http://www.garr.it
10		##########################################################################
11
12		import simplejson as json
13
14		from urlparse import urlparse
15		from urllib import quote_plus
16		from datetime import datetime
17
18		from django.conf import settings
19		from django.core.cache import cache
20		from django.core.urlresolvers import reverse
21		from django.db import models
22		from django.db.models import Count
23		from django.db.models.query import QuerySet
24		from django.utils.translation import ugettext_lazy as _
25
26		from met.metadataparser.templatetags import attributemap
27		from met.metadataparser.xmlparser import DESCRIPTOR_TYPES_DISPLAY
28
29		from met.metadataparser.models.base import JSONField, Base
30		from met.metadataparser.models.entity_type import EntityType
31		from met.metadataparser.models.entity_federations import Entity_Federations
32
33		TOP_LENGTH = getattr(settings, "TOP_LENGTH", 5)
34
35
36		def update_obj(mobj, obj, attrs=None):
37		for_attrs = attrs or getattr(mobj, 'all_attrs', [])
38		for attrb in attrs or for_attrs:
39		if (getattr(mobj, attrb, None) and
40		getattr(obj, attrb, None) and
41		getattr(mobj, attrb) != getattr(obj, attrb)):
42		setattr(obj, attrb, getattr(mobj, attrb))
43
44
45		class EntityQuerySet(QuerySet):
46		"""
47		Entity queryset.
48		"""
49
50		def iterator(self):
51		cached_federations = {}
52		for entity in super(EntityQuerySet, self).iterator():
53		if entity.file:
54		continue
55
56		federations = entity.federations.all()
57		if federations:
58		federation = federations[0]
59		else:
60		raise ValueError("Can't find entity metadata")
61
62		for federation in federations:
63		if not federation.id in cached_federations:
64		cached_federations[federation.id] = federation
65
66		cached_federation = cached_federations[federation.id]
67		try:
68		entity.load_metadata(federation=cached_federation)
69		except ValueError:
70		# Allow entity in federation but not in federation file
71		continue
72		else:
73		break
74
75		yield entity
76
77
78		class EntityManager(models.Manager):
79		"""
80		Manager for entities.
81		"""
82
83		def get_queryset(self):
84		return EntityQuerySet(self.model, using=self._db)
85
86
87		class Entity(Base):
88		"""
89		Model describin a federation entity (IdP, SP or AA).
90		"""
91
92		READABLE_PROTOCOLS = {
93		'urn:oasis:names:tc:SAML:1.1:protocol': 'SAML 1.1',
94		'urn:oasis:names:tc:SAML:2.0:protocol': 'SAML 2.0',
95		'urn:mace:shibboleth:1.0': 'Shiboleth 1.0',
96		}
97
98		entityid = models.CharField(blank=False, max_length=200, unique=True,
99		verbose_name=_(u'EntityID'), db_index=True)
100
101		federations = models.ManyToManyField('Federation', through='Entity_Federations',
102		verbose_name=_(u'Federations'))
103
104		types = models.ManyToManyField('EntityType', verbose_name=_(u'Type'))
105
106		name = JSONField(blank=True, null=True, max_length=2000,
107		verbose_name=_(u'Display Name'))
108
109		certstats = models.CharField(blank=True, null=True, max_length=200,
110		unique=False, verbose_name=_(u'Certificate Stats'))
111
112		_display_protocols = models.CharField(blank=True, null=True, max_length=300,
113		unique=False, verbose_name=_(u'Display Protocols'))
114
115		objects = models.Manager()
116
117		longlist = EntityManager()
118
119		curfed = None
120
121		@property
122		def certificates(self):
123		return json.loads(self.certstats)
124
125		@property
126		def registration_authority_xml(self):
127		return self._get_property('registration_authority')
128
129		@property
130		def registration_policy(self):
131		return self._get_property('registration_policy')
132
133		@property
134		def registration_instant(self):
135		reginstant = self._get_property('registration_instant')
136		if reginstant is None:
137		return None
138		reginstant = "%sZ" % reginstant[0:19]
139		return datetime.strptime(reginstant, '%Y-%m-%dT%H:%M:%SZ')
140
141		@property
142		def protocols(self):
143		try:
144		return ' '.join(self._get_property('protocols'))
145		except Exception:
146		return ''
147
148		@property
149		def languages(self):
150		try:
151		return ' '.join(self._get_property('languages'))
152		except Exception:
153		return ''
154
155		@property
156		def scopes(self):
157		try:
158		return ' '.join(self._get_property('scopes'))
159		except Exception:
160		return ''
161
162		@property
163		def attributes(self):
164		try:
165		attributes = self._get_property('attr_requested')
166		if not attributes:
167		return []
168		return attributes['required']
169		except Exception:
170		return []
171
172		@property
173		def attributes_optional(self):
174		try:
175		attributes = self._get_property('attr_requested')
176		if not attributes:
177		return []
178		return attributes['optional']
179		except Exception:
180		return []
181
182		@property
183		def organization(self):
184		organization = self._get_property('organization')
185		if not organization:
186		return []
187
188		vals = []
189		for lang, data in organization.items():
190		data['lang'] = lang
191		vals.append(data)
192
193		return vals
194
195		@property
196		def display_name(self):
197		try:
198		return self._get_property('displayName')
199		except Exception:
200		return ''
201
202		@property
203		def federations_count(self):
204		try:
205		return str(self.federations.all().count())
206		except Exception:
207		return ''
208
209		@property
210		def description(self):
211		try:
212		return self._get_property('description')
213		except Exception:
214		return ''
215
216		@property
217		def info_url(self):
218		try:
219		return self._get_property('infoUrl')
220		except Exception:
221		return ''
222
223		@property
224		def privacy_url(self):
225		try:
226		return self._get_property('privacyUrl')
227		except Exception:
228		return ''
229
230		@property
231		def xml(self):
232		try:
233		return self._get_property('xml')
234		except Exception:
235		return ''
236
237		@property
238		def xml_types(self):
239		try:
240		return self._get_property('entity_types')
241		except Exception:
242		return []
243
244		@property
245		def entity_categories(self):
246		if self.curfed:
247		efed = Entity_Federations.objects.get_or_create(federation=self.curfed, entity=self)[0]
248		return efed.entity_categories
249		return None
250
251		@property
252		def xml_categories(self):
253		try:
254		return self._get_property('entity_categories')
255		except Exception:
256		return []
257
258		@property
259		def display_protocols(self):
260		protocols = []
261
262		xml_protocols = self.protocols
263		if xml_protocols:
264		for proto in xml_protocols.split(' '):
265		protocols.append(self.READABLE_PROTOCOLS.get(proto, proto))
266
267		return protocols
268
269		def display_attributes(self):
270		attributes = {}
271		for [attr, friendly] in self.attributes:
272		if friendly:
273		attributes[attr] = friendly
274		elif attr in attributemap.MAP['fro']:
275		attributes[attr] = attributemap.MAP['fro'][attr]
276		else:
277		attributes[attr] = '?'
278		return attributes
279
280		def display_attributes_optional(self):
281		attributes = {}
282		for [attr, friendly] in self.attributes_optional:
283		if friendly:
284		attributes[attr] = friendly
285		elif attr in attributemap.MAP['fro']:
286		attributes[attr] = attributemap.MAP['fro'][attr]
287		else:
288		attributes[attr] = '?'
289		return attributes
290
291		@property
292		def contacts(self):
293		contacts = []
294		for cur_contact in self._get_property('contacts'):
295		if cur_contact['name'] and cur_contact['surname']:
296		contact_name = '%s %s' % (
297		cur_contact['name'], cur_contact['surname'])
298		elif cur_contact['name']:
299		contact_name = cur_contact['name']
300		elif cur_contact['surname']:
301		contact_name = cur_contact['surname']
302		else:
303		contact_name = urlparse(
304		cur_contact['email']).path.partition('?')[0]
305		c_type = 'undefined'
306		if cur_contact['type']:
307		c_type = cur_contact['type']
308		contacts.append(
309		{'name': contact_name, 'email': cur_contact['email'], 'type': c_type})
310		return contacts
311
312		@property
313		def logos(self):
314		logos = []
315		for cur_logo in self._get_property('logos'):
316		cur_logo['external'] = True
317		logos.append(cur_logo)
318
319		return logos
320
321		class Meta(object):
322		verbose_name = _(u'Entity')
323		verbose_name_plural = _(u'Entities')
324
325		def __unicode__(self):
326		return self.entityid
327
328		def load_metadata(self, federation=None, entity_data=None):
329		if hasattr(self, '_entity_cached'):
330		return
331
332		if self.file:
333		self._entity_cached = self.load_file().get_entity(self.entityid)
334		elif federation:
335		self._entity_cached = federation.get_entity_metadata(self.entityid)
336		elif entity_data:
337		self._entity_cached = entity_data
338		else:
339		right_fed = None
340		first_fed = None
341		for fed in self.federations.all():
342		if fed.registration_authority == self.registration_authority:
343		right_fed = fed
344		if first_fed is None:
345		first_fed = fed
346
347		if right_fed is not None:
348		entity_cached = right_fed.get_entity_metadata(self.entityid)
349		self._entity_cached = entity_cached
350		else:
351		entity_cached = first_fed.get_entity_metadata(self.entityid)
352		self._entity_cached = entity_cached
353
354		if not hasattr(self, '_entity_cached'):
355		raise ValueError("Can't find entity metadata")
356
357		def _get_property(self, prop, federation=None):
358		try:
359		self.load_metadata(federation or self.curfed)
360		except ValueError:
361		return None
362
363		if hasattr(self, '_entity_cached'):
364		return self._entity_cached.get(prop, None)
365		else:
366		raise ValueError("Not metadata loaded")
367
368	View Code Duplication	def _get_or_create_etypes(self, cached_entity_types):
		0 ignored issues – show Duplication introduced 2017-10-21 08:27 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
369		entity_types = []
370		cur_cached_types = [t.xmlname for t in self.types.all()]
371		for etype in self.xml_types:
372		if etype in cur_cached_types:
373		break
374
375		if cached_entity_types is None:
376		entity_type, _ = EntityType.objects.get_or_create(xmlname=etype,
377		name=DESCRIPTOR_TYPES_DISPLAY[etype])
378		else:
379		if etype in cached_entity_types:
380		entity_type = cached_entity_types[etype]
381		else:
382		entity_type = EntityType.objects.create(xmlname=etype,
383		name=DESCRIPTOR_TYPES_DISPLAY[etype])
384		entity_types.append(entity_type)
385		return entity_types
386
387		def process_metadata(self, auto_save=True, entity_data=None, cached_entity_types=None, federation=None):
388		if not entity_data:
389		self.load_metadata()
390
391		if self.entityid.lower() != entity_data.get('entityid').lower():
392		raise ValueError("EntityID is not the same: %s != %s" % (
393		self.entityid.lower(), entity_data.get('entityid').lower()))
394
395		self._entity_cached = entity_data
396
397		if self.xml_types:
398		entity_types = self._get_or_create_etypes(cached_entity_types)
399		if len(entity_types) > 0:
400		self.types.add(*entity_types)
401
402		newname = self._get_property('displayName')
403		if newname and newname != '':
404		self.name = newname
405
406		self.certstats = self._get_property('certstats')
407
408		newprotocols = self.protocols
409		if newprotocols and newprotocols != "":
410		self._display_protocols = newprotocols
411
412		if str(self._get_property('registration_authority')) != '':
413		self.registration_authority = self._get_property(
414		'registration_authority')
415
416		if auto_save:
417		self.save()
418
419		def to_dict(self):
420		self.load_metadata()
421
422		entity = self._entity_cached.copy()
423		entity["types"] = [unicode(f) for f in self.types.all()]
424		entity["federations"] = [{u"name": unicode(f), u"url": f.get_absolute_url()}
425		for f in self.federations.all()]
426
427		if self.registration_authority:
428		entity["registration_authority"] = self.registration_authority
429		if self.registration_instant:
430		entity["registration_instant"] = '%s' % self.registration_instant
431
432		if "file_id" in entity.keys():
433		del entity["file_id"]
434		if "entity_types" in entity.keys():
435		del entity["entity_types"]
436
437		return entity
438
439		def display_etype(value, separator=', '):
440		return separator.join([unicode(item) for item in value.all()])
441
442		@classmethod
443		def get_most_federated_entities(self, maxlength=TOP_LENGTH, cache_expire=None):
444		entities = None
445		if cache_expire:
446		entities = cache.get("most_federated_entities")
447
448		if not entities or len(entities) < maxlength:
449		# Entities with count how many federations belongs to, and sorted
450		# by most first
451		ob_entities = Entity.objects.all().annotate(
452		federationslength=Count("federations")).order_by("-federationslength")
453		ob_entities = ob_entities.prefetch_related('types', 'federations')
454		ob_entities = ob_entities[:maxlength]
455
456		entities = []
457		for entity in ob_entities:
458		entities.append({
459		'entityid': entity.entityid,
460		'name': entity.name,
461		'absolute_url': entity.get_absolute_url(),
462		'types': [unicode(item) for item in entity.types.all()],
463		'federations': [(unicode(item.name), item.get_absolute_url()) for item in entity.federations.all()],
464		})
465
466		if cache_expire:
467		cache.set("most_federated_entities", entities, cache_expire)
468
469		return entities[:maxlength]
470
471		def get_absolute_url(self):
472		return reverse('entity_view', args=[quote_plus(self.entityid.encode('utf-8'))])
473
474		def can_edit(self, user, delete):
475		permission = 'delete_entity' if delete else 'change_entity'
476		if user.is_superuser or (user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all()):
477		return True
478
479		for federation in self.federations.all():
480		if federation.can_edit(user, False):
481		return True
482
483		return False
484
485		def has_changed(self, entityid, name, registration_authority, certstats, display_protocols):
486		if self.entityid != entityid:
487		return True
488		if self.name != name:
489		return True
490		if self.registration_authority != registration_authority:
491		return True
492		if self.certstats != certstats:
493		return True
494		if self._display_protocols != display_protocols:
495		return True
496
497		return False
498

GEANT / met

Entity.xml_types() A last analyzed 2018-05-28 06:56 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Entity.xml_types() A
last analyzed 2018-05-28 06:56 UTC