#6504

Committed 01 Aug 2025 08:53AM UTC coverage: 74.843%. Remained the same

Build # #6504

Build Type

push

coveralls-python

Committed by

ccataalin

Commit Message

update exception messages to use `display_name` for object type

Run Details

0 of 2 new or added lines in 2 files covered. (0.0%)

13093 of 17494 relevant lines covered (74.84%)

0.75 hits per line

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

82.76

/ckanext-hdx_users/ckanext/hdx_users/controller_logic/notification_platform_logic.py

import logging

from typing import Any, Optional, Dict

import ckan.plugins.toolkit as tk
import ckan.model as model
from ckan.types import AlchemySession

from ckanext.hdx_theme.helpers.helpers import hdx_supports_notifications

from ckanext.hdx_users.general_token_model import generate_new_token_obj, validate_token, ObjectType, TokenType, \
    HDXGeneralToken, get_by_type_and_user_id_and_object, get_by_token_with_checks

log = logging.getLogger(__name__)

h = tk.h
config = tk.config
get_action = tk.get_action


def get_or_generate_email_validation_token(email: str, object_type: ObjectType, object_id: str,
                                           object_dict: Optional[dict[str, Any]] = None,
                                           extras: Optional[Dict] = None) -> HDXGeneralToken:
    object_supports_notifications = hdx_supports_notifications(object_type, object_id, object_dict)
    if object_supports_notifications:
        email_validation_token = get_by_type_and_user_id_and_object(TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, email,
                                                                    object_type, object_id)
        if email_validation_token:
            return email_validation_token
        else:
            return generate_new_token_obj(model.Session, TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, email,
                                          object_type=object_type, object_id=object_id, extras=extras)
    else:
        log.warning(
            f'Tried to generate token for {object_type.value} {object_id} but it does not support notifications')
        raise Exception(f'{object_type.display_name} {object_id} does not support notifications')


def get_or_generate_unsubscribe_token(session: AlchemySession, user_id: str, object_type: ObjectType, object_id: str,
                                      extras: Optional[Dict] = None, commit_tx: bool = True) -> HDXGeneralToken:
    existing_unsubscribe_token = get_by_type_and_user_id_and_object(TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, user_id,
                                                                    object_type, object_id)
    if existing_unsubscribe_token:
        return existing_unsubscribe_token
    else:
        return generate_new_token_obj(session, TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, user_id,
                                      object_type=object_type, object_id=object_id, extras=extras, commit_tx=commit_tx)


def verify_email_validation_token(token: str) -> HDXGeneralToken:
    return validate_token(model.Session, token, TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, True)


def get_unsubscribe_token(token: str) -> HDXGeneralToken:
    return get_by_token_with_checks(token, TokenType.UNSUBSCRIBE_FOR_NOTIFICATION)

1	import logging	1✔
2
3	from typing import Any, Optional, Dict	1✔
4
5	import ckan.plugins.toolkit as tk	1✔
6	import ckan.model as model	1✔
7	from ckan.types import AlchemySession	1✔
8
9	from ckanext.hdx_theme.helpers.helpers import hdx_supports_notifications	1✔
10
11	from ckanext.hdx_users.general_token_model import generate_new_token_obj, validate_token, ObjectType, TokenType, \	1✔
12	HDXGeneralToken, get_by_type_and_user_id_and_object, get_by_token_with_checks
13
14	log = logging.getLogger(__name__)	1✔
15
16	h = tk.h	1✔
17	config = tk.config	1✔
18	get_action = tk.get_action	1✔
19
20
21	def get_or_generate_email_validation_token(email: str, object_type: ObjectType, object_id: str,	1✔
22	object_dict: Optional[dict[str, Any]] = None,
23	extras: Optional[Dict] = None) -> HDXGeneralToken:
24	object_supports_notifications = hdx_supports_notifications(object_type, object_id, object_dict)	1✔
25	if object_supports_notifications:	1✔
26	email_validation_token = get_by_type_and_user_id_and_object(TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, email,	1✔
27	object_type, object_id)
28	if email_validation_token:	1✔
29	return email_validation_token	×
30	else:
31	return generate_new_token_obj(model.Session, TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, email,	1✔
32	object_type=object_type, object_id=object_id, extras=extras)
33	else:
34	log.warning(	×
35	f'Tried to generate token for {object_type.value} {object_id} but it does not support notifications')
NEW 36	raise Exception(f'{object_type.display_name} {object_id} does not support notifications')	×
37
38
39	def get_or_generate_unsubscribe_token(session: AlchemySession, user_id: str, object_type: ObjectType, object_id: str,	1✔
40	extras: Optional[Dict] = None, commit_tx: bool = True) -> HDXGeneralToken:
41	existing_unsubscribe_token = get_by_type_and_user_id_and_object(TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, user_id,	1✔
42	object_type, object_id)
43	if existing_unsubscribe_token:	1✔
44	return existing_unsubscribe_token	×
45	else:
46	return generate_new_token_obj(session, TokenType.UNSUBSCRIBE_FOR_NOTIFICATION, user_id,	1✔
47	object_type=object_type, object_id=object_id, extras=extras, commit_tx=commit_tx)
48
49
50	def verify_email_validation_token(token: str) -> HDXGeneralToken:	1✔
51	return validate_token(model.Session, token, TokenType.EMAIL_VALIDATION_FOR_NOTIFICATION, True)	1✔
52
53
54	def get_unsubscribe_token(token: str) -> HDXGeneralToken:	1✔
55	return get_by_token_with_checks(token, TokenType.UNSUBSCRIBE_FOR_NOTIFICATION)	×

OCHA-DAP / hdx-ckan / #6504

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