test.py

import inspect
from datetime import timedelta

from rest_framework.settings import import_from_string

from api.models.user import ParentChildRelation
from gimi.celery import app


class UnknownNotificationError(BaseException):
    pass


mapping = {
    'ADFRPU001': {
        'func': 'notifications.example_notifications.ADFRPU001',
        'delay': timedelta(hours=64),
        'enabled': True,
    }
}


_scheduled_notifications = []


def send_alpine_notification(notification_ref, **kwargs):
    try:
        spec = mapping[notification_ref]
    except KeyError:
        raise UnknownNotificationError("Unknown notification reference %s" % notification_ref)

    if not spec['enabled']:
        return

    func = import_from_string(spec['func'], None)
    missing_kwargs = set(kwargs.keys()).difference(inspect.signature(func.run).parameters.keys())
    if missing_kwargs:
        raise ValueError("Missing kwargs for notification ref %s:\n %s" % notification_ref, missing_kwargs)

    try:
        countdown = int(spec['delay'].total_seconds())
    except KeyError:
        countdown = 0

    func.apply_async(kwargs=kwargs, countdown=countdown)


@app.task
def ADFRPU001(parent_child_relation_id):
    try:
        relation = ParentChildRelation.objects.get(pk=parent_child_relation_id)
    except ParentChildRelation.DoesNotExist:
        return

    # You should have context of the action that triggered this notification call through the objects
    # provided.
    # Send push


# Run in tests
def validate_mapping():
    errors = []
    for notification_ref, spec in mapping.items():
        if 'func' not in spec:
            errors.append("Notification ref %s missing 'func'" % notification_ref)
        if 'enabled' not in spec:
            errors.append("Notification ref %s missing 'enabled'" % notification_ref)

        try:
            import_from_string(spec['func'], None)
        except ImportError:
            errors.append("Could not import function for notification ref %s" % notification_ref)

    if errors:
        raise AssertionError("Errors! %s" % ',\n'.join(errors))