#5935

Committed 05 Dec 2024 10:26AM UTC coverage: 74.957% (+0.07%) from 74.883%

Build # #5935

Build Type

Pull #6496

coveralls-python

Committed by

danmihaila

Commit Message

HDX-10398 add testing

Pull Request Pull Request #6496: HDX-10398 export data completeness and rename "completness"

Run Details

54 of 65 new or added lines in 4 files covered. (83.08%)

62 existing lines in 3 files now uncovered.

12562 of 16759 relevant lines covered (74.96%)

0.75 hits per line

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

66.35

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

import logging
import json

import ckan.plugins.toolkit as tk
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.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 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_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)
        return tk.redirect_to(dataset_url)
    return abort(404, 'Page not found')

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')
    dataset_id = tk.request.form.get('dataset_id')

    try:
        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)

        dataset_dict = tk.get_action('package_show')({}, {'id': dataset_id})
        token_obj = notification_platform_logic.get_or_generate_email_validation_token(email, dataset_dict['id'])

        subject = u'Please verify your email address'
        verify_email_link = _h.url_for(
            'hdx_notifications.subscribe_to_dataset',
            token=token_obj.token, qualified=True
        )
        email_data = {
            'verify_email_link': verify_email_link,
            'dataset_title': dataset_dict.get('title'),
            'dataset_id': dataset_id
        }
        hdx_mailer.mail_recipient([{'email': email}], subject, email_data, footer=None,
                                  snippet='email/content/notification_platform/verify_email.html')

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


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}
        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'/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
4	import ckan.plugins.toolkit as tk	1✔
5	import ckanext.hdx_users.helpers.helpers as usr_h	1✔
6	import ckanext.hdx_users.helpers.mailer as hdx_mailer	1✔
7
8	from flask import Blueprint, make_response	1✔
9	from ckan.lib.mailer import MailerException	1✔
10	from ckan.types import Response, DataDict, Context	1✔
11	from ckan.views.api import CONTENT_TYPES	1✔
12
13	from ckanext.hdx_theme.util.mail import hdx_validate_email	1✔
14	from ckanext.hdx_users.controller_logic import notification_platform_logic	1✔
15	from ckanext.hdx_users.helpers.analytics import EmailValidationAnalyticsSender	1✔
16
17	from hashlib import md5	1✔
18
19	_h = tk.h	1✔
20	abort = tk.abort	1✔
21	request = tk.request	1✔
22
23	log = logging.getLogger(__name__)	1✔
24
25	hdx_notifications = Blueprint(u'hdx_notifications', __name__, url_prefix=u'/notifications')	1✔
26
27
28	def subscribe_to_dataset() -> Response:	1✔
29	# Get parameters from the URL
30	# email = tk.request.args.get('email')
31	# dataset_id = tk.request.args.get('dataset_id')
32
33	if request.user_agent.string.strip() and request.method == 'GET':	1✔
34	# we don't want to run this for 'HEAD' requests or for requests that don't come from a browser
35	token = tk.request.args.get('token')	1✔
36
37	dataset_list_url = tk.url_for('dataset.search')	1✔
38	try:	1✔
39	token_obj = notification_platform_logic.verify_email_validation_token(token)	1✔
40	except Exception as e:	×
41	_h.flash_error('Your token is invalid. Your email address might have already been validated.')	×
42	EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()	×
43	return tk.redirect_to(dataset_list_url)	×
44
45	email = token_obj.user_id	1✔
46	dataset_id = token_obj.object_id	1✔
47	if not email or not dataset_id:	1✔
48	_h.flash_error('Couldn\'t find required parameters: email and dataset_id.')	×
49	EmailValidationAnalyticsSender('notification platform', False, '').send_to_queue()	×
50	return tk.redirect_to(dataset_list_url)	×
51
52	context = {'ignore_auth': True}	1✔
53
54	try:	1✔
55	unsubscribe_token = notification_platform_logic.get_or_generate_unsubscribe_token(email, dataset_id)	1✔
56	data_dict = {	1✔
57	'email': email,
58	'dataset_id': dataset_id,
59	'unsubscribe_token': unsubscribe_token.token,
60	}
61	result = _add_notification_subscription(context, data_dict)	1✔
62	_h.flash_success(tk._(	1✔
63	u'You have successfully set up email notifications for this dataset. These will be sent to {0} when the '
64	u'dataset is updated on HDX.'.format(
65	email)))
66	except tk.ValidationError as e:	×
67	log.error('An exception occurred:' + str(e))	×
68	_h.flash_error(str(e))	×
69	except Exception as e:	×
70	log.error('An exception occurred:' + str(e))	×
71	_h.flash_error('An error occurred: ' + str(e))	×
72
73	email_hash = md5(email.strip().lower().encode('utf8')).hexdigest()	1✔
74	EmailValidationAnalyticsSender('notification platform', True, email_hash).send_to_queue()	1✔
75
76	# Redirect to the dataset page
77	dataset_url = tk.url_for('dataset.read', id=dataset_id)	1✔
78	return tk.redirect_to(dataset_url)	1✔
UNCOV 79	return abort(404, 'Page not found')	×
80
81	def _add_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:	1✔
UNCOV 82	result = tk.get_action('hdx_add_notification_subscription')(context, data_dict)	×
83	return result	×
84
85
86	def subscription_confirmation() -> Response:	1✔
87	email = tk.request.form.get('email')	1✔
88	dataset_id = tk.request.form.get('dataset_id')	1✔
89
90	try:	1✔
91	usr_h.is_valid_captcha(tk.request.form.get('g-recaptcha-response'))	1✔
92
93	if not email:	1✔
UNCOV 94	raise tk.Invalid(tk._('Email address is missing'))	×
95	hdx_validate_email(email)	1✔
96
97	dataset_dict = tk.get_action('package_show')({}, {'id': dataset_id})	1✔
98	token_obj = notification_platform_logic.get_or_generate_email_validation_token(email, dataset_dict['id'])	1✔
99
100	subject = u'Please verify your email address'	1✔
101	verify_email_link = _h.url_for(	1✔
102	'hdx_notifications.subscribe_to_dataset',
103	token=token_obj.token, qualified=True
104	)
105	email_data = {	1✔
106	'verify_email_link': verify_email_link,
107	'dataset_title': dataset_dict.get('title'),
108	'dataset_id': dataset_id
109	}
110	hdx_mailer.mail_recipient([{'email': email}], subject, email_data, footer=None,	1✔
111	snippet='email/content/notification_platform/verify_email.html')
112
UNCOV 113	except tk.ValidationError as e:	×
114	return _build_json_response(	×
115	{
116	'success': False,
117	'error': {
118	'message': e.error_summary
119	}
120	}
121	)
UNCOV 122	except tk.Invalid as e:	×
123	return _build_json_response(	×
124	{
125	'success': False,
126	'error': {
127	'message': e.error
128	}
129	}
130	)
UNCOV 131	except MailerException as e:	×
132	log.error(e)	×
133	return _build_json_response(	×
134	{
135	'success': False,
136	'error': {
137	'message': u'Error sending the confirmation email, please try again.'
138	}
139	}
140	)
UNCOV 141	except Exception as e:	×
142	log.error(e)	×
143	return _build_json_response(	×
144	{
145	'success': False,
146	'error': {
147	'message': str(e)
148	}
149	}
150	)
151	return _build_json_response({'success': True})	1✔
152
153
154	def unsubscribe_confirmation() -> Response:	1✔
155	token = tk.request.form.get('token')	1✔
156
157	try:	1✔
158	token_obj = notification_platform_logic.verify_unsubscribe_token(token, inactivate=True)	1✔
159
160	context = {'ignore_auth': True}	1✔
161	data_dict = {'email': token_obj.user_id, 'dataset_id': token_obj.object_id}	1✔
162	result = _delete_notification_subscription(context, data_dict)	1✔
UNCOV 163	except tk.ValidationError as e:	×
164	log.error('An exception occurred:' + str(e))	×
165	return _build_json_response(	×
166	{
167	'success': False,
168	'error': {
169	'message': 'An exception occurred:' + str(e)
170	}
171	}
172	)
UNCOV 173	except Exception as e:	×
174	log.error('An exception occurred:' + str(e))	×
175	return _build_json_response(	×
176	{
177	'success': False,
178	'error': {
179	'message': 'An error occurred: ' + str(e)
180	}
181	}
182	)
183	return _build_json_response({'success': True})	1✔
184
185
186	def _delete_notification_subscription(context: Context, data_dict: DataDict) -> DataDict:	1✔
UNCOV 187	result = tk.get_action('hdx_delete_notification_subscription')(context, data_dict)	×
188	return result	×
189
190	def _build_json_response(data_dict: DataDict, status=200):	1✔
191	headers = {	1✔
192	'Content-Type': CONTENT_TYPES['json'],
193	}
194	body = json.dumps(data_dict)	1✔
195	response = make_response((body, status, headers))	1✔
196	return response	1✔
197
198
199	hdx_notifications.add_url_rule(u'/subscribe-to-dataset', view_func=subscribe_to_dataset)	1✔
200	hdx_notifications.add_url_rule(u'/subscription-confirmation', view_func=subscription_confirmation, methods=['POST'])	1✔
201	hdx_notifications.add_url_rule(u'/unsubscribe-confirmation', view_func=unsubscribe_confirmation, methods=['POST'])	1✔

OCHA-DAP / hdx-ckan / #5935

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