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

Federation.can_edit() B
last analyzed 2018-05-28 06:56 UTC

↳ Parent: Federation

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	8
rs	8.5454
cc	5

#################################################################
# MET v2 Metadate Explorer Tool
#
# 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 pytz
import simplejson as json

from datetime import datetime, time, timedelta

from django.core.urlresolvers import reverse
from django.db import models
from django.db.models import Max
from django.db.models.signals import pre_save
from django.utils.translation import ugettext_lazy as _
from django.utils import timezone
from django.dispatch import receiver
from django.template.defaultfilters import slugify

from met.metadataparser.xmlparser import MetadataParser

from met.metadataparser.models.base import Base, XmlDescriptionError
from met.metadataparser.models.entity import Entity
from met.metadataparser.models.entity_type import EntityType
from met.metadataparser.models.entity_stat import EntityStat, stats
from met.metadataparser.models.entity_category import EntityCategory
from met.metadataparser.models.entity_federations import Entity_Federations

FEDERATION_TYPES = (
    (None, ''),
    ('hub-and-spoke', 'Hub and Spoke'),
    ('mesh', 'Full Mesh'),
)


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 Federation(Base):
    """
    Model describing an identity federation.
    """

    name = models.CharField(blank=False, null=False, max_length=200,
                            unique=True, verbose_name=_(u'Name'))

    type = models.CharField(blank=True, null=True, max_length=100,
                            unique=False, verbose_name=_(u'Type'), choices=FEDERATION_TYPES)

    url = models.URLField(verbose_name='Federation url',
                          blank=True, null=True)

    fee_schedule_url = models.URLField(verbose_name='Fee schedule url',
                                       max_length=150, blank=True, null=True)

    logo = models.ImageField(upload_to='federation_logo', blank=True,
                             null=True, verbose_name=_(u'Federation logo'))

    is_interfederation = models.BooleanField(default=False, db_index=True,
                                             verbose_name=_(u'Is interfederation'))

    slug = models.SlugField(max_length=200, unique=True)

    country = models.CharField(blank=True, null=True, max_length=100,
                               unique=False, verbose_name=_(u'Country'))

    metadata_update = models.DateTimeField(blank=True, null=True,
                                           unique=False, verbose_name=_(u'Metadata update date and time'))

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

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

    @property
    def _metadata(self):
        if not hasattr(self, '_metadata_cache'):
            self._metadata_cache = self.load_file()
        return self._metadata_cache

    def __unicode__(self):
        return self.name

    def get_entity_metadata(self, entityid):
        return self._metadata.get_entity(entityid)

    def get_entity(self, entityid):
        return self.entity_set.get(entityid=entityid)

    def process_metadata(self):
        metadata = self.load_file()

        if self.file_id and metadata.file_id and metadata.file_id == self.file_id:
            return
        else:
            self.file_id = metadata.file_id

        if not metadata:
            return
        if not metadata.is_federation:
            raise XmlDescriptionError("XML Haven't federation form")

        update_obj(metadata.get_federation(), self)
        self.certstats = MetadataParser.get_certstats(metadata.rootelem)

    def _remove_deleted_entities(self, entities_from_xml):
        removed = 0
        for entity in self.entity_set.all():
            # Remove entity relation if does not exist in metadata
            if not entity.entityid in entities_from_xml:
                Entity_Federations.objects.filter(
                    federation=self, entity=entity).delete()
                removed += 1

        return removed

    def _get_or_create_ecategories(self, entity, cached_entity_categories):

        entity_categories = []
        efed = Entity_Federations.objects.get_or_create(federation=self, entity=entity)[0]
        cur_cached_categories = [
            t.category_id for t in efed.entity_categories.all()]
        for ecategory in entity.xml_categories:
            if ecategory in cur_cached_categories:
                break

            if cached_entity_categories is None:
                entity_category, _ = EntityCategory.objects.get_or_create(
                    category_id=ecategory)
            else:
                if ecategory in cached_entity_categories:
                    entity_category = cached_entity_categories[ecategory]
                else:
                    entity_category = EntityCategory.objects.create(
                        category_id=ecategory)
            entity_categories.append(entity_category)
        return entity_categories

    def _update_entities(self, entities_to_update, entities_to_add):
        for e in entities_to_update:
            e.save()

        for e in entities_to_add:
            membership = Entity_Federations.objects.get_or_create(federation=self, entity=e)[0]
            membership.registration_instant = e.registration_instant.date() if e.registration_instant else None

            if e.xml_categories:
                db_entity_categories = EntityCategory.objects.all()
                cached_entity_categories = {
                    entity_category.category_id: entity_category for entity_category in db_entity_categories}

                # Delete categories no more present in XML
                membership.entity_categories.clear()

                # Create all entities, if not alread in database
                entity_categories = self._get_or_create_ecategories(e, cached_entity_categories)

                # Add categories to entity
                if len(entity_categories) > 0:
                    membership.entity_categories.add(*entity_categories)
            else:
                # No categories in XML, delete eventual categorie sin DB
                membership.entity_categories.clear()

            membership.save()

    def _add_new_entities(self, entities, entities_from_xml, request, federation_slug):
        db_entity_types = EntityType.objects.all()
        cached_entity_types = {
            entity_type.xmlname: entity_type for entity_type in db_entity_types}

        entities_to_add = []
        entities_to_update = []

        for m_id in entities_from_xml:
            if request and federation_slug:
                request.session['%s_cur_entities' % federation_slug] += 1
                request.session.save()

            created = False
            if m_id in entities:
                entity = entities[m_id]
            else:
                entity, created = Entity.objects.get_or_create(entityid=m_id)

            entityid = entity.entityid
            name = entity.name
            registration_authority = entity.registration_authority
            certstats = entity.certstats
            display_protocols = entity._display_protocols

            entity_from_xml = self._metadata.get_entity(m_id, False)
            entity.process_metadata(False, entity_from_xml, cached_entity_types, self)

            if created or entity.has_changed(entityid, name, registration_authority, certstats, display_protocols):
                entities_to_update.append(entity)

            entities_to_add.append(entity)

        self._update_entities(entities_to_update, entities_to_add)
        return len(entities_to_update)

    @staticmethod
    def _daterange(start_date, end_date):
        for n in range(int((end_date - start_date).days + 1)):
            yield start_date + timedelta(n)

    def compute_new_stats(self):
        if not self._metadata: return ([], [])
        entities_from_xml = self._metadata.get_entities()

        entities = Entity.objects.filter(entityid__in=entities_from_xml)
        entities = entities.prefetch_related('types')
        Entity_Federations.objects.filter(federation=self)

        try:
            first_date = EntityStat.objects.filter(
                federation=self).aggregate(Max('time'))['time__max']
            if not first_date:
                raise Exception('Not able to find statistical data in the DB.')
        except Exception:
            first_date = datetime(2010, 1, 1)
            first_date = pytz.utc.localize(first_date)

        for curtimestamp in self._daterange(first_date, timezone.now() - timedelta(1)):
            computed = {}
            not_computed = []
            entity_stats = []
            for feature in stats['features'].keys():
                fun = getattr(self, 'get_%s' % feature, None)

                if callable(fun):
                    stat = EntityStat()
                    stat.feature = feature
                    stat.time = curtimestamp
                    stat.federation = self
                    stat.value = fun(
                        entities, stats['features'][feature], curtimestamp)
                    entity_stats.append(stat)
                    computed[feature] = stat.value
                else:
                    not_computed.append(feature)

            from_time = datetime.combine(curtimestamp, time.min)
            if timezone.is_naive(from_time):
                from_time = pytz.utc.localize(from_time)
            to_time = datetime.combine(curtimestamp, time.max)
            if timezone.is_naive(to_time):
                to_time = pytz.utc.localize(to_time)

            EntityStat.objects.filter(
                federation=self, time__gte=from_time, time__lte=to_time).delete()
            EntityStat.objects.bulk_create(entity_stats)

        return (computed, not_computed)

    def process_metadata_entities(self, request=None, federation_slug=None):
        if not self._metadata: return
        entities_from_xml = self._metadata.get_entities()
        removed = self._remove_deleted_entities(entities_from_xml)

        entities = {}
        db_entities = Entity.objects.filter(entityid__in=entities_from_xml)
        db_entities = db_entities.prefetch_related('types')
        #TODO add prefetch related, federations, entity_categories

        for entity in db_entities.all():
            entities[entity.entityid] = entity

        if request and federation_slug:
            request.session['%s_num_entities' %
                            federation_slug] = len(entities_from_xml)
            request.session['%s_cur_entities' % federation_slug] = 0
            request.session['%s_process_done' % federation_slug] = False
            request.session.save()

        updated = self._add_new_entities(
            entities, entities_from_xml, request, federation_slug)

        if request and federation_slug:
            request.session['%s_process_done' % federation_slug] = True
            request.session.save()

        return removed, updated

    def get_absolute_url(self):
        return reverse('federation_view', args=[self.slug])

    @classmethod
    def get_sp(cls, entities, xml_name, ref_date=None):
        if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
            selected = entities.filter(
                types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
        else:
            selected = entities.filter(types__xmlname=xml_name)
        return len(selected)

    @classmethod
    def get_idp(cls, entities, xml_name, ref_date=None):
        if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
            selected = entities.filter(
                types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
        else:
            selected = entities.filter(types__xmlname=xml_name)
        return len(selected)

    @classmethod
    def get_aa(cls, entities, xml_name, ref_date=None):
        if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
            selected = entities.filter(
                types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
        else:
            selected = entities.filter(types__xmlname=xml_name)
        return len(selected)

    def get_sp_saml1(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)

    def get_sp_saml2(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)

    def get_sp_shib1(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)

    def get_idp_saml1(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)

    def get_idp_saml2(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)

    def get_idp_shib1(self, entities, xml_name, ref_date=None):
        return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)

    def get_stat_protocol(self, entities, xml_name, service_type, ref_date):
        if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
            selected = entities.filter(types__xmlname=service_type,
                                       _display_protocols__contains=xml_name,
                                       entity_federations__registration_instant__lt=ref_date)
        else:
            selected = entities.filter(types__xmlname=service_type,
                                       _display_protocols__contains=xml_name)

        return len(selected)

    def can_edit(self, user, delete):
        if user.is_superuser:
            return True

        permission = 'delete_federation' if delete else 'change_federation'
        if user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all():
            return True
        return False


@receiver(pre_save, sender=Federation, dispatch_uid='federation_pre_save')
def federation_pre_save(sender, instance, **kwargs):
    # Skip pre_save if only file name is saved
    if kwargs.has_key('update_fields') and kwargs['update_fields'] == set(['file']):
        return

    #slug = slugify(unicode(instance.name))[:200]
    # if instance.file_url and instance.file_url != '':
    #    try:
    #        instance.fetch_metadata_file(slug)
    #    except Exception, e:
    #        pass

    if instance.name:
        instance.slug = slugify(unicode(instance))[:200]


@receiver(pre_save, sender=Entity, dispatch_uid='entity_pre_save')
def entity_pre_save(sender, instance, **kwargs):
    # if refetch and instance.file_url:
    #    slug = slugify(unicode(instance.name))[:200]
    #    instance.fetch_metadata_file(slug)
    #    instance.process_metadata()
    pass


1		#################################################################
2		# MET v2 Metadate Explorer Tool
3		#
4		# This Software is Open Source. See License: https://github.com/TERENA/met/blob/master/LICENSE.md
5		# Copyright (c) 2012, TERENA All rights reserved.
6		#
7		# This Software is based on MET v1 developed for TERENA by Yaco Sistemas, http://www.yaco.es/
8		# MET v2 was developed for TERENA by Tamim Ziai, DAASI International GmbH, http://www.daasi.de
9		# Current version of MET has been revised for performance improvements by Andrea Biancini,
10		# Consortium GARR, http://www.garr.it
11		##########################################################################
12
13		import pytz
14		import simplejson as json
15
16		from datetime import datetime, time, timedelta
17
18		from django.core.urlresolvers import reverse
19		from django.db import models
20		from django.db.models import Max
21		from django.db.models.signals import pre_save
22		from django.utils.translation import ugettext_lazy as _
23		from django.utils import timezone
24		from django.dispatch import receiver
25		from django.template.defaultfilters import slugify
26
27		from met.metadataparser.xmlparser import MetadataParser
28
29		from met.metadataparser.models.base import Base, XmlDescriptionError
30		from met.metadataparser.models.entity import Entity
31		from met.metadataparser.models.entity_type import EntityType
32		from met.metadataparser.models.entity_stat import EntityStat, stats
33		from met.metadataparser.models.entity_category import EntityCategory
34		from met.metadataparser.models.entity_federations import Entity_Federations
35
36		FEDERATION_TYPES = (
37		(None, ''),
38		('hub-and-spoke', 'Hub and Spoke'),
39		('mesh', 'Full Mesh'),
40		)
41
42
43		def update_obj(mobj, obj, attrs=None):
44		for_attrs = attrs or getattr(mobj, 'all_attrs', [])
45		for attrb in attrs or for_attrs:
46		if (getattr(mobj, attrb, None) and
47		getattr(obj, attrb, None) and
48		getattr(mobj, attrb) != getattr(obj, attrb)):
49		setattr(obj, attrb, getattr(mobj, attrb))
50
51
52		class Federation(Base):
53		"""
54		Model describing an identity federation.
55		"""
56
57		name = models.CharField(blank=False, null=False, max_length=200,
58		unique=True, verbose_name=_(u'Name'))
59
60		type = models.CharField(blank=True, null=True, max_length=100,
61		unique=False, verbose_name=_(u'Type'), choices=FEDERATION_TYPES)
62
63		url = models.URLField(verbose_name='Federation url',
64		blank=True, null=True)
65
66		fee_schedule_url = models.URLField(verbose_name='Fee schedule url',
67		max_length=150, blank=True, null=True)
68
69		logo = models.ImageField(upload_to='federation_logo', blank=True,
70		null=True, verbose_name=_(u'Federation logo'))
71
72		is_interfederation = models.BooleanField(default=False, db_index=True,
73		verbose_name=_(u'Is interfederation'))
74
75		slug = models.SlugField(max_length=200, unique=True)
76
77		country = models.CharField(blank=True, null=True, max_length=100,
78		unique=False, verbose_name=_(u'Country'))
79
80		metadata_update = models.DateTimeField(blank=True, null=True,
81		unique=False, verbose_name=_(u'Metadata update date and time'))
82
83		certstats = models.CharField(blank=True, null=True, max_length=200,
84		unique=False, verbose_name=_(u'Certificate Stats'))
85
86		@property
87		def certificates(self):
88		return json.loads(self.certstats)
89
90		@property
91		def _metadata(self):
92		if not hasattr(self, '_metadata_cache'):
93		self._metadata_cache = self.load_file()
94		return self._metadata_cache
95
96		def __unicode__(self):
97		return self.name
98
99		def get_entity_metadata(self, entityid):
100		return self._metadata.get_entity(entityid)
101
102		def get_entity(self, entityid):
103		return self.entity_set.get(entityid=entityid)
104
105		def process_metadata(self):
106		metadata = self.load_file()
107
108		if self.file_id and metadata.file_id and metadata.file_id == self.file_id:
109		return
110		else:
111		self.file_id = metadata.file_id
112
113		if not metadata:
114		return
115		if not metadata.is_federation:
116		raise XmlDescriptionError("XML Haven't federation form")
117
118		update_obj(metadata.get_federation(), self)
119		self.certstats = MetadataParser.get_certstats(metadata.rootelem)
120
121		def _remove_deleted_entities(self, entities_from_xml):
122		removed = 0
123		for entity in self.entity_set.all():
124		# Remove entity relation if does not exist in metadata
125		if not entity.entityid in entities_from_xml:
126		Entity_Federations.objects.filter(
127		federation=self, entity=entity).delete()
128		removed += 1
129
130		return removed
131
132	View Code Duplication	def _get_or_create_ecategories(self, entity, cached_entity_categories):
		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...
133		entity_categories = []
134		efed = Entity_Federations.objects.get_or_create(federation=self, entity=entity)[0]
135		cur_cached_categories = [
136		t.category_id for t in efed.entity_categories.all()]
137		for ecategory in entity.xml_categories:
138		if ecategory in cur_cached_categories:
139		break
140
141		if cached_entity_categories is None:
142		entity_category, _ = EntityCategory.objects.get_or_create(
143		category_id=ecategory)
144		else:
145		if ecategory in cached_entity_categories:
146		entity_category = cached_entity_categories[ecategory]
147		else:
148		entity_category = EntityCategory.objects.create(
149		category_id=ecategory)
150		entity_categories.append(entity_category)
151		return entity_categories
152
153		def _update_entities(self, entities_to_update, entities_to_add):
154		for e in entities_to_update:
155		e.save()
156
157		for e in entities_to_add:
158		membership = Entity_Federations.objects.get_or_create(federation=self, entity=e)[0]
159		membership.registration_instant = e.registration_instant.date() if e.registration_instant else None
160
161		if e.xml_categories:
162		db_entity_categories = EntityCategory.objects.all()
163		cached_entity_categories = {
164		entity_category.category_id: entity_category for entity_category in db_entity_categories}
165
166		# Delete categories no more present in XML
167		membership.entity_categories.clear()
168
169		# Create all entities, if not alread in database
170		entity_categories = self._get_or_create_ecategories(e, cached_entity_categories)
171
172		# Add categories to entity
173		if len(entity_categories) > 0:
174		membership.entity_categories.add(*entity_categories)
175		else:
176		# No categories in XML, delete eventual categorie sin DB
177		membership.entity_categories.clear()
178
179		membership.save()
180
181		def _add_new_entities(self, entities, entities_from_xml, request, federation_slug):
182		db_entity_types = EntityType.objects.all()
183		cached_entity_types = {
184		entity_type.xmlname: entity_type for entity_type in db_entity_types}
185
186		entities_to_add = []
187		entities_to_update = []
188
189		for m_id in entities_from_xml:
190		if request and federation_slug:
191		request.session['%s_cur_entities' % federation_slug] += 1
192		request.session.save()
193
194		created = False
195		if m_id in entities:
196		entity = entities[m_id]
197		else:
198		entity, created = Entity.objects.get_or_create(entityid=m_id)
199
200		entityid = entity.entityid
201		name = entity.name
202		registration_authority = entity.registration_authority
203		certstats = entity.certstats
204		display_protocols = entity._display_protocols
205
206		entity_from_xml = self._metadata.get_entity(m_id, False)
207		entity.process_metadata(False, entity_from_xml, cached_entity_types, self)
208
209		if created or entity.has_changed(entityid, name, registration_authority, certstats, display_protocols):
210		entities_to_update.append(entity)
211
212		entities_to_add.append(entity)
213
214		self._update_entities(entities_to_update, entities_to_add)
215		return len(entities_to_update)
216
217		@staticmethod
218		def _daterange(start_date, end_date):
219		for n in range(int((end_date - start_date).days + 1)):
220		yield start_date + timedelta(n)
221
222		def compute_new_stats(self):
223		if not self._metadata: return ([], [])
224		entities_from_xml = self._metadata.get_entities()
225
226		entities = Entity.objects.filter(entityid__in=entities_from_xml)
227		entities = entities.prefetch_related('types')
228		Entity_Federations.objects.filter(federation=self)
229
230		try:
231		first_date = EntityStat.objects.filter(
232		federation=self).aggregate(Max('time'))['time__max']
233		if not first_date:
234		raise Exception('Not able to find statistical data in the DB.')
235		except Exception:
236		first_date = datetime(2010, 1, 1)
237		first_date = pytz.utc.localize(first_date)
238
239		for curtimestamp in self._daterange(first_date, timezone.now() - timedelta(1)):
240		computed = {}
241		not_computed = []
242		entity_stats = []
243		for feature in stats['features'].keys():
244		fun = getattr(self, 'get_%s' % feature, None)
245
246		if callable(fun):
247		stat = EntityStat()
248		stat.feature = feature
249		stat.time = curtimestamp
250		stat.federation = self
251		stat.value = fun(
252		entities, stats['features'][feature], curtimestamp)
253		entity_stats.append(stat)
254		computed[feature] = stat.value
255		else:
256		not_computed.append(feature)
257
258		from_time = datetime.combine(curtimestamp, time.min)
259		if timezone.is_naive(from_time):
260		from_time = pytz.utc.localize(from_time)
261		to_time = datetime.combine(curtimestamp, time.max)
262		if timezone.is_naive(to_time):
263		to_time = pytz.utc.localize(to_time)
264
265		EntityStat.objects.filter(
266		federation=self, time__gte=from_time, time__lte=to_time).delete()
267		EntityStat.objects.bulk_create(entity_stats)
268
269		return (computed, not_computed)
270
271		def process_metadata_entities(self, request=None, federation_slug=None):
272		if not self._metadata: return
273		entities_from_xml = self._metadata.get_entities()
274		removed = self._remove_deleted_entities(entities_from_xml)
275
276		entities = {}
277		db_entities = Entity.objects.filter(entityid__in=entities_from_xml)
278		db_entities = db_entities.prefetch_related('types')
279		#TODO add prefetch related, federations, entity_categories
280
281		for entity in db_entities.all():
282		entities[entity.entityid] = entity
283
284		if request and federation_slug:
285		request.session['%s_num_entities' %
286		federation_slug] = len(entities_from_xml)
287		request.session['%s_cur_entities' % federation_slug] = 0
288		request.session['%s_process_done' % federation_slug] = False
289		request.session.save()
290
291		updated = self._add_new_entities(
292		entities, entities_from_xml, request, federation_slug)
293
294		if request and federation_slug:
295		request.session['%s_process_done' % federation_slug] = True
296		request.session.save()
297
298		return removed, updated
299
300		def get_absolute_url(self):
301		return reverse('federation_view', args=[self.slug])
302
303		@classmethod
304		def get_sp(cls, entities, xml_name, ref_date=None):
305		if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
306		selected = entities.filter(
307		types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
308		else:
309		selected = entities.filter(types__xmlname=xml_name)
310		return len(selected)
311
312		@classmethod
313		def get_idp(cls, entities, xml_name, ref_date=None):
314		if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
315		selected = entities.filter(
316		types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
317		else:
318		selected = entities.filter(types__xmlname=xml_name)
319		return len(selected)
320
321		@classmethod
322		def get_aa(cls, entities, xml_name, ref_date=None):
323		if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
324		selected = entities.filter(
325		types__xmlname=xml_name, entity_federations__registration_instant__lt=ref_date)
326		else:
327		selected = entities.filter(types__xmlname=xml_name)
328		return len(selected)
329
330		def get_sp_saml1(self, entities, xml_name, ref_date=None):
331		return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)
332
333		def get_sp_saml2(self, entities, xml_name, ref_date=None):
334		return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)
335
336		def get_sp_shib1(self, entities, xml_name, ref_date=None):
337		return self.get_stat_protocol(entities, xml_name, 'SPSSODescriptor', ref_date)
338
339		def get_idp_saml1(self, entities, xml_name, ref_date=None):
340		return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)
341
342		def get_idp_saml2(self, entities, xml_name, ref_date=None):
343		return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)
344
345		def get_idp_shib1(self, entities, xml_name, ref_date=None):
346		return self.get_stat_protocol(entities, xml_name, 'IDPSSODescriptor', ref_date)
347
348		def get_stat_protocol(self, entities, xml_name, service_type, ref_date):
349		if ref_date and ref_date < pytz.utc.localize(datetime.now() - timedelta(days=1)):
350		selected = entities.filter(types__xmlname=service_type,
351		_display_protocols__contains=xml_name,
352		entity_federations__registration_instant__lt=ref_date)
353		else:
354		selected = entities.filter(types__xmlname=service_type,
355		_display_protocols__contains=xml_name)
356
357		return len(selected)
358
359		def can_edit(self, user, delete):
360		if user.is_superuser:
361		return True
362
363		permission = 'delete_federation' if delete else 'change_federation'
364		if user.has_perm('metadataparser.%s' % permission) and user in self.editor_users.all():
365		return True
366		return False
367
368
369		@receiver(pre_save, sender=Federation, dispatch_uid='federation_pre_save')
370		def federation_pre_save(sender, instance, **kwargs):
371		# Skip pre_save if only file name is saved
372		if kwargs.has_key('update_fields') and kwargs['update_fields'] == set(['file']):
373		return
374
375		#slug = slugify(unicode(instance.name))[:200]
376		# if instance.file_url and instance.file_url != '':
377		# try:
378		# instance.fetch_metadata_file(slug)
379		# except Exception, e:
380		# pass
381
382		if instance.name:
383		instance.slug = slugify(unicode(instance))[:200]
384
385
386		@receiver(pre_save, sender=Entity, dispatch_uid='entity_pre_save')
387		def entity_pre_save(sender, instance, **kwargs):
388		# if refetch and instance.file_url:
389		# slug = slugify(unicode(instance.name))[:200]
390		# instance.fetch_metadata_file(slug)
391		# instance.process_metadata()
392		pass
393

GEANT / met

Federation.can_edit() B last analyzed 2018-05-28 06:56 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Federation.can_edit() B
last analyzed 2018-05-28 06:56 UTC