9287122129

Committed 29 May 2024 01:34PM UTC coverage: 91.912% (+0.5%) from 91.42%

Build # 9287122129

Build Type

Pull #60

github

Committed by

duchenean

Commit Message

change type of default oo_port to str

Pull Request Pull Request #60: Plone 6 compatibility

Run Details

64 of 85 new or added lines in 15 files covered. (75.29%)

7 existing lines in 2 files now uncovered.

2182 of 2374 relevant lines covered (91.91%)

1.84 hits per line

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

88.39

/src/collective/documentgenerator/utils.py

# -*- coding: utf-8 -*-
from collective.documentgenerator import _
from collective.documentgenerator.config import HAS_PLONE_5
from imio.helpers.security import fplog
from plone import api
from plone.namedfile.file import NamedBlobFile
from Products.CMFCore.utils import getToolByName
from Products.CMFPlone.utils import safe_unicode
from zope import i18n
from zope.component import getMultiAdapter
from zope.component.hooks import getSite
from zope.component.hooks import setSite
from zope.interface import Interface
from zope.interface import Invalid
from zope.lifecycleevent import Attributes
from zope.lifecycleevent import modified

import hashlib
import logging
import os
import re
import six
import tempfile


if six.PY2:
    from appy.bin.odfclean import Cleaner
else:
    from appy.bin.oclean import Cleaner

logger = logging.getLogger('collective.documentgenerator')


def translate(msgid, domain='collective.documentgenerator'):
    portal = api.portal.get()
    translation = i18n.translate(
        msgid,
        domain=domain,
        context=portal.REQUEST
    )
    return translation


def compute_md5(data):
    md5 = hashlib.md5(data).hexdigest()
    return md5


def update_templates(templates, profile='', force=False):
    """
        function to manage templates update.
        # see http://trac.imio.be/trac/ticket/9383 for full implementation
        :param list templates: list of tuples containing ('plone-template-path', 'os-template-path')
        :param str profile: profile path stored on template (or various identification)
        :param bool force: force overrides of templates
    """
    # Don't use profile now !
    portal = api.portal.getSite()
    ret = []
    for (ppath, ospath) in templates:
        ppath = ppath.strip('/ ')
        obj = portal.unrestrictedTraverse(ppath, default=None)
        if not obj:
            logger.warn("The plone template '%s' was not found" % ppath)
            ret.append((ppath, ospath, 'plone path error'))
            continue
        if not obj.odt_file:
            logger.warn("The plone template '%s' doesn't have odt file" % ppath)
            ret.append((ppath, ospath, 'no odt file'))
            continue
        if not os.path.exists(ospath):
            logger.warn("The template file '%s' doesn't exist" % ospath)
            ret.append((ppath, ospath, 'os path error'))
            continue
        with open(ospath, 'rb') as f:
            data = f.read()
            new_md5 = compute_md5(data)
            if obj.initial_md5 == new_md5:
                ret.append((ppath, ospath, 'unchanged'))
                continue
            elif obj.has_been_modified() and not force:
                ret.append((ppath, ospath, 'kept'))
                continue
            obj.initial_md5 = new_md5
            obj.style_modification_md5 = new_md5
            obj.odt_file.data = data
            modified(obj, Attributes(Interface, 'odt_file'))
            ret.append((ppath, ospath, 'replaced'))
    return ret


def update_dict_with_validation(original_dict, update_dict, error_message=_("Dict update collision on key")):
    for key in update_dict:
        if key in original_dict:
            raise Invalid(_("${error_message} for key = '${key}'",
                            mapping={'error_message': error_message, 'key': key}))

        original_dict[key] = update_dict[key]


def safe_encode(value, encoding='utf-8'):
    """
        Converts a value to encoding, only when it is not already encoded.
    """
    if isinstance(value, six.PY2 and unicode or bytes):  # noqa: F821
        return value.encode(encoding)
    return value


def ulocalized_time(date, long_format=None, time_only=None, custom_format=None,
                    domain='plonelocales', target_language=None, context=None,
                    request=None, month_lc=True, day_lc=True):
    """
        Return for a datetime the string value with week and mont translated.
        Take into account %a, %A, %b, %B
    """
    if not custom_format:
        # use toLocalizedTime
        plone = getMultiAdapter((context, request), name=u'plone')
        formatted_date = plone.toLocalizedTime(date, long_format, time_only)
    else:
        from Products.CMFPlone.i18nl10n import monthname_msgid
        from Products.CMFPlone.i18nl10n import monthname_msgid_abbr
        from Products.CMFPlone.i18nl10n import weekdayname_msgid
        from Products.CMFPlone.i18nl10n import weekdayname_msgid_abbr
        if request is None:
            portal = api.portal.get()
            request = portal.REQUEST
        # first replace parts to translate
        custom_format = custom_format.replace('%%', '_p_c_')

        conf = {
            'a': {'fct': weekdayname_msgid_abbr, 'fmt': '%w', 'low': day_lc},
            'A': {'fct': weekdayname_msgid, 'fmt': '%w', 'low': day_lc},
            'b': {'fct': monthname_msgid_abbr, 'fmt': '%m', 'low': month_lc},
            'B': {'fct': monthname_msgid, 'fmt': '%m', 'low': month_lc},
        }
        matches = re.findall(r'%([aAbB])', custom_format)
        for match in sorted(set(matches)):
            # function( int(date.strftime(format) )
            msgid = conf[match]['fct'](int(date.strftime(conf[match]['fmt'])))
            repl = i18n.translate(msgid, domain, context=request, target_language=target_language)
            if conf[match]['low']:
                repl = repl.lower()
            custom_format = re.sub('%{}'.format(match), repl, custom_format)

        # then format date
        custom_format = custom_format.replace('_p_c_', '%%')
        if HAS_PLONE_5:
            formatted_date = date.strftime(custom_format)
        else:
            formatted_date = date.strftime(custom_format.encode('utf8'))
    return safe_unicode(formatted_date)


def remove_tmp_file(filename):
    """Do not break if unable to remove temporary file, but log error if any."""
    try:
        os.remove(filename)
    except OSError:
        logger.warn("Could not remove temporary file at {0}".format(filename))


def update_oo_config():
    """ Update config following buildout var """
    key_template = 'collective.documentgenerator.browser.controlpanel.IDocumentGeneratorControlPanelSchema.{}'
    var = {'oo_server': 'OO_SERVER', 'oo_port_list': 'OO_PORT', 'uno_path': 'PYTHON_UNO'}
    for key in var.keys():
        full_key = key_template.format(key)
        configured_oo_option = api.portal.get_registry_record(full_key)
        env_value = os.getenv(var.get(key, 'NO_ONE'), None)
        if env_value:
            new_oo_option = type(configured_oo_option)(os.getenv(var.get(key, 'NO_ONE'), ''))
            if new_oo_option and new_oo_option != configured_oo_option:
                api.portal.set_registry_record(full_key, new_oo_option)
    logger.info("LibreOffice configuration updated for " + getSite().getId())


def update_oo_config_after_bigbang(event):
    setSite(event.object)
    try:
        update_oo_config()
    except Exception:
        logger.error("Update LibreOffice configuration failed", exc_info=1)


def get_site_root_relative_path(obj):
    return "/" + '/'.join(
        getToolByName(obj, 'portal_url').getRelativeContentPath(obj)
    )


def temporary_file_name(suffix=''):
    tmp_dir = os.getenv('CUSTOM_TMP', None)
    if tmp_dir and not os.path.exists(tmp_dir):
        os.mkdir(tmp_dir)
    return tempfile.mktemp(suffix=suffix, dir=tmp_dir)


def create_temporary_file(initial_file=None, base_name=''):
    tmp_filename = temporary_file_name(suffix=base_name)
    # create the file in any case
    with open(tmp_filename, 'wb') as tmp_file:
        if initial_file:
            tmp_file.write(initial_file.data)
    return tmp_file


def clean_notes(pod_template):
    """ Use appy.pod Cleaner to clean notes (comments). """
    cleaned = 0
    odt_file = pod_template.odt_file
    if odt_file:
        # write file to /tmp to be able to use appy.pod Cleaner
        tmp_file = create_temporary_file(odt_file, '-to-clean.odt')
        if six.PY2:
            cleaner = Cleaner(path=tmp_file.name, verbose=1)
        else:
            cleaner = Cleaner(path=tmp_file.name, silent=False)
        cleaned = cleaner.run()
        if cleaned:
            manually_modified = pod_template.has_been_modified()
            with open(tmp_file.name, 'rb') as res_file:
                # update template
                result = NamedBlobFile(
                    data=res_file.read(),
                    contentType=odt_file.contentType,
                    filename=pod_template.odt_file.filename)
            pod_template.odt_file = result
            if not manually_modified:
                pod_template.style_modification_md5 = pod_template.current_md5
            extras = 'pod_template={0} cleaned_parts={1}'.format(
                repr(pod_template), cleaned)
            fplog('clean_notes', extras=extras)
        remove_tmp_file(tmp_file.name)

    return bool(cleaned)

1	# -- coding: utf-8 --
2	from collective.documentgenerator import _	2✔
3	from collective.documentgenerator.config import HAS_PLONE_5	2✔
4	from imio.helpers.security import fplog	2✔
5	from plone import api	2✔
6	from plone.namedfile.file import NamedBlobFile	2✔
7	from Products.CMFCore.utils import getToolByName	2✔
8	from Products.CMFPlone.utils import safe_unicode	2✔
9	from zope import i18n	2✔
10	from zope.component import getMultiAdapter	2✔
11	from zope.component.hooks import getSite	2✔
12	from zope.component.hooks import setSite	2✔
13	from zope.interface import Interface	2✔
14	from zope.interface import Invalid	2✔
15	from zope.lifecycleevent import Attributes	2✔
16	from zope.lifecycleevent import modified	2✔
17
18	import hashlib	2✔
19	import logging	2✔
20	import os	2✔
21	import re	2✔
22	import six	2✔
23	import tempfile	2✔
24
25
26	if six.PY2:	2✔
NEW 27	from appy.bin.odfclean import Cleaner	×
28	else:
29	from appy.bin.oclean import Cleaner	2✔
30
31	logger = logging.getLogger('collective.documentgenerator')	2✔
32
33
34	def translate(msgid, domain='collective.documentgenerator'):	2✔
35	portal = api.portal.get()	2✔
36	translation = i18n.translate(	2✔
37	msgid,
38	domain=domain,
39	context=portal.REQUEST
40	)
41	return translation	2✔
42
43
44	def compute_md5(data):	2✔
45	md5 = hashlib.md5(data).hexdigest()	2✔
46	return md5	2✔
47
48
49	def update_templates(templates, profile='', force=False):	2✔
50	"""
51	function to manage templates update.
52	# see http://trac.imio.be/trac/ticket/9383 for full implementation
53	:param list templates: list of tuples containing ('plone-template-path', 'os-template-path')
54	:param str profile: profile path stored on template (or various identification)
55	:param bool force: force overrides of templates
56	"""
57	# Don't use profile now !
58	portal = api.portal.getSite()	2✔
59	ret = []	2✔
60	for (ppath, ospath) in templates:	2✔
61	ppath = ppath.strip('/ ')	2✔
62	obj = portal.unrestrictedTraverse(ppath, default=None)	2✔
63	if not obj:	2✔
64	logger.warn("The plone template '%s' was not found" % ppath)	2✔
65	ret.append((ppath, ospath, 'plone path error'))	2✔
66	continue	2✔
67	if not obj.odt_file:	2✔
68	logger.warn("The plone template '%s' doesn't have odt file" % ppath)	×
69	ret.append((ppath, ospath, 'no odt file'))	×
70	continue	×
71	if not os.path.exists(ospath):	2✔
72	logger.warn("The template file '%s' doesn't exist" % ospath)	2✔
73	ret.append((ppath, ospath, 'os path error'))	2✔
74	continue	2✔
75	with open(ospath, 'rb') as f:	2✔
76	data = f.read()	2✔
77	new_md5 = compute_md5(data)	2✔
78	if obj.initial_md5 == new_md5:	2✔
79	ret.append((ppath, ospath, 'unchanged'))	2✔
80	continue	2✔
81	elif obj.has_been_modified() and not force:	2✔
82	ret.append((ppath, ospath, 'kept'))	2✔
83	continue	2✔
84	obj.initial_md5 = new_md5	2✔
85	obj.style_modification_md5 = new_md5	2✔
86	obj.odt_file.data = data	2✔
87	modified(obj, Attributes(Interface, 'odt_file'))	2✔
88	ret.append((ppath, ospath, 'replaced'))	2✔
89	return ret	2✔
90
91
92	def update_dict_with_validation(original_dict, update_dict, error_message=_("Dict update collision on key")):	2✔
93	for key in update_dict:	2✔
94	if key in original_dict:	2✔
95	raise Invalid(_("${error_message} for key = '${key}'",	2✔
96	mapping={'error_message': error_message, 'key': key}))
97
98	original_dict[key] = update_dict[key]	2✔
99
100
101	def safe_encode(value, encoding='utf-8'):	2✔
102	"""
103	Converts a value to encoding, only when it is not already encoded.
104	"""
NEW 105	if isinstance(value, six.PY2 and unicode or bytes): # noqa: F821	×
NEW 106	return value.encode(encoding)	×
NEW 107	return value	×
108
109
110	def ulocalized_time(date, long_format=None, time_only=None, custom_format=None,	2✔
111	domain='plonelocales', target_language=None, context=None,
112	request=None, month_lc=True, day_lc=True):
113	"""
114	Return for a datetime the string value with week and mont translated.
115	Take into account %a, %A, %b, %B
116	"""
117	if not custom_format:	2✔
118	# use toLocalizedTime
119	plone = getMultiAdapter((context, request), name=u'plone')	2✔
120	formatted_date = plone.toLocalizedTime(date, long_format, time_only)	2✔
121	else:
122	from Products.CMFPlone.i18nl10n import monthname_msgid	2✔
123	from Products.CMFPlone.i18nl10n import monthname_msgid_abbr	2✔
124	from Products.CMFPlone.i18nl10n import weekdayname_msgid	2✔
125	from Products.CMFPlone.i18nl10n import weekdayname_msgid_abbr	2✔
126	if request is None:	2✔
127	portal = api.portal.get()	×
128	request = portal.REQUEST	×
129	# first replace parts to translate
130	custom_format = custom_format.replace('%%', '_p_c_')	2✔
131
132	conf = {	2✔
133	'a': {'fct': weekdayname_msgid_abbr, 'fmt': '%w', 'low': day_lc},
134	'A': {'fct': weekdayname_msgid, 'fmt': '%w', 'low': day_lc},
135	'b': {'fct': monthname_msgid_abbr, 'fmt': '%m', 'low': month_lc},
136	'B': {'fct': monthname_msgid, 'fmt': '%m', 'low': month_lc},
137	}
138	matches = re.findall(r'%([aAbB])', custom_format)	2✔
139	for match in sorted(set(matches)):	2✔
140	# function( int(date.strftime(format) )
141	msgid = conf[match]['fct'](int(date.strftime(conf[match]['fmt'])))	2✔
142	repl = i18n.translate(msgid, domain, context=request, target_language=target_language)	2✔
143	if conf[match]['low']:	2✔
144	repl = repl.lower()	2✔
145	custom_format = re.sub('%{}'.format(match), repl, custom_format)	2✔
146
147	# then format date
148	custom_format = custom_format.replace('_p_c_', '%%')	2✔
149	if HAS_PLONE_5:	2✔
150	formatted_date = date.strftime(custom_format)	2✔
151	else:
NEW 152	formatted_date = date.strftime(custom_format.encode('utf8'))	×
153	return safe_unicode(formatted_date)	2✔
154
155
156	def remove_tmp_file(filename):	2✔
157	"""Do not break if unable to remove temporary file, but log error if any."""
158	try:	2✔
159	os.remove(filename)	2✔
160	except OSError:	×
161	logger.warn("Could not remove temporary file at {0}".format(filename))	×
162
163
164	def update_oo_config():	2✔
165	""" Update config following buildout var """
166	key_template = 'collective.documentgenerator.browser.controlpanel.IDocumentGeneratorControlPanelSchema.{}'	2✔
167	var = {'oo_server': 'OO_SERVER', 'oo_port_list': 'OO_PORT', 'uno_path': 'PYTHON_UNO'}	2✔
168	for key in var.keys():	2✔
169	full_key = key_template.format(key)	2✔
170	configured_oo_option = api.portal.get_registry_record(full_key)	2✔
171	env_value = os.getenv(var.get(key, 'NO_ONE'), None)	2✔
172	if env_value:	2✔
173	new_oo_option = type(configured_oo_option)(os.getenv(var.get(key, 'NO_ONE'), ''))	2✔
174	if new_oo_option and new_oo_option != configured_oo_option:	2✔
175	api.portal.set_registry_record(full_key, new_oo_option)	2✔
176	logger.info("LibreOffice configuration updated for " + getSite().getId())	2✔
177
178
179	def update_oo_config_after_bigbang(event):	2✔
180	setSite(event.object)	×
181	try:	×
182	update_oo_config()	×
183	except Exception:	×
184	logger.error("Update LibreOffice configuration failed", exc_info=1)	×
185
186
187	def get_site_root_relative_path(obj):	2✔
188	return "/" + '/'.join(	2✔
189	getToolByName(obj, 'portal_url').getRelativeContentPath(obj)
190	)
191
192
193	def temporary_file_name(suffix=''):	2✔
194	tmp_dir = os.getenv('CUSTOM_TMP', None)	2✔
195	if tmp_dir and not os.path.exists(tmp_dir):	2✔
196	os.mkdir(tmp_dir)	2✔
197	return tempfile.mktemp(suffix=suffix, dir=tmp_dir)	2✔
198
199
200	def create_temporary_file(initial_file=None, base_name=''):	2✔
201	tmp_filename = temporary_file_name(suffix=base_name)	2✔
202	# create the file in any case
203	with open(tmp_filename, 'wb') as tmp_file:	2✔
204	if initial_file:	2✔
205	tmp_file.write(initial_file.data)	2✔
206	return tmp_file	2✔
207
208
209	def clean_notes(pod_template):	2✔
210	""" Use appy.pod Cleaner to clean notes (comments). """
211	cleaned = 0	2✔
212	odt_file = pod_template.odt_file	2✔
213	if odt_file:	2✔
214	# write file to /tmp to be able to use appy.pod Cleaner
215	tmp_file = create_temporary_file(odt_file, '-to-clean.odt')	2✔
216	if six.PY2:	2✔
NEW 217	cleaner = Cleaner(path=tmp_file.name, verbose=1)	×
218	else:
219	cleaner = Cleaner(path=tmp_file.name, silent=False)	2✔
220	cleaned = cleaner.run()	2✔
221	if cleaned:	2✔
222	manually_modified = pod_template.has_been_modified()	2✔
223	with open(tmp_file.name, 'rb') as res_file:	2✔
224	# update template
225	result = NamedBlobFile(	2✔
226	data=res_file.read(),
227	contentType=odt_file.contentType,
228	filename=pod_template.odt_file.filename)
229	pod_template.odt_file = result	2✔
230	if not manually_modified:	2✔
231	pod_template.style_modification_md5 = pod_template.current_md5	2✔
232	extras = 'pod_template={0} cleaned_parts={1}'.format(	2✔
233	repr(pod_template), cleaned)
234	fplog('clean_notes', extras=extras)	2✔
235	remove_tmp_file(tmp_file.name)	2✔
236
237	return bool(cleaned)	2✔

collective / collective.documentgenerator / 9287122129

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