#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

84.15

/ckanext-hdx_package/ckanext/hdx_package/views/light_dataset.py

from flask import Blueprint

import ckan.model as model
import ckan.plugins.toolkit as tk

from ckan.common import g
from ckan.types import DataDict, Request
import ckan.logic as logic

import ckanext.hdx_package.helpers.analytics as analytics
import ckanext.hdx_package.helpers.custom_pages as cp_h
from ckanext.hdx_search.controller_logic.search_logic import SearchLogic, ArchivedUrlHelper
from ckanext.hdx_theme.util.http_exception_helper import catch_http_exceptions
from ckanext.hdx_theme.util.light_redirect import check_redirect_needed
from ckanext.hdx_users.controller_logic.notification_platform_logic import verify_unsubscribe_token
from ckanext.hdx_users.helpers.notification_platform import check_notifications_enabled_for_dataset

get_action = tk.get_action
check_access = tk.check_access
render = tk.render
abort = tk.abort
_ = tk._

NotAuthorized = tk.NotAuthorized
NotFound = logic.NotFound

hdx_light_dataset = Blueprint(u'hdx_light_dataset', __name__, url_prefix=u'/m/dataset')
hdx_light_search = Blueprint(u'hdx_light_search', __name__, url_prefix=u'/m/search')


def _get_org_extras(org_id):
    """
    Get the extras for our orgs
    """
    if not org_id:
        return {}
    context = {'model': model, 'session': model.Session,
               'user': g.user or g.author,
               'include_datasets': False,
               'for_view': True}
    data_dict = {'id': org_id}
    org_info = get_action(
        'hdx_light_group_show')(context, data_dict)

    extras_dict = {item['key']: item['value'] for item in org_info.get('extras', {})}
    extras_dict['image_url'] = org_info.get('image_url', None)

    return extras_dict

@check_redirect_needed
@catch_http_exceptions
def read(id):
    context = {
        u'model': model,
        u'session': model.Session,
        u'user': g.user,
        u'auth_user_obj': g.userobj,
        u'for_view': True
    }
    data_dict = {
        u'id': id
    }

    dataset_dict = get_action('package_show')(context, data_dict)
    org_dict = dataset_dict.get('organization') or {}
    org_id = org_dict.get('id', None)
    org_info_dict = _get_org_extras(org_id)
    user_survey_url = org_info_dict.get('user_survey_url')
    if dataset_dict.get('type') == 'dataset':
        analytics_dict = _compute_analytics(dataset_dict, tk.request)
        dataset_dict['page_list'] = cp_h.hdx_get_page_list_for_dataset(context, dataset_dict)
        dataset_dict['link_list'] = get_action('hdx_package_links_by_id_list')(context, {'id': dataset_dict.get('name')})

        # notification platform
        supports_notifications = check_notifications_enabled_for_dataset(dataset_dict['id'])
        unsubscribe_token = tk.request.args.get('unsubscribe_token', None)
        if unsubscribe_token:
            try:
                token_obj = verify_unsubscribe_token(unsubscribe_token, inactivate=False)
            except Exception as e:
                unsubscribe_token = None
                tk.h.flash_error('Your token is invalid or has expired.')

        template_data = {
            'dataset_dict': dataset_dict,
            'analytics': analytics_dict,
            'user_survey_url': user_survey_url,
            'unsubscribe_token': unsubscribe_token,
            'supports_notifications': supports_notifications,
        }

        return render(u'light/dataset/read.html', template_data)
    else:
        raise NotFound


@check_redirect_needed
def search():
    return generic_search(u'light/search/search.html')


def generic_search(html_template):
    try:
        context = {'model': model, 'user': g.user,
                   'auth_user_obj': g.userobj}
        check_access('site_read', context)
    except NotAuthorized:
        abort(403, _('Not authorized to see this page'))

    search_logic = SearchLogic()

    try:
        search_logic._search(use_solr_collapse=True)
    except NotFound as ex:
        abort(404, _('Page not found'))

    archived_url_helper = search_logic.add_archived_url_helper()
    redirect_result = archived_url_helper.redirect_if_needed()
    if redirect_result:
        return redirect_result

    data_dict = {'data': search_logic.template_data}
    return render(html_template, data_dict)


def _compute_analytics(dataset_dict: DataDict, request: Request):
    result = {
        'is_cod': analytics.is_cod(dataset_dict),
        'is_indicator': analytics.is_indicator(dataset_dict),
        'is_archived': analytics.is_archived(dataset_dict),
        'analytics_group_names': (analytics.extract_locations_in_json(dataset_dict))[0],
        'analytics_group_ids': (analytics.extract_locations_in_json(dataset_dict))[1],
        'analytics_dataset_availability': analytics.dataset_availability(dataset_dict),
        'analytics_came_from': analytics.came_from(request.args),
        'analytics_supports_notifications': analytics.supports_notifications(dataset_dict),
    }
    return result


hdx_light_search.add_url_rule(u'/', view_func=search, strict_slashes=False)
hdx_light_dataset.add_url_rule(u'/', view_func=search, strict_slashes=False)
hdx_light_dataset.add_url_rule(u'/<id>', view_func=read)

1	from flask import Blueprint	1✔
2
3	import ckan.model as model	1✔
4	import ckan.plugins.toolkit as tk	1✔
5
6	from ckan.common import g	1✔
7	from ckan.types import DataDict, Request	1✔
8	import ckan.logic as logic	1✔
9
10	import ckanext.hdx_package.helpers.analytics as analytics	1✔
11	import ckanext.hdx_package.helpers.custom_pages as cp_h	1✔
12	from ckanext.hdx_search.controller_logic.search_logic import SearchLogic, ArchivedUrlHelper	1✔
13	from ckanext.hdx_theme.util.http_exception_helper import catch_http_exceptions	1✔
14	from ckanext.hdx_theme.util.light_redirect import check_redirect_needed	1✔
15	from ckanext.hdx_users.controller_logic.notification_platform_logic import verify_unsubscribe_token	1✔
16	from ckanext.hdx_users.helpers.notification_platform import check_notifications_enabled_for_dataset	1✔
17
18	get_action = tk.get_action	1✔
19	check_access = tk.check_access	1✔
20	render = tk.render	1✔
21	abort = tk.abort	1✔
22	_ = tk._	1✔
23
24	NotAuthorized = tk.NotAuthorized	1✔
25	NotFound = logic.NotFound	1✔
26
27	hdx_light_dataset = Blueprint(u'hdx_light_dataset', __name__, url_prefix=u'/m/dataset')	1✔
28	hdx_light_search = Blueprint(u'hdx_light_search', __name__, url_prefix=u'/m/search')	1✔
29
30
31	def _get_org_extras(org_id):	1✔
32	"""
33	Get the extras for our orgs
34	"""
35	if not org_id:	1✔
36	return {}	×
37	context = {'model': model, 'session': model.Session,	1✔
38	'user': g.user or g.author,
39	'include_datasets': False,
40	'for_view': True}
41	data_dict = {'id': org_id}	1✔
42	org_info = get_action(	1✔
43	'hdx_light_group_show')(context, data_dict)
44
45	extras_dict = {item['key']: item['value'] for item in org_info.get('extras', {})}	1✔
46	extras_dict['image_url'] = org_info.get('image_url', None)	1✔
47
48	return extras_dict	1✔
49
50	@check_redirect_needed	1✔
51	@catch_http_exceptions	1✔
52	def read(id):	1✔
53	context = {	1✔
54	u'model': model,
55	u'session': model.Session,
56	u'user': g.user,
57	u'auth_user_obj': g.userobj,
58	u'for_view': True
59	}
60	data_dict = {	1✔
61	u'id': id
62	}
63
64	dataset_dict = get_action('package_show')(context, data_dict)	1✔
65	org_dict = dataset_dict.get('organization') or {}	1✔
66	org_id = org_dict.get('id', None)	1✔
67	org_info_dict = _get_org_extras(org_id)	1✔
68	user_survey_url = org_info_dict.get('user_survey_url')	1✔
69	if dataset_dict.get('type') == 'dataset':	1✔
70	analytics_dict = _compute_analytics(dataset_dict, tk.request)	1✔
71	dataset_dict['page_list'] = cp_h.hdx_get_page_list_for_dataset(context, dataset_dict)	1✔
72	dataset_dict['link_list'] = get_action('hdx_package_links_by_id_list')(context, {'id': dataset_dict.get('name')})	1✔
73
74	# notification platform
75	supports_notifications = check_notifications_enabled_for_dataset(dataset_dict['id'])	1✔
76	unsubscribe_token = tk.request.args.get('unsubscribe_token', None)	1✔
77	if unsubscribe_token:	1✔
UNCOV 78	try:	×
79	token_obj = verify_unsubscribe_token(unsubscribe_token, inactivate=False)	×
80	except Exception as e:	×
81	unsubscribe_token = None	×
82	tk.h.flash_error('Your token is invalid or has expired.')	×
83
84	template_data = {	1✔
85	'dataset_dict': dataset_dict,
86	'analytics': analytics_dict,
87	'user_survey_url': user_survey_url,
88	'unsubscribe_token': unsubscribe_token,
89	'supports_notifications': supports_notifications,
90	}
91
92	return render(u'light/dataset/read.html', template_data)	1✔
93	else:
UNCOV 94	raise NotFound	×
95
96
97	@check_redirect_needed	1✔
98	def search():	1✔
UNCOV 99	return generic_search(u'light/search/search.html')	×
100
101
102	def generic_search(html_template):	1✔
103	try:	1✔
104	context = {'model': model, 'user': g.user,	1✔
105	'auth_user_obj': g.userobj}
106	check_access('site_read', context)	1✔
UNCOV 107	except NotAuthorized:	×
UNCOV 108	abort(403, _('Not authorized to see this page'))	×
109
110	search_logic = SearchLogic()	1✔
111
112	try:	1✔
113	search_logic._search(use_solr_collapse=True)	1✔
UNCOV 114	except NotFound as ex:	×
UNCOV 115	abort(404, _('Page not found'))	×
116
117	archived_url_helper = search_logic.add_archived_url_helper()	1✔
118	redirect_result = archived_url_helper.redirect_if_needed()	1✔
119	if redirect_result:	1✔
UNCOV 120	return redirect_result	×
121
122	data_dict = {'data': search_logic.template_data}	1✔
123	return render(html_template, data_dict)	1✔
124
125
126	def _compute_analytics(dataset_dict: DataDict, request: Request):	1✔
127	result = {	1✔
128	'is_cod': analytics.is_cod(dataset_dict),
129	'is_indicator': analytics.is_indicator(dataset_dict),
130	'is_archived': analytics.is_archived(dataset_dict),
131	'analytics_group_names': (analytics.extract_locations_in_json(dataset_dict))[0],
132	'analytics_group_ids': (analytics.extract_locations_in_json(dataset_dict))[1],
133	'analytics_dataset_availability': analytics.dataset_availability(dataset_dict),
134	'analytics_came_from': analytics.came_from(request.args),
135	'analytics_supports_notifications': analytics.supports_notifications(dataset_dict),
136	}
137	return result	1✔
138
139
140	hdx_light_search.add_url_rule(u'/', view_func=search, strict_slashes=False)	1✔
141	hdx_light_dataset.add_url_rule(u'/', view_func=search, strict_slashes=False)	1✔
142	hdx_light_dataset.add_url_rule(u'/<id>', view_func=read)	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