Progress VIII

* integration with quota holder
* new credit event for integration with aquarium

snf-astakos-app/astakos/im/endpoints/aquarium/producer.py

+# Copyright 2011-2012 GRNET S.A. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or
+# without modification, are permitted provided that the following
+# conditions are met:
+#
+#   1. Redistributions of source code must retain the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer.
+#
+#   2. Redistributions in binary form must reproduce the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer in the documentation and/or other materials
+#      provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
+# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# The views and conclusions contained in the software and
+# documentation are those of the authors and should not be
+# interpreted as representing official policies, either expressed
+# or implied, of GRNET S.A.
+
+import logging
+
+from functools import wraps
+from urlparse import urlparse
+
+from astakos.im.settings import QUEUE_CONNECTION
+
+if QUEUE_CONNECTION:
+    from synnefo.lib.queue import (exchange_connect, exchange_send,
+            exchange_close, UserEvent, Receipt
+    )
+
+QUEUE_CLIENT_ID = 3 # Astakos.
+INSTANCE_ID = '1'
+RESOURCE = 'addcredits'
+DEFAULT_CREDITS = 1000
+
+logging.basicConfig(format='%(asctime)s [%(levelname)s] %(name)s %(message)s',
+        datefmt='%Y-%m-%d %H:%M:%S'
+)
+logger = logging.getLogger('endpoint.aquarium')
+
+def wrapper(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        if not QUEUE_CONNECTION:
+            return
+        
+        try:
+            body, key = func(*args, **kwargs)
+            conn = exchange_connect(QUEUE_CONNECTION)
+            parts = urlparse(QUEUE_CONNECTION)
+            exchange = parts.path[1:]
+            routing_key = key % exchange
+            exchange_send(conn, routing_key, body)
+            exchange_close(conn)
+        except BaseException, e:
+            logger.exception(e)
+    return wrapper
+
+@wrapper
+def report_user_event(user, create=False):
+    eventType = 'create' if not create else 'modify'
+    body = UserEvent(QUEUE_CLIENT_ID, user.email, user.is_active, eventType, {}
+    ).format()
+    routing_key = '%s.user'
+    return body, routing_key
+
+@wrapper
+def report_user_credits_event(user):
+    body = Receipt(QUEUE_CLIENT_ID, user.email, INSTANCE_ID, RESOURCE,
+        DEFAULT_CREDITS, details={}
+    ).format()
+    routing_key = '%s.resource'
+    return body, routing_key
+
+def report_credits_event():
+    # better approach?
+    from astakos.im.models import AstakosUser
+    map(report_user_credits_event, AstakosUser.objects.all())

snf-astakos-app/astakos/im/endpoints/quotaholder.py

+# Copyright 2011-2012 GRNET S.A. All rights reserved.
+# 
+# Redistribution and use in source and binary forms, with or
+# without modification, are permitted provided that the following
+# conditions are met:
+# 
+#   1. Redistributions of source code must retain the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer.
+# 
+#   2. Redistributions in binary form must reproduce the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer in the documentation and/or other materials
+#      provided with the distribution.
+# 
+# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
+# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+# 
+# The views and conclusions contained in the software and
+# documentation are those of the authors and should not be
+# interpreted as representing official policies, either expressed
+# or implied, of GRNET S.A.
+
+import socket
+import logging
+
+from django.utils.translation import ugettext as _
+
+from commissioning.clients.quotaholder import QuotaholderHTTP
+
+from astakos.im.settings import QUOTA_HOLDER_URL, LOGGING_LEVEL
+
+ENTITY_KEY = '1'
+
+logger = logging.getLogger(__name__)
+
+def register_users(users, client=None):
+    if not users:
+        return
+    
+    if not QUOTA_HOLDER_URL:
+        return
+    
+    c = client or QuotaholderHTTP(QUOTA_HOLDER_URL)
+    data = []
+    append = data.append
+    for user in users:
+        try:
+            entity = user.email
+        except AttributeError:
+            continue
+        else:    
+            args = entity, owner, key, ownerkey = (
+                entity, 'system', ENTITY_KEY, ''
+            )
+            append(args)
+    
+    if not data:
+        return
+    
+    rejected = c.create_entity(
+        context={},
+        create_entity=data,
+    )
+    msg = _('Create entities: %s - Rejected: %s' % (data, rejected,))
+    logger.log(LOGGING_LEVEL, msg)
+    
+    created = filter(lambda u: unicode(u.email) not in rejected, users)
+    send_quota(created, c)
+    return rejected
+
+def send_quota(users, client=None):
+    if not users:
+        return
+    
+    if not QUOTA_HOLDER_URL:
+        return
+    
+    c = client or QuotaholderHTTP(QUOTA_HOLDER_URL)
+    data = []
+    append = data.append
+    for user in users:
+        try:
+            entity = user.email
+        except AttributeError:
+            continue
+        else:
+            for resource, limit in user.quota.iteritems():
+                args = entity, resource, key, quantity, capacity, import_limit , \
+                    export_limit, flags =(
+                        entity, resource, ENTITY_KEY, '0', str(limit), 0, 0, 0
+                )
+                append(args)
+    
+    if not data:
+        return
+    
+    rejected = c.set_quota(context={}, set_quota=data)
+    msg = _('Set quota: %s - Rejected: %s' % (data, rejected,))
+    logger.log(LOGGING_LEVEL, msg)
+    return rejected
+
+def get_quota(users, client=None):
+    if not users:
+        return
+    
+    if not QUOTA_HOLDER_URL:
+        return
+    
+    c = client or QuotaholderHTTP(QUOTA_HOLDER_URL)
+    data = []
+    append = data.append
+    for user in users:
+        try:
+            entity = user.email
+        except AttributeError:
+            continue
+        else:
+            for r in user.quota.keys():
+                args = entity, resource, key = entity, r, ENTITY_KEY
+                append(args)
+    
+    if not data:
+        return
+    
+    r = c.get_quota(context={}, get_quota=data)
+    msg = _('Get quota: %s' % data)
+    logger.log(LOGGING_LEVEL, msg)
+    return r
\ No newline at end of file

snf-astakos-app/astakos/im/management/commands/group_list.py

@@ -57,9 +57,6 @@ class Command(BaseCommand):
     )
     
     def handle(self, *args, **options):
-        if args:
-            raise CommandError("Command doesn't accept any arguments")
-        
         groups = AstakosGroup.objects.all()
         
         if options.get('pending'):

snf-astakos-app/astakos/im/management/commands/group_update.py

@@ -54,6 +54,11 @@ class Command(BaseCommand):
             dest='enable',
             default=False,
             help="Enable group"),
+        make_option('--disable',
+            action='store_true',
+            dest='disable',
+            default=False,
+            help="Disable group"),
     )
     
     def handle(self, *args, **options):
@@ -92,5 +97,8 @@ class Command(BaseCommand):
             
             if options.get('enable'):
                 group.enable()
+            elif options.get('disable'):
+                group.disable()
+        
         except Exception, e:
             raise CommandError(e)
\ No newline at end of file

snf-astakos-app/astakos/im/management/commands/quotaholder_bootstrap.py

+# Copyright 2012 GRNET S.A. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or
+# without modification, are permitted provided that the following
+# conditions are met:
+#
+#   1. Redistributions of source code must retain the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer.
+#
+#   2. Redistributions in binary form must reproduce the above
+#      copyright notice, this list of conditions and the following
+#      disclaimer in the documentation and/or other materials
+#      provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
+# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
+# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
+# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# The views and conclusions contained in the software and
+# documentation are those of the authors and should not be
+# interpreted as representing official policies, either expressed
+# or implied, of GRNET S.A.
+
+from django.core.management.base import BaseCommand, CommandError
+from django.db.utils import IntegrityError
+
+from astakos.im.models import AstakosUser
+from astakos.im.endpoints.quotaholder import register_users
+
+class Command(BaseCommand):
+    help = "Send user information and resource quota in the Quotaholder"
+    
+    def handle(self, *args, **options):
+        try:
+            r = register_users(AstakosUser.objects.all())
+            self.stdout.write("Rejected: %s\n" % r)
+        except BaseException, e:
+            print e
+            raise CommandError("Bootstrap failed.")
\ No newline at end of file

snf-astakos-app/astakos/im/management/commands/user_update.py

@@ -39,6 +39,7 @@ from django.core.exceptions import ValidationError
 from django.db.utils import IntegrityError
 
 from astakos.im.models import AstakosUser, AstakosGroup, Membership
+from astakos.im.endpoints.aquarium.producer import report_user_credits_event
 from ._common import remove_user_permission, add_user_permission
 
 class Command(BaseCommand):
@@ -104,6 +105,11 @@ class Command(BaseCommand):
         make_option('--delete-permission',
             dest='delete-permission',
             help="Delete user permission"),
+        make_option('--refill-credits',
+            action='store_true',
+            dest='refill',
+            default=False,
+            help="Refill user credits"),
         )
     
     def handle(self, *args, **options):
@@ -201,6 +207,9 @@ class Command(BaseCommand):
         if options['renew_token']:
             user.renew_token()
         
+        if options['refill']:
+            report_user_credits_event(user)
+        
         try:
             user.save()
         except ValidationError, e:

snf-astakos-app/astakos/im/models.py

@@ -38,7 +38,6 @@ import logging
 from time import asctime
 from datetime import datetime, timedelta
 from base64 import b64encode
-from urlparse import urlparse
 from random import randint
 from collections import defaultdict
 
@@ -47,14 +46,17 @@ from django.contrib.auth.models import User, UserManager, Group
 from django.utils.translation import ugettext as _
 from django.core.exceptions import ValidationError
 from django.db import transaction
-from django.db.models.signals import post_save, post_syncdb
+from django.db.models.signals import pre_save, post_save, post_syncdb, post_delete
+from django.dispatch import Signal
 from django.db.models import Q
 
-from astakos.im.settings import DEFAULT_USER_LEVEL, INVITATIONS_PER_LEVEL, \
-    AUTH_TOKEN_DURATION, BILLING_FIELDS, QUEUE_CONNECTION, \
-    EMAILCHANGE_ACTIVATION_DAYS, LOGGING_LEVEL
+from astakos.im.settings import (DEFAULT_USER_LEVEL, INVITATIONS_PER_LEVEL,
+    AUTH_TOKEN_DURATION, BILLING_FIELDS, EMAILCHANGE_ACTIVATION_DAYS, LOGGING_LEVEL
+)
+from astakos.im.endpoints.quotaholder import register_users, send_quota
+from astakos.im.endpoints.aquarium.producer import report_user_event
 
-QUEUE_CLIENT_ID = 3 # Astakos.
+from astakos.im.tasks import propagate_groupmembers_quota
 
 logger = logging.getLogger(__name__)
 
@@ -108,13 +110,23 @@ class GroupKind(models.Model):
 class AstakosGroup(Group):
     kind = models.ForeignKey(GroupKind)
     desc = models.TextField('Description', null=True)
-    policy = models.ManyToManyField(Resource, null=True, blank=True, through='AstakosGroupQuota')
-    creation_date = models.DateTimeField('Creation date', default=datetime.now())
+    policy = models.ManyToManyField(Resource, null=True, blank=True,
+        through='AstakosGroupQuota'
+    )
+    creation_date = models.DateTimeField('Creation date',
+        default=datetime.now()
+    )
     issue_date = models.DateTimeField('Issue date', null=True)
     expiration_date = models.DateTimeField('Expiration date', null=True)
-    moderation_enabled = models.BooleanField('Moderated membership?', default=True)
-    approval_date = models.DateTimeField('Activation date', null=True, blank=True)
-    estimated_participants = models.PositiveIntegerField('Estimated #participants', null=True)
+    moderation_enabled = models.BooleanField('Moderated membership?',
+        default=True
+    )
+    approval_date = models.DateTimeField('Activation date', null=True,
+        blank=True
+    )
+    estimated_participants = models.PositiveIntegerField('Estimated #members',
+        null=True
+    )
     
     @property
     def is_disabled(self):
@@ -137,17 +149,25 @@ class AstakosGroup(Group):
             return False
         return True
     
-    @property
-    def participants(self):
-        return len(self.approved_members)
+#     @property
+#     def participants(self):
+#         return len(self.approved_members)
     
     def enable(self):
+        if self.is_enabled:
+            return
         self.approval_date = datetime.now()
         self.save()
+        quota_disturbed.send(sender=self, users=approved_members)
+        update_groupmembers_quota.apply_async(args=[self], eta=self.issue_date)
+        update_groupmembers_quota.apply_async(args=[self], eta=self.expiration_date)
     
     def disable(self):
+        if self.is_disabled:
+            return
         self.approval_date = None
         self.save()
+        quota_disturbed.send(sender=self, users=self.approved_members)
     
     def approve_member(self, person):
         m, created = self.membership_set.get_or_create(person=person)
@@ -160,12 +180,11 @@ class AstakosGroup(Group):
     
     @property
     def members(self):
-        return map(lambda m:m.person, self.membership_set.all())
+        return [m.person for m in self.membership_set.all()]
     
     @property
     def approved_members(self):
-        f = filter(lambda m:m.is_approved, self.membership_set.all())
-        return map(lambda m:m.person, f)
+        return [m.person for m in self.membership_set.all() if m.is_approved]
     
     @property
     def quota(self):
@@ -174,11 +193,6 @@ class AstakosGroup(Group):
             d[q.resource] += q.limit
         return d
     
-    @property
-    def has_undefined_policies(self):
-        # TODO: can avoid query?
-        return Resource.objects.filter(~Q(astakosgroup=self)).exists()
-    
     @property
     def owners(self):
         return self.owner.all()
@@ -266,7 +280,10 @@ class AstakosUser(User):
         d = defaultdict(int)
         for q in  self.astakosuserquota_set.all():
             d[q.resource.name] += q.limit
-        for g in self.astakos_groups.all():
+        for m in self.membership_set.all():
+            if not m.is_approved:
+                continue
+            g = m.group
             if not g.is_enabled:
                 continue
             for r, limit in g.quota.iteritems():
@@ -294,7 +311,6 @@ class AstakosUser(User):
                     self.username = username
             if not self.provider:
                 self.provider = 'local'
-        report_user_event(self)
         self.validate_unique_email_isactive()
         if self.is_active and self.activation_sent:
             # reset the activation sent
@@ -374,9 +390,11 @@ class Membership(models.Model):
     def approve(self):
         self.date_joined = datetime.now()
         self.save()
-        
+        quota_disturbed.send(sender=self, users=(self.person,))
+    
     def disapprove(self):
         self.delete()
+        quota_disturbed.send(sender=self, users=(self.person,))
 
 class AstakosGroupQuota(models.Model):
     limit = models.PositiveIntegerField('Limit')
@@ -428,53 +446,6 @@ class Invitation(models.Model):
     def __unicode__(self):
         return '%s -> %s [%d]' % (self.inviter, self.username, self.code)
 
-def report_user_event(user):
-    def should_send(user):
-        # report event incase of new user instance
-        # or if specific fields are modified
-        if not user.id:
-            return True
-        try:
-            db_instance = AstakosUser.objects.get(id = user.id)
-        except AstakosUser.DoesNotExist:
-            return True
-        for f in BILLING_FIELDS:
-            if (db_instance.__getattribute__(f) != user.__getattribute__(f)):
-                return True
-        return False
-
-    if QUEUE_CONNECTION and should_send(user):
-
-        from astakos.im.queue.userevent import UserEvent
-        from synnefo.lib.queue import exchange_connect, exchange_send, \
-                exchange_close
-
-        eventType = 'create' if not user.id else 'modify'
-        body = UserEvent(QUEUE_CLIENT_ID, user, eventType, {}).format()
-        conn = exchange_connect(QUEUE_CONNECTION)
-        parts = urlparse(QUEUE_CONNECTION)
-        exchange = parts.path[1:]
-        routing_key = '%s.user' % exchange
-        exchange_send(conn, routing_key, body)
-        exchange_close(conn)
-
-def _generate_invitation_code():
-    while True:
-        code = randint(1, 2L**63 - 1)
-        try:
-            Invitation.objects.get(code=code)
-            # An invitation with this code already exists, try again
-        except Invitation.DoesNotExist:
-            return code
-
-def get_latest_terms():
-    try:
-        term = ApprovalTerms.objects.order_by('-id')[0]
-        return term
-    except IndexError:
-        pass
-    return None
-
 class EmailChangeManager(models.Manager):
     @transaction.commit_on_success
     def change_email(self, activation_key):
@@ -534,6 +505,23 @@ class AdditionalMail(models.Model):
     owner = models.ForeignKey(AstakosUser)
     email = models.EmailField()
 
+def _generate_invitation_code():
+    while True:
+        code = randint(1, 2L**63 - 1)
+        try:
+            Invitation.objects.get(code=code)
+            # An invitation with this code already exists, try again
+        except Invitation.DoesNotExist:
+            return code
+
+def get_latest_terms():
+    try:
+        term = ApprovalTerms.objects.order_by('-id')[0]
+        return term
+    except IndexError:
+        pass
+    return None
+
 def create_astakos_user(u):
     try:
         AstakosUser.objects.get(user_ptr=u.pk)
@@ -542,35 +530,88 @@ def create_astakos_user(u):
         extended_user.__dict__.update(u.__dict__)
         extended_user.renew_token()
         extended_user.save()
-    except:
+    except BaseException, e:
+        logger.exception(e)
         pass
 
-def superuser_post_syncdb(sender, **kwargs):
-    # if there was created a superuser
-    # associate it with an AstakosUser
+def fix_superusers(sender, **kwargs):
+    # Associate superusers with AstakosUser
     admins = User.objects.filter(is_superuser=True)
     for u in admins:
         create_astakos_user(u)
 
-post_syncdb.connect(superuser_post_syncdb)
-
-def superuser_post_save(sender, instance, **kwargs):
-    if instance.is_superuser:
-        create_astakos_user(instance)
-
-post_save.connect(superuser_post_save, sender=User)
-
-def set_default_group(sender, instance, created, **kwargs):
+def user_post_save(sender, instance, created, **kwargs):
     if not created:
         return
+    create_astakos_user(instance)
+
+def set_default_group(user):
     try: