#6168

Committed 12 Mar 2025 07:32AM UTC coverage: 74.93% (-0.06%) from 74.987%

Build # #6168

Build Type

push

coveralls-python

Committed by

danmihaila

Commit Message

HDX-10466 fix fresh flag test

Run Details

12598 of 16813 relevant lines covered (74.93%)

0.75 hits per line

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

83.95

/ckanext-hdx_package/ckanext/hdx_package/helpers/freshness_calculator.py

import datetime
import logging
from collections import OrderedDict

import dateutil.parser
from six import text_type
import ckan.plugins.toolkit as tk
from ckanext.hdx_package.helpers.extras import get_extra_from_dataset

log = logging.getLogger(__name__)
h=tk.h
UPDATE_FREQ_LIVE = '0'
UPDATE_FREQ_AS_NEEDED = '-2'
UPDATE_FREQ_NEVER = '-1'

UPDATE_FREQ_INFO = OrderedDict(
    (
        ('1', {
            'title': 'Every day',
            'special': False
        }),
        ('7', {
            'title': 'Every week',
            'special': False
        }),
        ('14', {
            'title': 'Every two weeks',
            'special': False
        }),
        ('30', {
            'title': 'Every month',
            'special': False
        }),
        ('90', {
            'title': 'Every three months',
            'special': False
        }),
        ('180', {
            'title': 'Every six months',
            'special': False
        }),
        ('365', {
            'title': 'Every year',
            'special': False
        }),
        (UPDATE_FREQ_LIVE, {
            'title': 'Live',
            'special': True
        }),
        (UPDATE_FREQ_AS_NEEDED, {
            'title': 'As needed',
            'special': True
        }),
        (UPDATE_FREQ_NEVER, {
            'title': 'Never',
            'special': True
        }),
    )
)

FRESHNESS_PROPERTY = 'is_fresh'

UPDATE_STATUS_PROPERTY = 'update_status'
UPDATE_STATUS_URL_FILTER = 'ext_' + UPDATE_STATUS_PROPERTY

UPDATE_STATUS_FRESH = 'fresh'
UPDATE_STATUS_UNKNOWN = 'unknown'
UPDATE_STATUS_NEEDS_UPDATE = 'needs_update'

DELTA_END_DATASET_DATE_INFINITE_DAYS = 365*100

def get_calculator_instance(dataset_dict, type='for-data-completeness'):
    # if type == 'for-data-completeness':
    #     return DataCompletenessFreshnessCalculator(dataset_dict)
    # else:
    return FreshnessCalculator(dataset_dict)


class FreshnessCalculator(object):

    def __init__(self, dataset_dict):

        self.surely_not_fresh = True
        self.dataset_dict = dataset_dict
        update_freq = get_extra_from_dataset('data_update_frequency', dataset_dict)
        try:
            dataset_date = get_extra_from_dataset('dataset_date', dataset_dict)
            self.end_dataset_date, self.is_end_dataset_date_star = h.hdx_end_of_dataset_date(dataset_date)
            if self.end_dataset_date and update_freq:
                self.update_freq_in_days = int(update_freq)
                self.surely_not_fresh = False
        except Exception as e:
            log.error(text_type(e))

    def is_fresh(self, now=datetime.datetime.utcnow()):
        """
        Using utcnow because this is used by core ckan, see ckan.model.package
        :return: True if fresh, otherwise False
        :rtype: bool
        """
        update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)
        if update_freq == UPDATE_FREQ_LIVE:
            return True
        if update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:
            return False

        start_of_expiration = self.compute_range_beginnings()
        if start_of_expiration:
            now = datetime.datetime.utcnow() # using utcnow bc this is used by core ckan, see ckan.model.package
            fresh = now < start_of_expiration
            return fresh
        else:
            return False

    def populate_with_freshness(self):
        is_fresh = self.is_fresh()
        self.dataset_dict[FRESHNESS_PROPERTY] = is_fresh

        if is_fresh:
            self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_FRESH
        elif self.dataset_dict.get('due_date'):
            self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_NEEDS_UPDATE
        else:
            self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_UNKNOWN

    def populate_with_date_ranges(self):
        update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)
        if update_freq == UPDATE_FREQ_LIVE or update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:
            self.dataset_dict['due_date'] = None
        else:
            start_of_due_range= self.compute_range_beginnings()
            if start_of_due_range:
                self.dataset_dict['due_date'] = start_of_due_range.isoformat()

    def compute_range_beginnings(self):
        if not self.surely_not_fresh:
            if self.is_end_dataset_date_star or not self.update_freq_in_days or self.update_freq_in_days == -1:
                start_of_due_range = (
                    self.end_dataset_date + datetime.timedelta(days=DELTA_END_DATASET_DATE_INFINITE_DAYS)
                ).replace(microsecond=0)
            else:
                start_of_due_range = (self.end_dataset_date + datetime.timedelta(days=self.update_freq_in_days))\
                .replace(microsecond=0)
            return start_of_due_range #, start_of_overdue_range #, start_of_delinquent_range
        else:
            return None

    def read_due_overdue_dates(self):
        try:
            if 'due_date' in self.dataset_dict:
                update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)
                if update_freq == UPDATE_FREQ_LIVE or update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:
                    return None
                due_date_str = self.dataset_dict.get('due_date')
                due_date = dateutil.parser.parse(due_date_str[0:-1])
                # overdue_date_str = self.dataset_dict.get('overdue_date')
                # overdue_date = dateutil.parser.parse(overdue_date_str[0:-1])
                return due_date #, overdue_date
        except Exception as e:
            log.warn(str(e))
        return None

1	import datetime	1✔
2	import logging	1✔
3	from collections import OrderedDict	1✔
4
5	import dateutil.parser	1✔
6	from six import text_type	1✔
7	import ckan.plugins.toolkit as tk	1✔
8	from ckanext.hdx_package.helpers.extras import get_extra_from_dataset	1✔
9
10	log = logging.getLogger(__name__)	1✔
11	h=tk.h	1✔
12	UPDATE_FREQ_LIVE = '0'	1✔
13	UPDATE_FREQ_AS_NEEDED = '-2'	1✔
14	UPDATE_FREQ_NEVER = '-1'	1✔
15
16	UPDATE_FREQ_INFO = OrderedDict(	1✔
17	(
18	('1', {
19	'title': 'Every day',
20	'special': False
21	}),
22	('7', {
23	'title': 'Every week',
24	'special': False
25	}),
26	('14', {
27	'title': 'Every two weeks',
28	'special': False
29	}),
30	('30', {
31	'title': 'Every month',
32	'special': False
33	}),
34	('90', {
35	'title': 'Every three months',
36	'special': False
37	}),
38	('180', {
39	'title': 'Every six months',
40	'special': False
41	}),
42	('365', {
43	'title': 'Every year',
44	'special': False
45	}),
46	(UPDATE_FREQ_LIVE, {
47	'title': 'Live',
48	'special': True
49	}),
50	(UPDATE_FREQ_AS_NEEDED, {
51	'title': 'As needed',
52	'special': True
53	}),
54	(UPDATE_FREQ_NEVER, {
55	'title': 'Never',
56	'special': True
57	}),
58	)
59	)
60
61	FRESHNESS_PROPERTY = 'is_fresh'	1✔
62
63	UPDATE_STATUS_PROPERTY = 'update_status'	1✔
64	UPDATE_STATUS_URL_FILTER = 'ext_' + UPDATE_STATUS_PROPERTY	1✔
65
66	UPDATE_STATUS_FRESH = 'fresh'	1✔
67	UPDATE_STATUS_UNKNOWN = 'unknown'	1✔
68	UPDATE_STATUS_NEEDS_UPDATE = 'needs_update'	1✔
69
70	DELTA_END_DATASET_DATE_INFINITE_DAYS = 365*100	1✔
71
72	def get_calculator_instance(dataset_dict, type='for-data-completeness'):	1✔
73	# if type == 'for-data-completeness':
74	# return DataCompletenessFreshnessCalculator(dataset_dict)
75	# else:
76	return FreshnessCalculator(dataset_dict)	1✔
77
78
79	class FreshnessCalculator(object):	1✔
80
81	def __init__(self, dataset_dict):	1✔
82
83	self.surely_not_fresh = True	1✔
84	self.dataset_dict = dataset_dict	1✔
85	update_freq = get_extra_from_dataset('data_update_frequency', dataset_dict)	1✔
86	try:	1✔
87	dataset_date = get_extra_from_dataset('dataset_date', dataset_dict)	1✔
88	self.end_dataset_date, self.is_end_dataset_date_star = h.hdx_end_of_dataset_date(dataset_date)	1✔
89	if self.end_dataset_date and update_freq:	1✔
90	self.update_freq_in_days = int(update_freq)	1✔
91	self.surely_not_fresh = False	1✔
92	except Exception as e:	×
93	log.error(text_type(e))	×
94
95	def is_fresh(self, now=datetime.datetime.utcnow()):	1✔
96	"""
97	Using utcnow because this is used by core ckan, see ckan.model.package
98	:return: True if fresh, otherwise False
99	:rtype: bool
100	"""
101	update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)	1✔
102	if update_freq == UPDATE_FREQ_LIVE:	1✔
103	return True	1✔
104	if update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:	1✔
105	return False	1✔
106
107	start_of_expiration = self.compute_range_beginnings()	1✔
108	if start_of_expiration:	1✔
109	now = datetime.datetime.utcnow() # using utcnow bc this is used by core ckan, see ckan.model.package	1✔
110	fresh = now < start_of_expiration	1✔
111	return fresh	1✔
112	else:
113	return False	1✔
114
115	def populate_with_freshness(self):	1✔
116	is_fresh = self.is_fresh()	1✔
117	self.dataset_dict[FRESHNESS_PROPERTY] = is_fresh	1✔
118
119	if is_fresh:	1✔
120	self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_FRESH	1✔
121	elif self.dataset_dict.get('due_date'):	1✔
122	self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_NEEDS_UPDATE	1✔
123	else:
124	self.dataset_dict[UPDATE_STATUS_PROPERTY] = UPDATE_STATUS_UNKNOWN	1✔
125
126	def populate_with_date_ranges(self):	1✔
127	update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)	1✔
128	if update_freq == UPDATE_FREQ_LIVE or update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:	1✔
129	self.dataset_dict['due_date'] = None	1✔
130	else:
131	start_of_due_range= self.compute_range_beginnings()	1✔
132	if start_of_due_range:	1✔
133	self.dataset_dict['due_date'] = start_of_due_range.isoformat()	1✔
134
135	def compute_range_beginnings(self):	1✔
136	if not self.surely_not_fresh:	1✔
137	if self.is_end_dataset_date_star or not self.update_freq_in_days or self.update_freq_in_days == -1:	1✔
138	start_of_due_range = (	1✔
139	self.end_dataset_date + datetime.timedelta(days=DELTA_END_DATASET_DATE_INFINITE_DAYS)
140	).replace(microsecond=0)
141	else:
142	start_of_due_range = (self.end_dataset_date + datetime.timedelta(days=self.update_freq_in_days))\	1✔
143	.replace(microsecond=0)
144	return start_of_due_range #, start_of_overdue_range #, start_of_delinquent_range	1✔
145	else:
146	return None	1✔
147
148	def read_due_overdue_dates(self):	1✔
149	try:	×
150	if 'due_date' in self.dataset_dict:	×
151	update_freq = get_extra_from_dataset('data_update_frequency', self.dataset_dict)	×
152	if update_freq == UPDATE_FREQ_LIVE or update_freq == UPDATE_FREQ_NEVER or update_freq == UPDATE_FREQ_AS_NEEDED:	×
153	return None	×
154	due_date_str = self.dataset_dict.get('due_date')	×
155	due_date = dateutil.parser.parse(due_date_str[0:-1])	×
156	# overdue_date_str = self.dataset_dict.get('overdue_date')
157	# overdue_date = dateutil.parser.parse(overdue_date_str[0:-1])
158	return due_date #, overdue_date	×
159	except Exception as e:	×
160	log.warn(str(e))	×
161	return None	×

OCHA-DAP / hdx-ckan / #6168

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