#6854

Committed 13 Nov 2025 01:05PM UTC coverage: 78.305% (-0.006%) from 78.311%

Build # #6854

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Merge c1fe41c14 into 9ceff6136

Run Details

13517 of 17262 relevant lines covered (78.3%)

0.78 hits per line

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

75.65

/ckanext-hdx_users/ckanext/hdx_users/helpers/helpers.py

import json
import random
import string
import requests

import ckan.authz as authz
import ckan.lib.navl.dictization_functions as df
import ckan.model.user as user_model
import ckan.plugins.toolkit as tk
from ckan.logic.validators import name_validator, name_match, PACKAGE_NAME_MAX_LENGTH
from ckanext.hdx_users.helpers.notification_service import get_notification_service

get_action = tk.get_action
check_access = tk.check_access
NotAuthorized = tk.NotAuthorized
NotFound = tk.ObjectNotFound
h = tk.h
g = tk.g
config = tk.config
_ = tk._
CaptchaNotValid = _('Captcha is not valid')
ValidationError = tk.ValidationError
Invalid = df.Invalid




def find_first_global_settings_url():
    context = {'user': g.user}
    url = None
    if authz.is_sysadmin(g.user):
        url = h.url_for('admin.index')

    if not url:
        try:
            check_access('hdx_carousel_update', context, {})
            url = h.url_for('hdx_carousel.show')
        except NotAuthorized as e:
            pass

    if not url:
        try:
            check_access('hdx_request_data_admin_list', context, {})
            url = h.url_for('requestdata_ckanadmin.requests_data')
        except NotAuthorized as e:
            pass

    if not url:
        try:
            check_access('admin_page_list', context, {})
            url = h.url_for('hdx_custom_pages.index')
        except NotAuthorized as e:
            pass

    if not url:
        try:
            check_access('hdx_quick_links_update', context, {})
            url = h.url_for('hdx_quick_links.show')
        except NotAuthorized as e:
            pass

    return url


def hdx_get_user_notifications():
    # return get_notification_service().get_notifications()
    try:
        if not g.hdx_user_notifications:
            # this part is for pylons, flask gives an exception (see below)
            g.hdx_user_notifications = get_notification_service().get_notifications()
    except AttributeError as e:
        # if we are in flask we get an AttributeError before setting the property in g the first time
        g.hdx_user_notifications = get_notification_service().get_notifications()

    return g.hdx_user_notifications


def find_user_id(username_or_id):
    user = user_model.User.get(username_or_id)
    if not user:
        raise NotFound()
    user_id = user.id
    return user_id


def is_valid_captcha(captcha_response):
    is_captcha_enabled = config.get('hdx.captcha')
    if is_captcha_enabled:
        # captcha_response = request.params.get('g-recaptcha-response')
        if not validate_captcha(response=captcha_response):
            raise ValidationError(CaptchaNotValid, error_summary=CaptchaNotValid)
        return True
    else:
        return None


def validate_captcha(response):
    url = config.get('hdx.captcha.url')
    secret = config.get('ckan.recaptcha.privatekey')
    params = {'secret': secret, "response": response}
    r = requests.get(url, params=params, verify=True)
    res = json.loads(r.content)
    return 'success' in res and res['success'] == True


def name_validator_with_changed_msg(val, context):
    """This is just a wrapper function around the validator.name_validator function.
        The wrapper function just changes the message in case the name_match doesn't match.
        The only purpose for still calling that function here is to keep the link visible and
        in case of a ckan upgrade to still be able to raise any new Invalid exceptions

    """
    try:
        return name_validator(val, context)
    except Invalid as invalid:
        if val in ['new', 'edit', 'search']:
            raise Invalid(_('That name cannot be used'))

        if len(val) < 2:
            raise Invalid(_('Name must be at least %s characters long') % 2)
        if len(val) > PACKAGE_NAME_MAX_LENGTH:
            raise Invalid(_('Name must be a maximum of %i characters long') % \
                          PACKAGE_NAME_MAX_LENGTH)
        if not name_match.match(val):
            raise Invalid(_('Username should be lowercase letters and/or numbers and/or these symbols: -_'))

        raise invalid


# Generate a password
# generate_password(12)
def generate_password(length=10):
    if length < 10:
        raise ValueError('Password must be at least 10 characters long.')

    # Ensure the password contains at least one character from three of the four groups
    uppercase = random.choice(string.ascii_uppercase)
    lowercase = random.choice(string.ascii_lowercase)
    digit = random.choice(string.digits)
    special = random.choice(string.punctuation)

    # Choose at least three of the four categories
    mandatory_chars = [uppercase, lowercase, digit, special]
    random.shuffle(mandatory_chars)
    password_list = mandatory_chars[:3]

    # Fill the remaining characters randomly
    remaining_length = length - len(password_list)
    all_chars = string.ascii_letters + string.digits + string.punctuation
    password_list.extend(random.choices(all_chars, k=remaining_length))

    # Shuffle the final password to avoid predictable patterns
    random.shuffle(password_list)
    return ''.join(password_list)

# Generate a sample username
# generate_username(5, 20)
def generate_username(min_length=2, max_length=100):
    if min_length < 2 or max_length > 100:
        raise ValueError('Username length must be between 2 and 100 characters.')

    # Define the allowed characters
    allowed_chars = string.ascii_lowercase + string.digits + '-_'

    # Generate a random username within the specified length range
    length = random.randint(min_length, max_length)
    username = ''.join(random.choices(allowed_chars, k=length))

    # Ensure it doesn't start with a special character
    while username[0] in '-_':
        username = ''.join(random.choices(allowed_chars, k=length))

    return username


1	import json	1✔
2	import random	1✔
3	import string	1✔
4	import requests	1✔
5
6	import ckan.authz as authz	1✔
7	import ckan.lib.navl.dictization_functions as df	1✔
8	import ckan.model.user as user_model	1✔
9	import ckan.plugins.toolkit as tk	1✔
10	from ckan.logic.validators import name_validator, name_match, PACKAGE_NAME_MAX_LENGTH	1✔
11	from ckanext.hdx_users.helpers.notification_service import get_notification_service	1✔
12
13	get_action = tk.get_action	1✔
14	check_access = tk.check_access	1✔
15	NotAuthorized = tk.NotAuthorized	1✔
16	NotFound = tk.ObjectNotFound	1✔
17	h = tk.h	1✔
18	g = tk.g	1✔
19	config = tk.config	1✔
20	_ = tk._	1✔
21	CaptchaNotValid = _('Captcha is not valid')	1✔
22	ValidationError = tk.ValidationError	1✔
23	Invalid = df.Invalid	1✔
24
25
26
27
28	def find_first_global_settings_url():	1✔
29	context = {'user': g.user}	1✔
30	url = None	1✔
31	if authz.is_sysadmin(g.user):	1✔
32	url = h.url_for('admin.index')	1✔
33
34	if not url:	1✔
35	try:	1✔
36	check_access('hdx_carousel_update', context, {})	1✔
37	url = h.url_for('hdx_carousel.show')	×
38	except NotAuthorized as e:	1✔
39	pass	1✔
40
41	if not url:	1✔
42	try:	1✔
43	check_access('hdx_request_data_admin_list', context, {})	1✔
44	url = h.url_for('requestdata_ckanadmin.requests_data')	×
45	except NotAuthorized as e:	1✔
46	pass	1✔
47
48	if not url:	1✔
49	try:	1✔
50	check_access('admin_page_list', context, {})	1✔
51	url = h.url_for('hdx_custom_pages.index')	×
52	except NotAuthorized as e:	1✔
53	pass	1✔
54
55	if not url:	1✔
56	try:	1✔
57	check_access('hdx_quick_links_update', context, {})	1✔
58	url = h.url_for('hdx_quick_links.show')	×
59	except NotAuthorized as e:	1✔
60	pass	1✔
61
62	return url	1✔
63
64
65	def hdx_get_user_notifications():	1✔
66	# return get_notification_service().get_notifications()
67	try:	1✔
68	if not g.hdx_user_notifications:	1✔
69	# this part is for pylons, flask gives an exception (see below)
70	g.hdx_user_notifications = get_notification_service().get_notifications()	1✔
71	except AttributeError as e:	1✔
72	# if we are in flask we get an AttributeError before setting the property in g the first time
73	g.hdx_user_notifications = get_notification_service().get_notifications()	1✔
74
75	return g.hdx_user_notifications	1✔
76
77
78	def find_user_id(username_or_id):	1✔
79	user = user_model.User.get(username_or_id)	1✔
80	if not user:	1✔
81	raise NotFound()	1✔
82	user_id = user.id	1✔
83	return user_id	1✔
84
85
86	def is_valid_captcha(captcha_response):	1✔
87	is_captcha_enabled = config.get('hdx.captcha')	1✔
88	if is_captcha_enabled:	1✔
89	# captcha_response = request.params.get('g-recaptcha-response')
90	if not validate_captcha(response=captcha_response):	×
91	raise ValidationError(CaptchaNotValid, error_summary=CaptchaNotValid)	×
92	return True	×
93	else:
94	return None	1✔
95
96
97	def validate_captcha(response):	1✔
98	url = config.get('hdx.captcha.url')	×
99	secret = config.get('ckan.recaptcha.privatekey')	×
100	params = {'secret': secret, "response": response}	×
101	r = requests.get(url, params=params, verify=True)	×
102	res = json.loads(r.content)	×
103	return 'success' in res and res['success'] == True	×
104
105
106	def name_validator_with_changed_msg(val, context):	1✔
107	"""This is just a wrapper function around the validator.name_validator function.
108	The wrapper function just changes the message in case the name_match doesn't match.
109	The only purpose for still calling that function here is to keep the link visible and
110	in case of a ckan upgrade to still be able to raise any new Invalid exceptions
111
112	"""
113	try:	×
114	return name_validator(val, context)	×
115	except Invalid as invalid:	×
116	if val in ['new', 'edit', 'search']:	×
117	raise Invalid(_('That name cannot be used'))	×
118
119	if len(val) < 2:	×
120	raise Invalid(_('Name must be at least %s characters long') % 2)	×
121	if len(val) > PACKAGE_NAME_MAX_LENGTH:	×
122	raise Invalid(_('Name must be a maximum of %i characters long') % \	×
123	PACKAGE_NAME_MAX_LENGTH)
124	if not name_match.match(val):	×
125	raise Invalid(_('Username should be lowercase letters and/or numbers and/or these symbols: -_'))	×
126
127	raise invalid	×
128
129
130	# Generate a password
131	# generate_password(12)
132	def generate_password(length=10):	1✔
133	if length < 10:	1✔
134	raise ValueError('Password must be at least 10 characters long.')	×
135
136	# Ensure the password contains at least one character from three of the four groups
137	uppercase = random.choice(string.ascii_uppercase)	1✔
138	lowercase = random.choice(string.ascii_lowercase)	1✔
139	digit = random.choice(string.digits)	1✔
140	special = random.choice(string.punctuation)	1✔
141
142	# Choose at least three of the four categories
143	mandatory_chars = [uppercase, lowercase, digit, special]	1✔
144	random.shuffle(mandatory_chars)	1✔
145	password_list = mandatory_chars[:3]	1✔
146
147	# Fill the remaining characters randomly
148	remaining_length = length - len(password_list)	1✔
149	all_chars = string.ascii_letters + string.digits + string.punctuation	1✔
150	password_list.extend(random.choices(all_chars, k=remaining_length))	1✔
151
152	# Shuffle the final password to avoid predictable patterns
153	random.shuffle(password_list)	1✔
154	return ''.join(password_list)	1✔
155
156	# Generate a sample username
157	# generate_username(5, 20)
158	def generate_username(min_length=2, max_length=100):	1✔
159	if min_length < 2 or max_length > 100:	1✔
160	raise ValueError('Username length must be between 2 and 100 characters.')	×
161
162	# Define the allowed characters
163	allowed_chars = string.ascii_lowercase + string.digits + '-_'	1✔
164
165	# Generate a random username within the specified length range
166	length = random.randint(min_length, max_length)	1✔
167	username = ''.join(random.choices(allowed_chars, k=length))	1✔
168
169	# Ensure it doesn't start with a special character
170	while username[0] in '-_':	1✔
171	username = ''.join(random.choices(allowed_chars, k=length))	×
172
173	return username	1✔
174

OCHA-DAP / hdx-ckan / #6854

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