#5649

Committed 26 Aug 2024 11:34AM UTC coverage: 72.882% (-0.1%) from 72.998%

Build # #5649

Build Type

Pull #6412

coveralls-python

Committed by

web-flow

Commit Message

Merge branch 'dev' into feature/HDX-9990-implement-new-contact-contributor-form

Pull Request Pull Request #6412: HDX-9960 & HDX-9987 contact the contributor & HDX Connect new pages

Run Details

146 of 321 new or added lines in 10 files covered. (45.48%)

1 existing line in 1 file now uncovered.

11750 of 16122 relevant lines covered (72.88%)

0.73 hits per line

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

76.14

/ckanext-hdx_package/ckanext/hdx_package/actions/authorize.py

import logging
import ckan.authz as new_authz
import ckan.logic.auth.create as create
import ckan.logic.auth.update as update
import ckan.plugins.toolkit as tk
from ckan.types import Context, DataDict
from ckanext.hdx_users.helpers.permissions import Permissions

log = logging.getLogger(__name__)
get_action = tk.get_action
auth_allow_anonymous_access = tk.auth_allow_anonymous_access
_ = tk._


def package_create(context, data_dict=None):
    retvalue = True
    if data_dict and 'groups' in data_dict:
        temp_groups = data_dict['groups']
        del data_dict['groups']
        # check original package_create auth
        log.debug('Removed groups from data_dict: ' + str(data_dict))
        retvalue = create.package_create(context, data_dict)
        data_dict['groups'] = temp_groups
    else:
        retvalue = create.package_create(context, data_dict)

    return retvalue


def package_update(context, data_dict=None):
    retvalue = True
    if data_dict and 'groups' in data_dict:
        temp_groups = data_dict['groups']
        del data_dict['groups']
        # check original package_create auth
        log.debug('Removed groups from data_dict: ' + str(data_dict))
        retvalue = update.package_update(context, data_dict)
        data_dict['groups'] = temp_groups
    else:
        retvalue = update.package_update(context, data_dict)

    return retvalue


def hdx_resource_id_list(context, data_dict=None):
    '''
    Only sysadmins are allowed to call this action
    '''
    return {'success': False, 'msg': _('Only sysadmins can get the entire list of resource ids')}


def hdx_send_mail_contributor(context, data_dict):
    '''
    Only a logged in user has access.
    '''

    user_obj = context.get('auth_user_obj') or context.get('user_obj')
    if user_obj:
        return {
            'success': True
        }

    return {
        'success': False,
        'msg': _('Not authorized to perform this request')
    }


def hdx_send_mail_members(context, data_dict):
    '''
    Only a logged in user has access and member of dataset's owner_org .
    '''

    user_obj = context.get('auth_user_obj') or context.get('user_obj')
    if user_obj:
        org_members = get_action('hdx_member_list')(context, {'org_id': data_dict.get('org_id')})
        if org_members and org_members.get('is_member'):
            return {
                'success': True
            }

    return {
        'success': False,
        'msg': _('Not authorized to perform this request')
    }


# def hdx_create_screenshot_for_cod(context, data_dict=None):
#     '''
#     Only sysadmins are allowed to call this action
#     '''
#     return {'success': False, 'msg': _('Only sysadmins can create a screenshot of a dataset\'s viz')}


@auth_allow_anonymous_access
def hdx_resource_download(context, resource_dict):
    if resource_dict.get('in_quarantine', False):
        return {'success': False, 'msg': _('Only sysadmins can download quarantined resources')}
    return {'success': True}


def hdx_mark_qa_completed(context, data_dict=None):
    username_or_id = context.get('user')
    result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_QA)
    return {'success': result}


def hdx_mark_resource_in_quarantine(context, data_dict=None):
    username_or_id = context.get('user')
    result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_QA)
    return {'success': result}


def hdx_qa_resource_patch(context, data_dict=None):
    '''
    Only sysadmins are allowed to call this action
    '''
    return {'success': False, 'msg': _('Only sysadmins can change the qa script related flags')}


def hdx_fs_check_resource_revise(context, data_dict=None):
    '''
    Only sysadmins are allowed to call this action
    '''
    return {'success': False, 'msg': _('Only sysadmins can change the file structure check info')}

def hdx_qa_hapi_report_view(context, data_dict=None):
    '''
    Only sysadmins are allowed to call this action
    '''
    return {'success': False, 'msg': _('Only sysadmins can change the file structure check info')}

def hdx_cod_update(context, data_dict):
    return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_COD)


def hdx_dataseries_update(context, data_dict):
    return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_DATASERIES)


def _check_hdx_user_permission(context, permission):
    username_or_id = context.get('user')
    result = Permissions(username_or_id).has_permission(permission)
    return {'success': result}


def hdx_p_coded_resource_update(context, data_dict):
    username_or_id = context.get('user')
    result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_P_CODES)
    return {'success': result}


def hdx_send_mail_request_tags(context, data_dict):
    '''
    Only a logged in user has access.
    '''

    user_obj = context.get('auth_user_obj') or context.get('user_obj')
    if user_obj:
        return {
            'success': True
        }

    return {
        'success': False,
        'msg': _('Not authorized to perform this request')
    }


def hdx_mark_resource_in_hapi(context: Context, data_dict: DataDict):
    return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_IN_HAPI_FLAG)


def hdx_request_access(context: Context):
    """
    Only a logged-in user can request data access.
    """

    user_obj = context.get('auth_user_obj') or context.get('user_obj')
    if user_obj:
        return {'success': True}

    return {'success': False, 'msg': _('Not authorized to perform this request.')}

1	import logging	1✔
2	import ckan.authz as new_authz	1✔
3	import ckan.logic.auth.create as create	1✔
4	import ckan.logic.auth.update as update	1✔
5	import ckan.plugins.toolkit as tk	1✔
6	from ckan.types import Context, DataDict	1✔
7	from ckanext.hdx_users.helpers.permissions import Permissions	1✔
8
9	log = logging.getLogger(__name__)	1✔
10	get_action = tk.get_action	1✔
11	auth_allow_anonymous_access = tk.auth_allow_anonymous_access	1✔
12	_ = tk._	1✔
13
14
15	def package_create(context, data_dict=None):	1✔
16	retvalue = True	1✔
17	if data_dict and 'groups' in data_dict:	1✔
18	temp_groups = data_dict['groups']	1✔
19	del data_dict['groups']	1✔
20	# check original package_create auth
21	log.debug('Removed groups from data_dict: ' + str(data_dict))	1✔
22	retvalue = create.package_create(context, data_dict)	1✔
23	data_dict['groups'] = temp_groups	1✔
24	else:
25	retvalue = create.package_create(context, data_dict)	1✔
26
27	return retvalue	1✔
28
29
30	def package_update(context, data_dict=None):	1✔
31	retvalue = True	1✔
32	if data_dict and 'groups' in data_dict:	1✔
33	temp_groups = data_dict['groups']	1✔
34	del data_dict['groups']	1✔
35	# check original package_create auth
36	log.debug('Removed groups from data_dict: ' + str(data_dict))	1✔
37	retvalue = update.package_update(context, data_dict)	1✔
38	data_dict['groups'] = temp_groups	1✔
39	else:
40	retvalue = update.package_update(context, data_dict)	1✔
41
42	return retvalue	1✔
43
44
45	def hdx_resource_id_list(context, data_dict=None):	1✔
46	'''
47	Only sysadmins are allowed to call this action
48	'''
49	return {'success': False, 'msg': _('Only sysadmins can get the entire list of resource ids')}	×
50
51
52	def hdx_send_mail_contributor(context, data_dict):	1✔
53	'''
54	Only a logged in user has access.
55	'''
56
57	user_obj = context.get('auth_user_obj') or context.get('user_obj')	1✔
58	if user_obj:	1✔
59	return {	1✔
60	'success': True
61	}
62
63	return {	×
64	'success': False,
65	'msg': _('Not authorized to perform this request')
66	}
67
68
69	def hdx_send_mail_members(context, data_dict):	1✔
70	'''
71	Only a logged in user has access and member of dataset's owner_org .
72	'''
73
74	user_obj = context.get('auth_user_obj') or context.get('user_obj')	×
75	if user_obj:	×
76	org_members = get_action('hdx_member_list')(context, {'org_id': data_dict.get('org_id')})	×
77	if org_members and org_members.get('is_member'):	×
78	return {	×
79	'success': True
80	}
81
82	return {	×
83	'success': False,
84	'msg': _('Not authorized to perform this request')
85	}
86
87
88	# def hdx_create_screenshot_for_cod(context, data_dict=None):
89	# '''
90	# Only sysadmins are allowed to call this action
91	# '''
92	# return {'success': False, 'msg': _('Only sysadmins can create a screenshot of a dataset\'s viz')}
93
94
95	@auth_allow_anonymous_access	1✔
96	def hdx_resource_download(context, resource_dict):	1✔
97	if resource_dict.get('in_quarantine', False):	1✔
98	return {'success': False, 'msg': _('Only sysadmins can download quarantined resources')}	×
99	return {'success': True}	1✔
100
101
102	def hdx_mark_qa_completed(context, data_dict=None):	1✔
103	username_or_id = context.get('user')	×
104	result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_QA)	×
105	return {'success': result}	×
106
107
108	def hdx_mark_resource_in_quarantine(context, data_dict=None):	1✔
109	username_or_id = context.get('user')	1✔
110	result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_QA)	1✔
111	return {'success': result}	1✔
112
113
114	def hdx_qa_resource_patch(context, data_dict=None):	1✔
115	'''
116	Only sysadmins are allowed to call this action
117	'''
118	return {'success': False, 'msg': _('Only sysadmins can change the qa script related flags')}	1✔
119
120
121	def hdx_fs_check_resource_revise(context, data_dict=None):	1✔
122	'''
123	Only sysadmins are allowed to call this action
124	'''
125	return {'success': False, 'msg': _('Only sysadmins can change the file structure check info')}	×
126
127	def hdx_qa_hapi_report_view(context, data_dict=None):	1✔
128	'''
129	Only sysadmins are allowed to call this action
130	'''
131	return {'success': False, 'msg': _('Only sysadmins can change the file structure check info')}	1✔
132
133	def hdx_cod_update(context, data_dict):	1✔
134	return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_COD)	1✔
135
136
137	def hdx_dataseries_update(context, data_dict):	1✔
138	return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_DATASERIES)	1✔
139
140
141	def _check_hdx_user_permission(context, permission):	1✔
142	username_or_id = context.get('user')	1✔
143	result = Permissions(username_or_id).has_permission(permission)	1✔
144	return {'success': result}	1✔
145
146
147	def hdx_p_coded_resource_update(context, data_dict):	1✔
148	username_or_id = context.get('user')	1✔
149	result = Permissions(username_or_id).has_permission(Permissions.PERMISSION_MANAGE_P_CODES)	1✔
150	return {'success': result}	1✔
151
152
153	def hdx_send_mail_request_tags(context, data_dict):	1✔
154	'''
155	Only a logged in user has access.
156	'''
157
158	user_obj = context.get('auth_user_obj') or context.get('user_obj')	×
159	if user_obj:	×
160	return {	×
161	'success': True
162	}
163
164	return {	×
165	'success': False,
166	'msg': _('Not authorized to perform this request')
167	}
168
169
170	def hdx_mark_resource_in_hapi(context: Context, data_dict: DataDict):	1✔
171	return _check_hdx_user_permission(context, Permissions.PERMISSION_MANAGE_IN_HAPI_FLAG)	1✔
172
173
174	def hdx_request_access(context: Context):	1✔
175	"""
176	Only a logged-in user can request data access.
177	"""
178
NEW 179	user_obj = context.get('auth_user_obj') or context.get('user_obj')	×
NEW 180	if user_obj:	×
NEW 181	return {'success': True}	×
182
NEW 183	return {'success': False, 'msg': _('Not authorized to perform this request.')}	×

OCHA-DAP / hdx-ckan / #5649

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