#6405

Committed 13 Jun 2025 01:08PM UTC coverage: 74.954% (+0.02%) from 74.933%

Build # #6405

Build Type

Pull #6645

coveralls-python

Committed by

ccataalin

Commit Message

HDX-10666 fix wrong value for object type and name

Pull Request Pull Request #6645: PR: HDX-10666 & HDX-10877: notifications hub and unsubscribe logic

Run Details

8 of 13 new or added lines in 5 files covered. (61.54%)

4 existing lines in 4 files now uncovered.

12934 of 17256 relevant lines covered (74.95%)

0.75 hits per line

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

64.63

/ckanext-hdx_users/ckanext/hdx_users/views/notification_platform.py

import logging
import json
from typing import Dict

from ckanext.hdx_users.general_token_model import ObjectType
from ckanext.hdx_users.notifications_subscription_model import EventType

import ckan.plugins.toolkit as tk
import ckan.model as model
import ckanext.hdx_users.helpers.helpers as usr_h
import ckanext.hdx_users.helpers.mailer as hdx_mailer

from flask import Blueprint, make_response
from ckan.common import current_user
from ckan.lib.mailer import MailerException
from ckan.types import Response, DataDict, Context
from ckan.views.api import CONTENT_TYPES

from ckanext.hdx_theme.util.mail import hdx_validate_email
from ckanext.hdx_users.controller_logic import notification_platform_logic
from ckanext.hdx_users.helpers.analytics import EmailValidationAnalyticsSender
from ckanext.hdx_users.helpers.constants import NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY, \
    NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY
from ckanext.hdx_users.notifications_subscription_model import ObjectType

from hashlib import md5

_h = tk.h
abort = tk.abort
request = tk.request

log = logging.getLogger(__name__)

hdx_notifications = Blueprint(u'hdx_notifications', __name__, url_prefix=u'/notifications')

def subscribe_to_object() -> Response:
    """
    Subscribe to an object (dataset, organization, group, crisis) for notifications as a guest user
    using an email validation token.
    There are 2 cases:
    1. User has an account either shadow or active - we need the email validation token to find the
       user and to register the new subscription
    2. User doesn't have an account - we need the email validation token to create a shadow account
       and to register the new subscription
    """



    dataset_list_url = tk.url_for('dataset.search')
    # we don't want to run this for 'HEAD' requests or for requests that don't come from a browser
    if request.user_agent.string.strip() and request.method == 'GET':
        token = request.args.get('token')
        try:
            token_obj = notification_platform_logic.verify_email_validation_token(token)
        except Exception as e:
            _h.flash_error('Your token is invalid. Your email address might have already been validated.')
            EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
            return tk.redirect_to(dataset_list_url)

        email = token_obj.user_id
        object_type = token_obj.object_type
        object = token_obj.object_id
        event_type = token_obj.extras.get(NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY, EventType.DATASET_UPDATED.value)
        if not email or not object:
            _h.flash_error('Couldn\'t find required parameters: email and dataset_id.')
            EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
            return tk.redirect_to(dataset_list_url)

        # create shadow account if needed
        context: Context = {'model': model,'session': model.Session, 'ignore_auth': True}
        user_dict = tk.get_action('hdx_shadow_user_create')(context, {'email': email})
        user_id = user_dict['id']
    else:
        return abort(404, 'Page not found')

    try:
        redirect_url = _generate_url_for(object_type, object)
    except Exception as e:
        log.error('An exception occurred:' + str(e))
        return abort(500, 'An error occurred')

    data_dict = {
        'user_id': user_id,
        'object': object,
        'object_type': object_type,
        'event_type': event_type,
    }
    try:
        tk.get_action('hdx_notifications_subscription_create')(context, data_dict)
        _h.flash_success(tk._(
            u'You have successfully set up email notifications. These will be sent to {0} when there '
            u'is an update.'.format(
                current_user.email)))
    except tk.ValidationError as e:
        log.error('An exception occurred:' + str(e))
        _h.flash_error(str(e))
    except Exception as e:
        log.error('An exception occurred:' + str(e))
        _h.flash_error('An error occurred: ' + str(e))
    return tk.redirect_to(redirect_url)


# def subscribe_to_dataset() -> Response:
#     # Get parameters from the URL
#     # email = tk.request.args.get('email')
#     # dataset_id = tk.request.args.get('dataset_id')
#
#     if request.user_agent.string.strip() and request.method == 'GET':
#         # we don't want to run this for 'HEAD' requests or for requests that don't come from a browser
#         token = tk.request.args.get('token')
#
#         dataset_list_url = tk.url_for('dataset.search')
#         try:
#             token_obj = notification_platform_logic.verify_email_validation_token(token)
#         except Exception as e:
#             _h.flash_error('Your token is invalid. Your email address might have already been validated.')
#             EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
#             return tk.redirect_to(dataset_list_url)
#
#         email = token_obj.user_id
#         dataset_id = token_obj.object_id
#         if not email or not dataset_id:
#             _h.flash_error('Couldn\'t find required parameters: email and dataset_id.')
#             EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
#             return tk.redirect_to(dataset_list_url)
#
#         context = {'ignore_auth': True}
#
#         try:
#             unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, dataset_id)
#             data_dict = {
#                 'email': email,
#                 'dataset_id': dataset_id,
#                 'unsubscribe_token': unsubscribe_token.token,
#             }
#             result = _add_notification_subscription(context, data_dict)
#             _h.flash_success(tk._(
#                 u'You have successfully set up email notifications for this dataset. These will be sent to {0} when the '
#                 u'dataset is updated on HDX.'.format(
#                     email)))
#         except tk.ValidationError as e:
#             log.error('An exception occurred:' + str(e))
#             _h.flash_error(str(e))
#         except Exception as e:
#             log.error('An exception occurred:' + str(e))
#             _h.flash_error('An error occurred: ' + str(e))
#
#         email_hash = md5(email.strip().lower().encode('utf8')).hexdigest()
#         EmailValidationAnalyticsSender('notification platform', True, email_hash).send_to_queue()
#
#         # Redirect to the dataset page
#         dataset_url = tk.url_for('dataset.read', id=dataset_id, came_from='notification_platform_subscription',
#                                  u=data_dict.get('unsubscribe_token'))
#         return tk.redirect_to(dataset_url)
#     return abort(404, 'Page not found')

def _generate_url_for(object_type: str, object: str, external: bool = False) -> str:
    if object_type == ObjectType.DATASET.value:
        endpoint = 'dataset.read'
    elif object_type == ObjectType.ORGANIZATION.value:
        endpoint = 'organization.read'
    elif object_type == ObjectType.GROUP.value:
        endpoint = 'group.read'
    elif object_type == ObjectType.CRISIS.value:
        page_dict = tk.get_action('page_show')({}, {'id': object})
        if page_dict.get('type') == 'event':
            endpoint = 'hdx_light_event.read_light_event'
        else:
            endpoint = 'hdx_light_dashboard.read_light_dashboard'
    else:
        raise tk.ValidationError(f'Invalid object_type: {object_type}')

    return tk.url_for(endpoint, id=object, _external=external)

def _add_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:
    result = tk.get_action('hdx_add_notification_subscription')(context, data_dict)
    return result


def subscription_confirmation() -> Response:
    email = tk.request.form.get('email')
    object_id = tk.request.form.get('object_id')
    object_type_str = tk.request.form.get('object_type')
    object_type = ObjectType(object_type_str)
    dataset_updates = tk.request.form.get('dataset_updates') == 'true'

    json_response_dict: Dict[str: any] = {
        'success': True
    }
    error_message = None

    try:

        if not current_user.is_authenticated:
            usr_h.is_valid_captcha(tk.request.form.get('g-recaptcha-response'))

            if not email:
                raise tk.Invalid(tk._('Email address is missing'))
            hdx_validate_email(email)

            action = None
            if object_type == ObjectType.DATASET.value:
                action = 'package_show'
            elif object_type == ObjectType.GROUP.value:
                action = 'group_show'
            elif object_type == ObjectType.ORGANIZATION.value:
                action = 'organization_show'
            elif object_type == ObjectType.CRISIS.value:
                action = 'page_show'

            try:
                context: Context = {}
                object_dict = tk.get_action(action)(context, {'id': object_id})
            except tk.ObjectNotFound:
                raise tk.ValidationError(f'{object_type.value} {object_id} does not exist')
            except Exception as e:
                log.error(f'Error retrieving target or user: {e}')
                raise e

            extras = {
                NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY: EventType.DATASET_UPDATED.value if dataset_updates else EventType.NEW_DATASET_ADDED.value
            }
            token_obj = notification_platform_logic.get_or_generate_email_validation_token(email, object_type,
                                                                                           object_dict['id'], extras)

            subject = u'Please verify your email address'
            verify_email_link = _h.url_for(
                'hdx_notifications.subscribe_to_object',
                token=token_obj.token, qualified=True
            )
            email_data = {
                'verify_email_link': verify_email_link,
                'object_title': object_dict.get('title'),
                'object_id': object_id,
                'object_link': _generate_url_for(object_type.value, object_id, True),
                'object_type': object_type.value,
                'dataset_updates': dataset_updates,
            }
            hdx_mailer.mail_recipient([{'email': email}], subject, email_data, footer=None,
                                      snippet='email/content/notification_platform/verify_email.html')

        # user is authenticated
        else:
            context: Context = {'session': model.Session, 'user': current_user.name}

            # data_dict = {
            #     'email': email,
            #     'object_id': object_id,
            #     'object_type': object_type,
            #     'unsubscribe_token': unsubscribe_token.token,
            # }
            # result = _add_notification_subscription(context, data_dict)

            data_dict = {
                'user_id': current_user.id,
                'object': object_id,
                'object_type': object_type,
                'event_type': EventType.DATASET_UPDATED.value if dataset_updates else EventType.NEW_DATASET_ADDED.value,
            }

            subscription = tk.get_action('hdx_notifications_subscription_create')(context, data_dict)

            email = current_user.email
            extras = {
                NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY: subscription.get('id')
            }
            unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, object_type,
                                                                                              object_id, extras)

            email_hash = md5(email.strip().lower().encode('utf8')).hexdigest()
            EmailValidationAnalyticsSender('notification platform', True, email_hash).send_to_queue()

            json_response_dict['unsubscribe_token'] = unsubscribe_token.token

    except tk.ValidationError as e:
        error_message =  e.error_summary
    except tk.Invalid as e:
        error_message = e.error
    except MailerException as e:
        log.error(e)
        error_message = 'Error sending the confirmation email, please try again.'
    except Exception as e:
        log.error(e)
        error_message = str(e)
    if error_message:
        json_response_dict = {
            'success': False,
            'error': {
                'message': error_message
            }
        }
    return _build_json_response(json_response_dict)


def unsubscribe_confirmation() -> Response:
    token = tk.request.form.get('token')

    try:
        token_obj = notification_platform_logic.verify_unsubscribe_token(token, inactivate=True)

        context = {'ignore_auth': True}
        data_dict = {'email': token_obj.user_id, 'dataset_id': token_obj.object_id,
                     'subscription_id': token_obj.extras.get(NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY)}
        result = _delete_notification_subscription(context, data_dict)
    except tk.ValidationError as e:
        log.error('An exception occurred:' + str(e))
        return _build_json_response(
            {
                'success': False,
                'error': {
                    'message': 'An exception occurred:' + str(e)
                }
            }
        )
    except Exception as e:
        log.error('An exception occurred:' + str(e))
        return _build_json_response(
            {
                'success': False,
                'error': {
                    'message': 'An error occurred: ' + str(e)
                }
            }
        )
    return _build_json_response({'success': True})


def _delete_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:
    result = tk.get_action('hdx_delete_notification_subscription')(context, data_dict)
    return result

def _build_json_response(data_dict: DataDict, status=200):
    headers = {
        'Content-Type': CONTENT_TYPES['json'],
    }
    body = json.dumps(data_dict)
    response = make_response((body, status, headers))
    return response


# hdx_notifications.add_url_rule(u'/subscribe-to-dataset', view_func=subscribe_to_dataset)
hdx_notifications.add_url_rule(u'/subscribe-to-object', view_func=subscribe_to_object, methods=['GET', 'POST'])
hdx_notifications.add_url_rule(u'/subscription-confirmation', view_func=subscription_confirmation, methods=['POST'])
hdx_notifications.add_url_rule(u'/unsubscribe-confirmation', view_func=unsubscribe_confirmation, methods=['POST'])

1	import logging	1✔
2	import json	1✔
3	from typing import Dict	1✔
4
5	from ckanext.hdx_users.general_token_model import ObjectType	1✔
6	from ckanext.hdx_users.notifications_subscription_model import EventType	1✔
7
8	import ckan.plugins.toolkit as tk	1✔
9	import ckan.model as model	1✔
10	import ckanext.hdx_users.helpers.helpers as usr_h	1✔
11	import ckanext.hdx_users.helpers.mailer as hdx_mailer	1✔
12
13	from flask import Blueprint, make_response	1✔
14	from ckan.common import current_user	1✔
15	from ckan.lib.mailer import MailerException	1✔
16	from ckan.types import Response, DataDict, Context	1✔
17	from ckan.views.api import CONTENT_TYPES	1✔
18
19	from ckanext.hdx_theme.util.mail import hdx_validate_email	1✔
20	from ckanext.hdx_users.controller_logic import notification_platform_logic	1✔
21	from ckanext.hdx_users.helpers.analytics import EmailValidationAnalyticsSender	1✔
22	from ckanext.hdx_users.helpers.constants import NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY, \	1✔
23	NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY
24	from ckanext.hdx_users.notifications_subscription_model import ObjectType	1✔
25
26	from hashlib import md5	1✔
27
28	_h = tk.h	1✔
29	abort = tk.abort	1✔
30	request = tk.request	1✔
31
32	log = logging.getLogger(__name__)	1✔
33
34	hdx_notifications = Blueprint(u'hdx_notifications', __name__, url_prefix=u'/notifications')	1✔
35
36	def subscribe_to_object() -> Response:	1✔
37	"""
38	Subscribe to an object (dataset, organization, group, crisis) for notifications as a guest user
39	using an email validation token.
40	There are 2 cases:
41	1. User has an account either shadow or active - we need the email validation token to find the
42	user and to register the new subscription
43	2. User doesn't have an account - we need the email validation token to create a shadow account
44	and to register the new subscription
45	"""
46
47
48
49	dataset_list_url = tk.url_for('dataset.search')	1✔
50	# we don't want to run this for 'HEAD' requests or for requests that don't come from a browser
51	if request.user_agent.string.strip() and request.method == 'GET':	1✔
52	token = request.args.get('token')	1✔
53	try:	1✔
54	token_obj = notification_platform_logic.verify_email_validation_token(token)	1✔
55	except Exception as e:	×
56	_h.flash_error('Your token is invalid. Your email address might have already been validated.')	×
57	EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()	×
58	return tk.redirect_to(dataset_list_url)	×
59
60	email = token_obj.user_id	1✔
61	object_type = token_obj.object_type	1✔
62	object = token_obj.object_id	1✔
63	event_type = token_obj.extras.get(NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY, EventType.DATASET_UPDATED.value)	1✔
64	if not email or not object:	1✔
65	_h.flash_error('Couldn\'t find required parameters: email and dataset_id.')	×
66	EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()	×
67	return tk.redirect_to(dataset_list_url)	×
68
69	# create shadow account if needed
70	context: Context = {'model': model,'session': model.Session, 'ignore_auth': True}	1✔
71	user_dict = tk.get_action('hdx_shadow_user_create')(context, {'email': email})	1✔
72	user_id = user_dict['id']	1✔
73	else:
74	return abort(404, 'Page not found')	×
75
76	try:	1✔
77	redirect_url = _generate_url_for(object_type, object)	1✔
78	except Exception as e:	×
79	log.error('An exception occurred:' + str(e))	×
80	return abort(500, 'An error occurred')	×
81
82	data_dict = {	1✔
83	'user_id': user_id,
84	'object': object,
85	'object_type': object_type,
86	'event_type': event_type,
87	}
88	try:	1✔
89	tk.get_action('hdx_notifications_subscription_create')(context, data_dict)	1✔
90	_h.flash_success(tk._(	1✔
91	u'You have successfully set up email notifications. These will be sent to {0} when there '
92	u'is an update.'.format(
93	current_user.email)))
94	except tk.ValidationError as e:	×
95	log.error('An exception occurred:' + str(e))	×
96	_h.flash_error(str(e))	×
97	except Exception as e:	×
98	log.error('An exception occurred:' + str(e))	×
99	_h.flash_error('An error occurred: ' + str(e))	×
100	return tk.redirect_to(redirect_url)	1✔
101
102
103	# def subscribe_to_dataset() -> Response:
104	# # Get parameters from the URL
105	# # email = tk.request.args.get('email')
106	# # dataset_id = tk.request.args.get('dataset_id')
107	#
108	# if request.user_agent.string.strip() and request.method == 'GET':
109	# # we don't want to run this for 'HEAD' requests or for requests that don't come from a browser
110	# token = tk.request.args.get('token')
111	#
112	# dataset_list_url = tk.url_for('dataset.search')
113	# try:
114	# token_obj = notification_platform_logic.verify_email_validation_token(token)
115	# except Exception as e:
116	# _h.flash_error('Your token is invalid. Your email address might have already been validated.')
117	# EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
118	# return tk.redirect_to(dataset_list_url)
119	#
120	# email = token_obj.user_id
121	# dataset_id = token_obj.object_id
122	# if not email or not dataset_id:
123	# _h.flash_error('Couldn\'t find required parameters: email and dataset_id.')
124	# EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()
125	# return tk.redirect_to(dataset_list_url)
126	#
127	# context = {'ignore_auth': True}
128	#
129	# try:
130	# unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, dataset_id)
131	# data_dict = {
132	# 'email': email,
133	# 'dataset_id': dataset_id,
134	# 'unsubscribe_token': unsubscribe_token.token,
135	# }
136	# result = _add_notification_subscription(context, data_dict)
137	# _h.flash_success(tk._(
138	# u'You have successfully set up email notifications for this dataset. These will be sent to {0} when the '
139	# u'dataset is updated on HDX.'.format(
140	# email)))
141	# except tk.ValidationError as e:
142	# log.error('An exception occurred:' + str(e))
143	# _h.flash_error(str(e))
144	# except Exception as e:
145	# log.error('An exception occurred:' + str(e))
146	# _h.flash_error('An error occurred: ' + str(e))
147	#
148	# email_hash = md5(email.strip().lower().encode('utf8')).hexdigest()
149	# EmailValidationAnalyticsSender('notification platform', True, email_hash).send_to_queue()
150	#
151	# # Redirect to the dataset page
152	# dataset_url = tk.url_for('dataset.read', id=dataset_id, came_from='notification_platform_subscription',
153	# u=data_dict.get('unsubscribe_token'))
154	# return tk.redirect_to(dataset_url)
155	# return abort(404, 'Page not found')
156
157	def _generate_url_for(object_type: str, object: str, external: bool = False) -> str:	1✔
158	if object_type == ObjectType.DATASET.value:	1✔
159	endpoint = 'dataset.read'	1✔
160	elif object_type == ObjectType.ORGANIZATION.value:	×
161	endpoint = 'organization.read'	×
162	elif object_type == ObjectType.GROUP.value:	×
163	endpoint = 'group.read'	×
164	elif object_type == ObjectType.CRISIS.value:	×
165	page_dict = tk.get_action('page_show')({}, {'id': object})	×
166	if page_dict.get('type') == 'event':	×
167	endpoint = 'hdx_light_event.read_light_event'	×
168	else:
169	endpoint = 'hdx_light_dashboard.read_light_dashboard'	×
170	else:
171	raise tk.ValidationError(f'Invalid object_type: {object_type}')	×
172
173	return tk.url_for(endpoint, id=object, _external=external)	1✔
174
175	def _add_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:	1✔
176	result = tk.get_action('hdx_add_notification_subscription')(context, data_dict)	×
177	return result	×
178
179
180	def subscription_confirmation() -> Response:	1✔
181	email = tk.request.form.get('email')	1✔
182	object_id = tk.request.form.get('object_id')	1✔
183	object_type_str = tk.request.form.get('object_type')	1✔
184	object_type = ObjectType(object_type_str)	1✔
185	dataset_updates = tk.request.form.get('dataset_updates') == 'true'	1✔
186
187	json_response_dict: Dict[str: any] = {	1✔
188	'success': True
189	}
190	error_message = None	1✔
191
192	try:	1✔
193
194	if not current_user.is_authenticated:	1✔
195	usr_h.is_valid_captcha(tk.request.form.get('g-recaptcha-response'))	1✔
196
197	if not email:	1✔
198	raise tk.Invalid(tk._('Email address is missing'))	×
199	hdx_validate_email(email)	1✔
200
201	action = None	1✔
202	if object_type == ObjectType.DATASET.value:	1✔
203	action = 'package_show'	1✔
204	elif object_type == ObjectType.GROUP.value:	×
205	action = 'group_show'	×
206	elif object_type == ObjectType.ORGANIZATION.value:	×
207	action = 'organization_show'	×
208	elif object_type == ObjectType.CRISIS.value:	×
209	action = 'page_show'	×
210
211	try:	1✔
212	context: Context = {}	1✔
213	object_dict = tk.get_action(action)(context, {'id': object_id})	1✔
214	except tk.ObjectNotFound:	×
215	raise tk.ValidationError(f'{object_type.value} {object_id} does not exist')	×
216	except Exception as e:	×
217	log.error(f'Error retrieving target or user: {e}')	×
218	raise e	×
219
220	extras = {	1✔
221	NOTIFICATION_PLATFORM_EVENT_TYPE_EXTRAS_KEY: EventType.DATASET_UPDATED.value if dataset_updates else EventType.NEW_DATASET_ADDED.value
222	}
223	token_obj = notification_platform_logic.get_or_generate_email_validation_token(email, object_type,	1✔
224	object_dict['id'], extras)
225
226	subject = u'Please verify your email address'	1✔
227	verify_email_link = _h.url_for(	1✔
228	'hdx_notifications.subscribe_to_object',
229	token=token_obj.token, qualified=True
230	)
231	email_data = {	1✔
232	'verify_email_link': verify_email_link,
233	'object_title': object_dict.get('title'),
234	'object_id': object_id,
235	'object_link': _generate_url_for(object_type.value, object_id, True),
236	'object_type': object_type.value,
237	'dataset_updates': dataset_updates,
238	}
239	hdx_mailer.mail_recipient([{'email': email}], subject, email_data, footer=None,	1✔
240	snippet='email/content/notification_platform/verify_email.html')
241
242	# user is authenticated
243	else:
244	context: Context = {'session': model.Session, 'user': current_user.name}	1✔
245
246	# data_dict = {
247	# 'email': email,
248	# 'object_id': object_id,
249	# 'object_type': object_type,
250	# 'unsubscribe_token': unsubscribe_token.token,
251	# }
252	# result = _add_notification_subscription(context, data_dict)
253
254	data_dict = {	1✔
255	'user_id': current_user.id,
256	'object': object_id,
257	'object_type': object_type,
258	'event_type': EventType.DATASET_UPDATED.value if dataset_updates else EventType.NEW_DATASET_ADDED.value,
259	}
260
261	subscription = tk.get_action('hdx_notifications_subscription_create')(context, data_dict)	1✔
262
263	email = current_user.email	1✔
264	extras = {	1✔
265	NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY: subscription.get('id')
266	}
267	unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, object_type,	1✔
268	object_id, extras)
269
270	email_hash = md5(email.strip().lower().encode('utf8')).hexdigest()	1✔
271	EmailValidationAnalyticsSender('notification platform', True, email_hash).send_to_queue()	1✔
272
273	json_response_dict['unsubscribe_token'] = unsubscribe_token.token	1✔
274
275	except tk.ValidationError as e:	×
276	error_message = e.error_summary	×
277	except tk.Invalid as e:	×
278	error_message = e.error	×
279	except MailerException as e:	×
280	log.error(e)	×
281	error_message = 'Error sending the confirmation email, please try again.'	×
282	except Exception as e:	×
283	log.error(e)	×
284	error_message = str(e)	×
285	if error_message:	1✔
286	json_response_dict = {	×
287	'success': False,
288	'error': {
289	'message': error_message
290	}
291	}
292	return _build_json_response(json_response_dict)	1✔
293
294
295	def unsubscribe_confirmation() -> Response:	1✔
296	token = tk.request.form.get('token')	1✔
297
298	try:	1✔
299	token_obj = notification_platform_logic.verify_unsubscribe_token(token, inactivate=True)	1✔
300
301	context = {'ignore_auth': True}	1✔
302	data_dict = {'email': token_obj.user_id, 'dataset_id': token_obj.object_id,	1✔
303	'subscription_id': token_obj.extras.get(NOTIFICATION_PLATFORM_SUBSCRIPTION_ID_EXTRAS_KEY)}
304	result = _delete_notification_subscription(context, data_dict)	×
305	except tk.ValidationError as e:	1✔
306	log.error('An exception occurred:' + str(e))	×
307	return _build_json_response(	×
308	{
309	'success': False,
310	'error': {
311	'message': 'An exception occurred:' + str(e)
312	}
313	}
314	)
315	except Exception as e:	1✔
316	log.error('An exception occurred:' + str(e))	1✔
317	return _build_json_response(	1✔
318	{
319	'success': False,
320	'error': {
321	'message': 'An error occurred: ' + str(e)
322	}
323	}
324	)
UNCOV 325	return _build_json_response({'success': True})	×
326
327
328	def _delete_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:	1✔
329	result = tk.get_action('hdx_delete_notification_subscription')(context, data_dict)	×
330	return result	×
331
332	def _build_json_response(data_dict: DataDict, status=200):	1✔
333	headers = {	1✔
334	'Content-Type': CONTENT_TYPES['json'],
335	}
336	body = json.dumps(data_dict)	1✔
337	response = make_response((body, status, headers))	1✔
338	return response	1✔
339
340
341	# hdx_notifications.add_url_rule(u'/subscribe-to-dataset', view_func=subscribe_to_dataset)
342	hdx_notifications.add_url_rule(u'/subscribe-to-object', view_func=subscribe_to_object, methods=['GET', 'POST'])	1✔
343	hdx_notifications.add_url_rule(u'/subscription-confirmation', view_func=subscription_confirmation, methods=['POST'])	1✔
344	hdx_notifications.add_url_rule(u'/unsubscribe-confirmation', view_func=unsubscribe_confirmation, methods=['POST'])	1✔

OCHA-DAP / hdx-ckan / #6405

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