#5847

Committed 09 Nov 2024 05:10PM UTC coverage: 74.501% (+0.4%) from 74.062%

Build # #5847

Build Type

Pull #6470

coveralls-python

Committed by

danmihaila

Commit Message

HDX-10191 replace xls with xlsx label

Pull Request Pull Request #6470: HDX-10191 org stats download as xlsx

Run Details

75 of 115 new or added lines in 5 files covered. (65.22%)

1 existing line in 1 file now uncovered.

12403 of 16648 relevant lines covered (74.5%)

0.75 hits per line

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

53.51

/ckanext-hdx_org_group/ckanext/hdx_org_group/views/organization.py

import logging

from flask import Blueprint
from six.moves.urllib.parse import urlencode
from ckan.types import Context
import ckan.lib.plugins as lib_plugins
import ckan.model as model
import ckan.plugins.toolkit as tk
import ckanext.hdx_org_group.helpers.org_meta_dao as org_meta_dao
import ckanext.hdx_org_group.helpers.organization_helper as helper
import ckanext.hdx_org_group.helpers.static_lists as static_lists
import ckanext.hdx_theme.helpers.helpers as hdx_helpers
from ckan.views.group import CreateGroupView, EditGroupView, _get_group_template
from ckanext.hdx_org_group.controller_logic.organization_read_logic import OrgReadLogic
from ckanext.hdx_org_group.controller_logic.organization_stats_logic import (
    OrganizationStatsLogic,
)
from ckanext.hdx_org_group.views.light_organization import _index
from ckanext.hdx_theme.util.light_redirect import check_redirect_needed
from ckanext.hdx_theme.util.mail import NoRecipientException

g = tk.g
config = tk.config
request = tk.request
render = tk.render
redirect = tk.redirect_to
url_for = tk.url_for
get_action = tk.get_action
check_access = tk.check_access
NotFound = tk.ObjectNotFound
NotAuthorized = tk.NotAuthorized
ValidationError = tk.ValidationError
abort = tk.abort
_ = tk._
h = tk.h

log = logging.getLogger(__name__)

hdx_org = Blueprint(u'hdx_org', __name__, url_prefix=u'/organization')


@check_redirect_needed
def index():
    return _index('organization/index.html', False, True)


@check_redirect_needed
def read(id):
    context: Context = {
        'model': model,
        'session': model.Session,
        'for_view': True,
        'with_private': False
    }
    try:
        check_access('site_read', context)
    except NotAuthorized:
        abort(403, _('Not authorized to see this page'))

    try:
        read_logic = OrgReadLogic(id, g.user, g.userobj)
        read_logic.read()
        if read_logic.redirect_result:
            return read_logic.redirect_result

        if read_logic.org_meta.is_custom:
            template_data = _generate_template_data_for_custom_org(read_logic)
            result = render('organization/custom/custom_org.html', template_data)
            return result
        else:
            org_dict = read_logic.org_meta.org_dict
            org_dict.update({
                'search_template_data': read_logic.search_template_data,
                'datasets_num': read_logic.search_template_data.get('facets').get('extras_archived').get('fals'),
                'archived_package_count': read_logic.search_template_data.get('facets').get('extras_archived').get('true'),
                'allow_req_membership': read_logic.org_meta.allow_req_membership,
                # 'group_message_info': read_logic.org_meta.group_message_info,
            })

            template_data = {
                'org_dict': org_dict,
            }
            template_file = _get_group_template('read_template', 'organization')
            return render(template_file, template_data)
    except NotFound:
        abort(404, _('Page not found'))
    except NotAuthorized:
        abort(403, _('Not authorized to see this page'))


def _generate_template_data_for_custom_org(org_read_logic):
    """
    :param org_read_logic:
    :type org_read_logic: OrgReadLogic
    :returns: the template data dict
    :rtype: dict
    """
    org_meta = org_read_logic.org_meta
    org_dict = org_meta.org_dict
    org_id = org_dict['id']

    # org_dict['group_message_info'] = org_meta.group_message_info
    template_data = {
        'data': {
            'org_info': {
                'id': org_id,
                'display_name': org_dict.get('display_name', ''),
                'description': org_dict.get('description'),
                'name': org_dict['name'],
                'link': org_dict.get('extras', {}).get('org_url'),
                # 'topline_resource': org_meta.customization.get('topline_resource'),
                'modified_at': org_dict.get('modified_at', ''),
                'image_sq': org_meta.customization.get('image_sq'),
                'image_rect': org_meta.customization.get('image_rect'),
                # 'visualization_config': result.get('visualization_config', ''),
            },
            'search_template_data': org_read_logic.search_template_data,
            #'custom_css_path': org_read_logic.org_meta.custom_css_path,
            # 'member_count': hdx_helpers.get_group_members(org_id),
            'follower_count': org_read_logic.follower_count,
            'top_line_items': org_read_logic.top_line_items,
            # 'search_results': {
            # 'facets': facets,
            # 'activities': activities,
            # 'query_placeholder': query_placeholder
            # },
            # 'links': {
            #     'edit': org_read_logic.links.edit,
            #     'members': org_read_logic.links.members,
            #     'request_membership': org_read_logic.links.request_membership,
            #     'add_data': org_read_logic.links.add_data
            # },
            'request_params': request.params,
            'permissions': {
                'edit': org_read_logic.allow_edit,
                'add_dataset': org_read_logic.allow_add_dataset,
                'view_members': org_read_logic.allow_basic_user_info,
                'request_membership': org_read_logic.allow_req_membership
            },
            'show_admin_menu': org_read_logic.allow_add_dataset or org_read_logic.allow_edit,
            'show_visualization': 'Choose Visualization Type' != org_read_logic.viz_config.get('type'),
            'visualization': {
                'config': org_read_logic.viz_config,
                'config_type': org_read_logic.viz_config.get('type'),
                'config_url': urlencode(org_read_logic.viz_config, True),
                # 'embed_url': org_read_logic.links.embed_url,

            },

            # This is here for compatibility with the custom_org_header.html template, which is still
            # used from pylon controllers
            'org_meta': {
                'id': org_dict['name'],
                'custom_rect_logo_url': org_meta.custom_rect_logo_url,
                'custom_sq_logo_url': org_meta.custom_sq_logo_url,
                'followers_num': org_meta.followers_num,
                'members_num': org_meta.members_num,
                'allow_req_membership': org_meta.allow_req_membership,
                'allow_basic_user_info': org_meta.allow_basic_user_info,
                'allow_add_dataset': org_meta.allow_add_dataset,
                'allow_edit': org_meta.allow_edit,
                'org_dict': org_dict,
            },

        },
        'errors': org_read_logic.errors,
        'error_summary': org_read_logic.error_summary,

    }
    if template_data['data']['show_visualization']:
        template_data['data']['show_visualization'] = \
            hdx_helpers.check_all_str_fields_not_empty(template_data['data']['visualization'],
                                                       'Visualization config field "{}" is empty',
                                                       skipped_keys=['config'],
                                                       errors=template_data['errors'])
    return template_data


def request_new():
    context: Context = {'model': model, 'session': model.Session, 'user': g.user}
    try:
        check_access('hdx_send_new_org_request', context)
    except NotAuthorized:
        abort(403, _('Unauthorized to send a new org request'))

    errors = {}
    error_summary = {}
    data = {'from': request.form.get('from', '')}

    sent_successfully = False
    if 'save' in request.form and request.method == 'POST':
        try:
            data = _process_new_org_request()
            _validate_new_org_request_field(data)

            get_action('hdx_send_new_org_request')(context, _transform_dict_for_mailing(data))

            data.clear()
            h.flash_success(_('Request sent successfully'))
            sent_successfully = True
        except NoRecipientException as e:
            h.flash_error(_(str(e)))
        except ValidationError as e:
            errors = e.error_dict
            error_summary = e.error_summary
        except Exception as e:
            log.error(str(e))
            h.flash_error(_('Request can not be sent. Contact an administrator'))
        if sent_successfully:
            return h.redirect_to('dashboard.organizations')

    hdx_org_type_list = [{'value': '-1', 'text': _('-- Please select --')}] + \
                        [{'value': t[1], 'text': _(t[0])} for t in static_lists.ORGANIZATION_TYPE_LIST]
    template_data = {
        'data': {
            'action': 'new',
            'hdx_org_type_list': hdx_org_type_list
        },
        'form_data': data,
        'errors': errors,
        'error_summary': error_summary,

    }
    g.form = render('organization/request_organization_form.html', template_data)
    return render('organization/request_new.html')


def _process_new_org_request():
    hdx_org_type = None
    hdx_org_type_code = request.form.get('hdx_org_type', '')

    if hdx_org_type_code:
        hdx_org_type = next(
            (_type[0] for _type in static_lists.ORGANIZATION_TYPE_LIST if _type[1] == hdx_org_type_code), '-1')

    data = {
        'name': request.form.get('name', ''),
        'title': request.form.get('title', ''),
        'org_url': request.form.get('org_url', ''),
        'description': request.form.get('description', ''),
        'your_email': request.form.get('your_email', ''),
        'your_name': request.form.get('your_name', ''),
        'org_acronym': request.form.get('org_acronym', ''),
        'hdx_org_type': hdx_org_type,
        'hdx_org_type_code': hdx_org_type_code,  # This is needed for the form when validation fails
    }
    return data


def _validate_new_org_request_field(data):
    errors = {}
    for field in ['title', 'description', 'your_email', 'your_name', 'hdx_org_type']:
        if data[field].strip() in ['', '-1']:
            errors[field] = [_('should not be empty')]

    if len(errors) > 0:
        raise ValidationError(errors)


def _transform_dict_for_mailing(data_dict):
    data_dict_for_mailing = {
        'name': data_dict.get('title'),
        'acronym': data_dict.get('org_acronym'),
        'description': data_dict.get('description'),
        'org_type': data_dict.get('hdx_org_type', ''),
        'org_url': data_dict.get('org_url', ''),

        'work_email': data_dict.get('your_email', ''),
        'your_name': data_dict.get('your_name', ''),
    }
    return data_dict_for_mailing


def new_org_template_variables(data_dict):
    data_dict['hdx_org_type_list'] = [{'value': '-1', 'text': _('-- Please select --')}] + \
                              [{'value': t[1], 'text': _(t[0])} for t in static_lists.ORGANIZATION_TYPE_LIST]


def stats(id):
    stats_logic = OrganizationStatsLogic(id, g.user, g.userobj)
    org_dict = stats_logic.org_meta_dao.org_dict
    org_dict.update({
        'allow_req_membership': stats_logic.org_meta_dao.allow_req_membership,
        # 'group_message_info': stats_logic.org_meta_dao.group_message_info,
    })
    template_data = {
        'data': stats_logic.fetch_stats(),
        'org_meta': stats_logic.org_meta_dao,
        'org_dict': org_dict,
    }

    if stats_logic.is_custom():
        return render('organization/custom_stats.html', template_data)
    else:
        return render('organization/stats.html', template_data)


def restore(id):
    context: Context = {
        'model': model, 'session': model.Session,
        'user': g.user,
        'for_edit': True,
    }

    try:
        check_access('organization_patch', context, {'id': id})
    except NotAuthorized:
        return abort(403, _('Unauthorized to restore this organization'))

    try:
        get_action('organization_patch')(context, {
            'id': id,
            'state': 'active'
        })
        return redirect('organization.read', id=id)
    except NotAuthorized:
        return abort(403, _('Unauthorized to read group %s') % id)
    except NotFound:
        return abort(404, _('Group not found'))
    except ValidationError as e:
        errors = e.error_dict
        error_summary = e.error_summary
        core_view = EditGroupView()
        return core_view.get(id, 'organization', True, errors=errors, error_summary=error_summary)


def activity(id):
    return activity_offset(id)


def activity_offset(id, offset=0):
    """
     Modified core functionality to use the new OrgMetaDao class
    for fetching information needed on all org-related pages.

    Render this group's public activity stream page.

    :param id:
    :type id: str
    :param offset:
    :type offset: int
    :return:
    """
    org_meta = org_meta_dao.OrgMetaDao(id, g.user, g.userobj)
    org_meta.fetch_all()
    org_dict = org_meta.org_dict
    # org_dict['group_message_info'] = org_meta.group_message_info

    helper.org_add_last_updated_field([org_dict])

    # Add the group's activity stream (already rendered to HTML) to the
    # template context for the group/read.html template to retrieve later.
    context: Context = {'model': model, 'session': model.Session,
               'user': g.user, 'for_view': True}
    group_activity_stream = get_action('organization_activity_list')(
        context, {'id': org_dict['id'], 'offset': offset})

    extra_vars = {
        'org_dict': org_dict,
        'org_meta': org_meta,
        'group_activity_stream': group_activity_stream,

    }
    if org_meta.is_custom:
        template = 'organization/custom_activity_stream.html'
    else:
        template = lib_plugins.lookup_group_plugin('organization').activity_template()
    return render(template, extra_vars)

def download_organization_stats(id):
    """
        Handles downloading .xlsx organization stats

        :returns: xlsx
    """

    context: Context = {
        'model': model,
        'session': model.Session,
        'user': g.user or g.author,
        'auth_user_obj': g.userobj
    }

    try:
        check_access('organization_update', context, {'id': id})
    except NotAuthorized:
        return abort(403, _('Unauthorized to restore this organization'))

    # check if organization exists
    try:
        org_dict = get_action('organization_show')(context, {'id': id})
        output = helper.hdx_generate_organization_stats(org_dict)
        return output

    except NotFound:
        return abort(404, _('Organization not found'))
    except NotAuthorized:
        return abort(404, _('Organization not found'))
    except Exception as e:
        return abort(404, _('Something went wrong, please contact us'))


hdx_org.add_url_rule(u'/', view_func=index, strict_slashes=False)
hdx_org.add_url_rule(
        u'/new',
        methods=[u'GET', u'POST'],
        view_func=CreateGroupView.as_view(str(u'new')),
        defaults={
            'group_type': 'organization',
            'is_organization': True
        }
)
hdx_org.add_url_rule(u'/request_new', view_func=request_new, methods=[u'GET', u'POST'])
hdx_org.add_url_rule(u'/<id>', view_func=read)
hdx_org.add_url_rule(u'/stats/<id>', view_func=stats)
hdx_org.add_url_rule(u'/restore/<id>', view_func=restore, methods=[u'POST'])
hdx_org.add_url_rule(u'/activity/<id>', view_func=activity)
hdx_org.add_url_rule(u'/activity/<id>/<int:offset>', view_func=activity_offset, defaults={'offset': 0})
hdx_org.add_url_rule(u'/<id>/download_stats', view_func=download_organization_stats)

1	import logging	1✔
2
3	from flask import Blueprint	1✔
4	from six.moves.urllib.parse import urlencode	1✔
5	from ckan.types import Context	1✔
6	import ckan.lib.plugins as lib_plugins	1✔
7	import ckan.model as model	1✔
8	import ckan.plugins.toolkit as tk	1✔
9	import ckanext.hdx_org_group.helpers.org_meta_dao as org_meta_dao	1✔
10	import ckanext.hdx_org_group.helpers.organization_helper as helper	1✔
11	import ckanext.hdx_org_group.helpers.static_lists as static_lists	1✔
12	import ckanext.hdx_theme.helpers.helpers as hdx_helpers	1✔
13	from ckan.views.group import CreateGroupView, EditGroupView, _get_group_template	1✔
14	from ckanext.hdx_org_group.controller_logic.organization_read_logic import OrgReadLogic	1✔
15	from ckanext.hdx_org_group.controller_logic.organization_stats_logic import (	1✔
16	OrganizationStatsLogic,
17	)
18	from ckanext.hdx_org_group.views.light_organization import _index	1✔
19	from ckanext.hdx_theme.util.light_redirect import check_redirect_needed	1✔
20	from ckanext.hdx_theme.util.mail import NoRecipientException	1✔
21
22	g = tk.g	1✔
23	config = tk.config	1✔
24	request = tk.request	1✔
25	render = tk.render	1✔
26	redirect = tk.redirect_to	1✔
27	url_for = tk.url_for	1✔
28	get_action = tk.get_action	1✔
29	check_access = tk.check_access	1✔
30	NotFound = tk.ObjectNotFound	1✔
31	NotAuthorized = tk.NotAuthorized	1✔
32	ValidationError = tk.ValidationError	1✔
33	abort = tk.abort	1✔
34	_ = tk._	1✔
35	h = tk.h	1✔
36
37	log = logging.getLogger(__name__)	1✔
38
39	hdx_org = Blueprint(u'hdx_org', __name__, url_prefix=u'/organization')	1✔
40
41
42	@check_redirect_needed	1✔
43	def index():	1✔
44	return _index('organization/index.html', False, True)	1✔
45
46
47	@check_redirect_needed	1✔
48	def read(id):	1✔
49	context: Context = {	1✔
50	'model': model,
51	'session': model.Session,
52	'for_view': True,
53	'with_private': False
54	}
55	try:	1✔
56	check_access('site_read', context)	1✔
57	except NotAuthorized:	×
58	abort(403, _('Not authorized to see this page'))	×
59
60	try:	1✔
61	read_logic = OrgReadLogic(id, g.user, g.userobj)	1✔
62	read_logic.read()	1✔
63	if read_logic.redirect_result:	1✔
64	return read_logic.redirect_result	×
65
66	if read_logic.org_meta.is_custom:	1✔
67	template_data = _generate_template_data_for_custom_org(read_logic)	×
68	result = render('organization/custom/custom_org.html', template_data)	×
69	return result	×
70	else:
71	org_dict = read_logic.org_meta.org_dict	1✔
72	org_dict.update({	1✔
73	'search_template_data': read_logic.search_template_data,
74	'datasets_num': read_logic.search_template_data.get('facets').get('extras_archived').get('fals'),
75	'archived_package_count': read_logic.search_template_data.get('facets').get('extras_archived').get('true'),
76	'allow_req_membership': read_logic.org_meta.allow_req_membership,
77	# 'group_message_info': read_logic.org_meta.group_message_info,
78	})
79
80	template_data = {	1✔
81	'org_dict': org_dict,
82	}
83	template_file = _get_group_template('read_template', 'organization')	1✔
84	return render(template_file, template_data)	1✔
NEW 85	except NotFound:	×
86	abort(404, _('Page not found'))	×
NEW 87	except NotAuthorized:	×
88	abort(403, _('Not authorized to see this page'))	×
89
90
91	def _generate_template_data_for_custom_org(org_read_logic):	1✔
92	"""
93	:param org_read_logic:
94	:type org_read_logic: OrgReadLogic
95	:returns: the template data dict
96	:rtype: dict
97	"""
98	org_meta = org_read_logic.org_meta	1✔
99	org_dict = org_meta.org_dict	1✔
100	org_id = org_dict['id']	1✔
101
102	# org_dict['group_message_info'] = org_meta.group_message_info
103	template_data = {	1✔
104	'data': {
105	'org_info': {
106	'id': org_id,
107	'display_name': org_dict.get('display_name', ''),
108	'description': org_dict.get('description'),
109	'name': org_dict['name'],
110	'link': org_dict.get('extras', {}).get('org_url'),
111	# 'topline_resource': org_meta.customization.get('topline_resource'),
112	'modified_at': org_dict.get('modified_at', ''),
113	'image_sq': org_meta.customization.get('image_sq'),
114	'image_rect': org_meta.customization.get('image_rect'),
115	# 'visualization_config': result.get('visualization_config', ''),
116	},
117	'search_template_data': org_read_logic.search_template_data,
118	#'custom_css_path': org_read_logic.org_meta.custom_css_path,
119	# 'member_count': hdx_helpers.get_group_members(org_id),
120	'follower_count': org_read_logic.follower_count,
121	'top_line_items': org_read_logic.top_line_items,
122	# 'search_results': {
123	# 'facets': facets,
124	# 'activities': activities,
125	# 'query_placeholder': query_placeholder
126	# },
127	# 'links': {
128	# 'edit': org_read_logic.links.edit,
129	# 'members': org_read_logic.links.members,
130	# 'request_membership': org_read_logic.links.request_membership,
131	# 'add_data': org_read_logic.links.add_data
132	# },
133	'request_params': request.params,
134	'permissions': {
135	'edit': org_read_logic.allow_edit,
136	'add_dataset': org_read_logic.allow_add_dataset,
137	'view_members': org_read_logic.allow_basic_user_info,
138	'request_membership': org_read_logic.allow_req_membership
139	},
140	'show_admin_menu': org_read_logic.allow_add_dataset or org_read_logic.allow_edit,
141	'show_visualization': 'Choose Visualization Type' != org_read_logic.viz_config.get('type'),
142	'visualization': {
143	'config': org_read_logic.viz_config,
144	'config_type': org_read_logic.viz_config.get('type'),
145	'config_url': urlencode(org_read_logic.viz_config, True),
146	# 'embed_url': org_read_logic.links.embed_url,
147
148	},
149
150	# This is here for compatibility with the custom_org_header.html template, which is still
151	# used from pylon controllers
152	'org_meta': {
153	'id': org_dict['name'],
154	'custom_rect_logo_url': org_meta.custom_rect_logo_url,
155	'custom_sq_logo_url': org_meta.custom_sq_logo_url,
156	'followers_num': org_meta.followers_num,
157	'members_num': org_meta.members_num,
158	'allow_req_membership': org_meta.allow_req_membership,
159	'allow_basic_user_info': org_meta.allow_basic_user_info,
160	'allow_add_dataset': org_meta.allow_add_dataset,
161	'allow_edit': org_meta.allow_edit,
162	'org_dict': org_dict,
163	},
164
165	},
166	'errors': org_read_logic.errors,
167	'error_summary': org_read_logic.error_summary,
168
169	}
170	if template_data['data']['show_visualization']:	1✔
171	template_data['data']['show_visualization'] = \	1✔
172	hdx_helpers.check_all_str_fields_not_empty(template_data['data']['visualization'],
173	'Visualization config field "{}" is empty',
174	skipped_keys=['config'],
175	errors=template_data['errors'])
176	return template_data	1✔
177
178
179	def request_new():	1✔
180	context: Context = {'model': model, 'session': model.Session, 'user': g.user}	1✔
181	try:	1✔
182	check_access('hdx_send_new_org_request', context)	1✔
183	except NotAuthorized:	1✔
184	abort(403, _('Unauthorized to send a new org request'))	1✔
185
186	errors = {}	×
187	error_summary = {}	×
188	data = {'from': request.form.get('from', '')}	×
189
190	sent_successfully = False	×
191	if 'save' in request.form and request.method == 'POST':	×
192	try:	×
193	data = _process_new_org_request()	×
194	_validate_new_org_request_field(data)	×
195
196	get_action('hdx_send_new_org_request')(context, _transform_dict_for_mailing(data))	×
197
198	data.clear()	×
199	h.flash_success(_('Request sent successfully'))	×
200	sent_successfully = True	×
201	except NoRecipientException as e:	×
202	h.flash_error(_(str(e)))	×
203	except ValidationError as e:	×
204	errors = e.error_dict	×
205	error_summary = e.error_summary	×
206	except Exception as e:	×
207	log.error(str(e))	×
208	h.flash_error(_('Request can not be sent. Contact an administrator'))	×
209	if sent_successfully:	×
210	return h.redirect_to('dashboard.organizations')	×
211
212	hdx_org_type_list = [{'value': '-1', 'text': _('-- Please select --')}] + \	×
213	[{'value': t[1], 'text': _(t[0])} for t in static_lists.ORGANIZATION_TYPE_LIST]
214	template_data = {	×
215	'data': {
216	'action': 'new',
217	'hdx_org_type_list': hdx_org_type_list
218	},
219	'form_data': data,
220	'errors': errors,
221	'error_summary': error_summary,
222
223	}
224	g.form = render('organization/request_organization_form.html', template_data)	×
225	return render('organization/request_new.html')	×
226
227
228	def _process_new_org_request():	1✔
229	hdx_org_type = None	×
230	hdx_org_type_code = request.form.get('hdx_org_type', '')	×
231
232	if hdx_org_type_code:	×
233	hdx_org_type = next(	×
234	(_type[0] for _type in static_lists.ORGANIZATION_TYPE_LIST if _type[1] == hdx_org_type_code), '-1')
235
236	data = {	×
237	'name': request.form.get('name', ''),
238	'title': request.form.get('title', ''),
239	'org_url': request.form.get('org_url', ''),
240	'description': request.form.get('description', ''),
241	'your_email': request.form.get('your_email', ''),
242	'your_name': request.form.get('your_name', ''),
243	'org_acronym': request.form.get('org_acronym', ''),
244	'hdx_org_type': hdx_org_type,
245	'hdx_org_type_code': hdx_org_type_code, # This is needed for the form when validation fails
246	}
247	return data	×
248
249
250	def _validate_new_org_request_field(data):	1✔
251	errors = {}	×
252	for field in ['title', 'description', 'your_email', 'your_name', 'hdx_org_type']:	×
253	if data[field].strip() in ['', '-1']:	×
254	errors[field] = [_('should not be empty')]	×
255
256	if len(errors) > 0:	×
257	raise ValidationError(errors)	×
258
259
260	def _transform_dict_for_mailing(data_dict):	1✔
261	data_dict_for_mailing = {	×
262	'name': data_dict.get('title'),
263	'acronym': data_dict.get('org_acronym'),
264	'description': data_dict.get('description'),
265	'org_type': data_dict.get('hdx_org_type', ''),
266	'org_url': data_dict.get('org_url', ''),
267
268	'work_email': data_dict.get('your_email', ''),
269	'your_name': data_dict.get('your_name', ''),
270	}
271	return data_dict_for_mailing	×
272
273
274	def new_org_template_variables(data_dict):	1✔
275	data_dict['hdx_org_type_list'] = [{'value': '-1', 'text': _('-- Please select --')}] + \	1✔
276	[{'value': t[1], 'text': _(t[0])} for t in static_lists.ORGANIZATION_TYPE_LIST]
277
278
279	def stats(id):	1✔
280	stats_logic = OrganizationStatsLogic(id, g.user, g.userobj)	1✔
281	org_dict = stats_logic.org_meta_dao.org_dict	1✔
282	org_dict.update({	1✔
283	'allow_req_membership': stats_logic.org_meta_dao.allow_req_membership,
284	# 'group_message_info': stats_logic.org_meta_dao.group_message_info,
285	})
286	template_data = {	1✔
287	'data': stats_logic.fetch_stats(),
288	'org_meta': stats_logic.org_meta_dao,
289	'org_dict': org_dict,
290	}
291
292	if stats_logic.is_custom():	1✔
293	return render('organization/custom_stats.html', template_data)	×
294	else:
295	return render('organization/stats.html', template_data)	1✔
296
297
298	def restore(id):	1✔
NEW 299	context: Context = {	×
300	'model': model, 'session': model.Session,
301	'user': g.user,
302	'for_edit': True,
303	}
304
305	try:	×
306	check_access('organization_patch', context, {'id': id})	×
NEW 307	except NotAuthorized:	×
308	return abort(403, _('Unauthorized to restore this organization'))	×
309
310	try:	×
311	get_action('organization_patch')(context, {	×
312	'id': id,
313	'state': 'active'
314	})
315	return redirect('organization.read', id=id)	×
316	except NotAuthorized:	×
317	return abort(403, _('Unauthorized to read group %s') % id)	×
318	except NotFound:	×
319	return abort(404, _('Group not found'))	×
320	except ValidationError as e:	×
321	errors = e.error_dict	×
322	error_summary = e.error_summary	×
323	core_view = EditGroupView()	×
324	return core_view.get(id, 'organization', True, errors=errors, error_summary=error_summary)	×
325
326
327	def activity(id):	1✔
328	return activity_offset(id)	×
329
330
331	def activity_offset(id, offset=0):	1✔
332	"""
333	Modified core functionality to use the new OrgMetaDao class
334	for fetching information needed on all org-related pages.
335
336	Render this group's public activity stream page.
337
338	:param id:
339	:type id: str
340	:param offset:
341	:type offset: int
342	:return:
343	"""
344	org_meta = org_meta_dao.OrgMetaDao(id, g.user, g.userobj)	×
345	org_meta.fetch_all()	×
346	org_dict = org_meta.org_dict	×
347	# org_dict['group_message_info'] = org_meta.group_message_info
348
349	helper.org_add_last_updated_field([org_dict])	×
350
351	# Add the group's activity stream (already rendered to HTML) to the
352	# template context for the group/read.html template to retrieve later.
NEW 353	context: Context = {'model': model, 'session': model.Session,	×
354	'user': g.user, 'for_view': True}
355	group_activity_stream = get_action('organization_activity_list')(	×
356	context, {'id': org_dict['id'], 'offset': offset})
357
UNCOV 358	extra_vars = {	×
359	'org_dict': org_dict,
360	'org_meta': org_meta,
361	'group_activity_stream': group_activity_stream,
362
363	}
364	if org_meta.is_custom:	×
365	template = 'organization/custom_activity_stream.html'	×
366	else:
367	template = lib_plugins.lookup_group_plugin('organization').activity_template()	×
368	return render(template, extra_vars)	×
369
370	def download_organization_stats(id):	1✔
371	"""
372	Handles downloading .xlsx organization stats
373
374	:returns: xlsx
375	"""
376
377	context: Context = {	1✔
378	'model': model,
379	'session': model.Session,
380	'user': g.user or g.author,
381	'auth_user_obj': g.userobj
382	}
383
384	try:	1✔
385	check_access('organization_update', context, {'id': id})	1✔
386	except NotAuthorized:	1✔
387	return abort(403, _('Unauthorized to restore this organization'))	1✔
388
389	# check if organization exists
390	try:	1✔
391	org_dict = get_action('organization_show')(context, {'id': id})	1✔
392	output = helper.hdx_generate_organization_stats(org_dict)	1✔
393	return output	1✔
394
NEW 395	except NotFound:	×
NEW 396	return abort(404, _('Organization not found'))	×
NEW 397	except NotAuthorized:	×
NEW 398	return abort(404, _('Organization not found'))	×
NEW 399	except Exception as e:	×
NEW 400	return abort(404, _('Something went wrong, please contact us'))	×
401
402
403	hdx_org.add_url_rule(u'/', view_func=index, strict_slashes=False)	1✔
404	hdx_org.add_url_rule(	1✔
405	u'/new',
406	methods=[u'GET', u'POST'],
407	view_func=CreateGroupView.as_view(str(u'new')),
408	defaults={
409	'group_type': 'organization',
410	'is_organization': True
411	}
412	)
413	hdx_org.add_url_rule(u'/request_new', view_func=request_new, methods=[u'GET', u'POST'])	1✔
414	hdx_org.add_url_rule(u'/<id>', view_func=read)	1✔
415	hdx_org.add_url_rule(u'/stats/<id>', view_func=stats)	1✔
416	hdx_org.add_url_rule(u'/restore/<id>', view_func=restore, methods=[u'POST'])	1✔
417	hdx_org.add_url_rule(u'/activity/<id>', view_func=activity)	1✔
418	hdx_org.add_url_rule(u'/activity/<id>/<int:offset>', view_func=activity_offset, defaults={'offset': 0})	1✔
419	hdx_org.add_url_rule(u'/<id>/download_stats', view_func=download_organization_stats)	1✔

OCHA-DAP / hdx-ckan / #5847

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