Skip to content

cloudooo
cloudooo
Commit d4abf3ff authored by Jérome Perrin's avatar Jérome Perrin
Browse files
Options

*: target python3

parent b9d9540f
Show whitespace changes
Inline Side-by-side
Showing with 420 additions and 396 deletions
README.rst
View file @ d4abf3ff
.. warning:: this documentation is outdated. See https://lab.nexedi.com/nexedi/cloudooo , https://lab.nexedi.com/nexedi/slapos/tree/master/software/cloudooo or https://cloudooo.nexedi.com/ for more up to date information.
Install Cloudooo
================
::
$ python2.6 setup.py install
Warnings:
- you must have installed setuptools>=0.6c11 in this python.
$ python setup.py install
Install LibreOffice / OpenOffice.org
====================================
......
cloudooo/bin/cloudooo_tester.py
View file @ d4abf3ff
......@@ -2,11 +2,11 @@
import unittest
import sys
from base64 import encodestring
from xmlrpclib import ServerProxy
from base64 import encodebytes
from xmlrpc.client import ServerProxy
from getopt import getopt, GetoptError
DOCUMENT_STRING = """MemoryMonitor - TimeoutMonitor -
DOCUMENT_STRING = b"""MemoryMonitor - TimeoutMonitor -
RequestMonitor\n\nOOHandler\n\nMimemapper\n\nERP5\n"""
HOSTNAME = PORT = None
......@@ -15,17 +15,17 @@ class CloudoooTestCase(unittest.TestCase):
""" """
def setUp(self):
self.proxy_address = "http://%s:%s" % (HOSTNAME, PORT)
self.proxy_address = f"http://{HOSTNAME}:{PORT}"
def test_run_generate(self):
data = encodestring(DOCUMENT_STRING)
data = encodebytes(DOCUMENT_STRING)
proxy = ServerProxy(self.proxy_address, allow_none=True)
res = proxy.run_generate("t.text", data, None, 'pdf', 'text/plain')
self.assertEqual(res[1]['mime'], "application/pdf")
self.assertEqual(res[0], 200)
def test_set_metadata(self):
data = encodestring(DOCUMENT_STRING)
data = encodebytes(DOCUMENT_STRING)
proxy = ServerProxy(self.proxy_address, allow_none=True)
odt_data = proxy.convertFile(data, 'txt', 'odt')
metadata_dict = proxy.getFileMetadataItemList(odt_data, 'odt')
......@@ -45,7 +45,7 @@ def main():
opt_list, _ = getopt(sys.argv[1:], "",
["port=", "hostname="])
except GetoptError as e:
print >> sys.stderr, "%s \nUse --port and --hostname" % e
print("%s \nUse --port and --hostname" % e, file=sys.stderr)
sys.exit(2)
for opt, arg in opt_list:
......@@ -55,7 +55,7 @@ def main():
HOSTNAME = arg
if not HOSTNAME and not PORT:
print >> sys.stderr, "Use --port and --hostname"
print("Use --port and --hostname", file=sys.stderr)
sys.exit(2)
suite = unittest.TestLoader().loadTestsFromTestCase(CloudoooTestCase)
unittest.TextTestRunner(verbosity=2).run(suite)
cloudooo/bin/echo_cloudooo_conf.py
View file @ d4abf3ff
......@@ -6,4 +6,4 @@ def main():
cloudooo_conf_path = pkg_resources.resource_filename("cloudooo",
path.join("sample", "cloudooo.conf.in"))
print open(cloudooo_conf_path).read()
print(open(cloudooo_conf_path).read())
cloudooo/file.py
View file @ d4abf3ff
......@@ -39,12 +39,12 @@ from cloudooo.interfaces.file import IFile
@implementer(IFile)
class File(object):
class File:
"""File is used to manipulate one temporary file
stored into the filesystem.
"""
def __init__(self, base_folder_url, data, source_format):
def __init__(self, base_folder_url:str, data:bytes, source_format:str):
"""Create an file into file system and store the URL.
Keyword arguments:
base_folder_url -- Full path to create a temporary folder
......@@ -60,7 +60,7 @@ class File(object):
def _createDirectory(self):
return tempfile.mkdtemp(dir=self.base_folder_url)
def load(self):
def load(self) -> str:
"""Creates one Temporary Document and write data into it.
Return the url for the document.
"""
......@@ -92,7 +92,7 @@ class File(object):
remove(zipfile_path)
return file_path
def getContent(self, zip=False):
def getContent(self, zip=False) -> bytes:
"""Open the file and returns the content.
Keyword Arguments:
zip -- Boolean attribute. If True"""
......@@ -116,7 +116,7 @@ class File(object):
with open(self.url, 'rb') as f:
return f.read()
def getUrl(self):
def getUrl(self) -> str:
"""Returns full path."""
return self.url
......
cloudooo/handler/ffmpeg/handler.py
View file @ d4abf3ff
......@@ -36,7 +36,7 @@ from subprocess import Popen, PIPE
from tempfile import mktemp
@implementer(IHandler)
class Handler(object):
class Handler:
"""FFMPEG Handler is used to handler inputed audio and video files"""
def __init__(self, base_folder_url, data, source_format, **kw):
......
cloudooo/handler/imagemagick/handler.py
View file @ d4abf3ff
......@@ -38,7 +38,7 @@ from tempfile import mktemp
@implementer(IHandler)
class Handler(object):
class Handler:
"""ImageMagic Handler is used to handler images."""
def __init__(self, base_folder_url, data, source_format, **kw):
......@@ -73,6 +73,7 @@ class Handler(object):
stdout=PIPE,
stderr=PIPE,
close_fds=True,
text=True,
env=self.environment).communicate()
self.file.trash()
metadata_dict = {}
......
cloudooo/handler/imagemagick/tests/testImagemagickHandler.py
View file @ d4abf3ff
......@@ -61,7 +61,7 @@ class TestHandler(HandlerTestCase):
def testsetMetadata(self):
""" Test if metadata are inserted correclty """
handler = Handler(self.tmp_url, "", "png", **self.kw)
handler = Handler(self.tmp_url, b"", "png", **self.kw)
self.assertRaises(NotImplementedError, handler.setMetadata)
cloudooo/handler/ooo/application/application.py
View file @ d4abf3ff
......@@ -36,7 +36,7 @@ from psutil import pid_exists, Process, AccessDenied
@implementer(IApplication)
class Application(object):
class Application:
"""Base object to create an object that is possible manipulation a
process"""
......
cloudooo/handler/ooo/application/openoffice.py
View file @ d4abf3ff
......@@ -35,7 +35,7 @@ from psutil import AccessDenied, NoSuchProcess
from os.path import exists, join
from threading import Lock
from zope.interface import implementer
from application import Application
from .application import Application
from cloudooo.interfaces.lockable import ILockable
from cloudooo.util import logger
from cloudooo.handler.ooo.util import waitStartDaemon, \
......@@ -155,12 +155,12 @@ class OpenOffice(Application):
env["TMPDIR"] = self.path_user_installation
self._startProcess(self.command, env)
self._cleanRequest()
Application.start(self)
super().start()
def stop(self):
"""Stop the instance by pid. By the default
the signal is 15."""
Application.stop(self)
super().stop()
if socketStatus(self.hostname, self.port):
self._releaseOpenOfficePort()
self._cleanRequest()
......
cloudooo/handler/ooo/document.py
View file @ d4abf3ff
......@@ -30,7 +30,7 @@
from zope.interface import implementer
from zipfile import ZipFile
from StringIO import StringIO
from io import BytesIO
from lxml import etree
from cloudooo.interfaces.file import IOdfDocument
from cloudooo.file import File
......@@ -41,9 +41,10 @@ class FileSystemDocument(File):
@implementer(IOdfDocument)
class OdfDocument(object):
class OdfDocument:
"""Manipulates odf documents in memory"""
def __init__(self, data, source_format):
"""Open the the file in memory.
......@@ -51,22 +52,22 @@ class OdfDocument(object):
data -- Content of the document
source_format -- Document Extension
"""
self._zipfile = ZipFile(StringIO(data))
self._zipfile = ZipFile(BytesIO(data))
self.source_format = source_format
# XXX - Maybe parsed_content should not be here, but on OOGranulate
self.parsed_content = etree.fromstring(self.getContentXml())
def getContentXml(self):
"""Returns the content.xml file as string"""
def getContentXml(self) -> bytes:
"""Returns the content.xml file as bytes"""
return self._zipfile.read('content.xml')
def getFile(self, path):
"""If exists, returns file as string, else return an empty string"""
def getFile(self, path:str) -> bytes:
"""If exists, returns file as bytes, otherwise b''"""
try:
return self._zipfile.read(path)
except KeyError:
return ''
return b''
def trash(self):
"""Remove the file in memory."""
......
cloudooo/handler/ooo/filter.py
View file @ d4abf3ff
......@@ -33,9 +33,10 @@ from cloudooo.interfaces.filter import IFilter
@implementer(IFilter)
class Filter(object):
class Filter:
"""Filter of OOo."""
def __init__(self, extension, filter, mimetype, document_service, **kwargs):
"""Receives extension, filter and mimetype of filter and saves in object.
"""
......
cloudooo/handler/ooo/granulator.py
View file @ d4abf3ff
......@@ -29,7 +29,7 @@
##############################################################################
from zipfile import ZipFile
from StringIO import StringIO
from io import BytesIO
from lxml import etree
from os import path
from cloudooo.util import logger
......@@ -61,17 +61,17 @@ def getTemplatePath(format):
return path.join(path.dirname(__file__), 'template.%s' % format)
class OOGranulator(object):
class OOGranulator:
"""Granulate an OpenOffice document into tables, images, chapters and
paragraphs."""
def __init__(self, file, source_format):
def __init__(self, file:bytes, source_format:str):
self.document = OdfDocument(file, source_format)
def _odfWithoutContentXml(self, format='odt'):
"""Returns an odf document without content.xml
It is a way to escape from this issue: http://bugs.python.org/issue6818"""
new_odf_document = ZipFile(StringIO(), 'a')
new_odf_document = ZipFile(BytesIO(), 'a')
template_path = getTemplatePath(format)
template_file = ZipFile(template_path)
for item in template_file.filelist:
......@@ -202,7 +202,7 @@ class OOGranulator(object):
image_list = []
for xml_image in xml_image_list:
id = xml_image.values()[0].split('/')[-1]
id = list(xml_image.values())[0].split('/')[-1]
title = ''.join(xml_image.xpath(IMAGE_TITLE_XPATH_QUERY%(stylename_list[0], name_list[0], id),
namespaces=xml_image.nsmap))
if title != '':
......@@ -262,22 +262,17 @@ class OOGranulator(object):
chapter_list = self.document.parsed_content.xpath(
CHAPTER_XPATH_QUERY,
namespaces=self.document.parsed_content.nsmap)
return chapter_list
return [str(x) for x in chapter_list]
def getChapterItemList(self):
"""Returns the list of chapters in the form of (id, level)."""
id = 0
chapter_list = []
for chapter in self._getChapterList():
chapter_list.append((id, chapter.encode('utf-8')))
id += 1
return chapter_list
return list(enumerate(self._getChapterList()))
def getChapterItem(self, chapter_id):
"""Return the chapter in the form of (title, level)."""
chapter_list = self._getChapterList()
try:
chapter = chapter_list[chapter_id].encode('utf-8')
chapter = chapter_list[chapter_id]
return [chapter, chapter_id]
except IndexError:
msg = "Unable to find chapter %s at chapter list." % chapter_id
......
cloudooo/handler/ooo/handler.py
View file @ d4abf3ff
......@@ -31,7 +31,7 @@
import json
import pkg_resources
import mimetypes
from base64 import decodestring, encodestring
from base64 import decodebytes, encodebytes
from os import environ, path
from subprocess import Popen, PIPE
from cloudooo.handler.ooo.application.openoffice import openoffice
......@@ -46,7 +46,7 @@ from psutil import pid_exists
@implementer(IHandler)
class Handler(object):
class Handler:
"""OOO Handler is used to access the one Document and OpenOffice.
For each Document inputed is created on instance of this class to manipulate
the document. This Document must be able to create and remove a temporary
......@@ -76,9 +76,9 @@ class Handler(object):
# backward compatibility.
# The heuristic is "if it's not utf-8", let's assume it's iso-8859-15.
try:
unicode(data, 'utf-8')
data.decode('utf-8')
except UnicodeDecodeError:
data = unicode(data, 'iso-8859-15').encode('utf-8')
data = data.decode('iso-8859-15').encode('utf-8')
logger.warn("csv data is not utf-8, assuming iso-8859-15")
self.document = FileSystemDocument(
base_folder_url,
......@@ -99,7 +99,7 @@ class Handler(object):
'--document_url=%s' % self.document.getUrl()]
for arg in args:
command_list.insert(3, "--%s" % arg)
for k, v in kw.iteritems():
for k, v in kw.items():
command_list.append("--%s=%s" % (k, v))
return command_list
......@@ -138,16 +138,16 @@ class Handler(object):
stdout, stderr = self._subprocess(command_list)
if not stdout and stderr:
first_error = stderr
logger.error(stderr)
logger.error(stderr.decode())
self.document.restoreOriginal()
openoffice.restart()
kw['document_url'] = self.document.getUrl()
command = self._getCommand(*feature_list, **kw)
stdout, stderr = self._subprocess(command)
if not stdout and stderr:
second_error = "\nerror of the second run: " + stderr
logger.error(second_error)
raise Exception(first_error + second_error)
second_error = b"\nerror of the second run: " + stderr
logger.error(second_error.decode())
raise Exception((first_error + second_error).decode(errors='replace'))
return stdout, stderr
......@@ -186,10 +186,10 @@ class Handler(object):
kw['refresh'] = json.dumps(self.refresh)
openoffice.acquire()
try:
stdout, stderr = self._callUnoConverter(*['convert'], **kw)
stdout, _ = self._callUnoConverter(*['convert'], **kw)
finally:
openoffice.release()
url = stdout.replace('\n', '')
url = stdout.replace(b'\n', b'')
self.document.reload(url)
content = self.document.getContent(self.zip)
self.document.trash()
......@@ -198,8 +198,8 @@ class Handler(object):
def getMetadata(self, base_document=False):
"""Returns a dictionary with all metadata of document.
Keywords Arguments:
base_document -- Boolean variable. if true, the document is also returned
along with the metadata."""
base_document -- Boolean variable. if true, the document content (as bytes)
is also returned along with the metadata."""
logger.debug("getMetadata")
kw = dict(mimemapper=self._serializeMimemapper())
if base_document:
......@@ -208,10 +208,10 @@ class Handler(object):
feature_list = ['getmetadata']
openoffice.acquire()
try:
stdout, stderr = self._callUnoConverter(*feature_list, **kw)
stdout, _ = self._callUnoConverter(*feature_list, **kw)
finally:
openoffice.release()
metadata = json.loads(decodestring(stdout))
metadata = json.loads(decodebytes(stdout))
if 'document_url' in metadata:
self.document.reload(metadata['document_url'])
metadata['Data'] = self.document.getContent()
......@@ -224,12 +224,11 @@ class Handler(object):
Keyword arguments:
metadata -- expected an dictionary with metadata.
"""
metadata_pickled = json.dumps(metadata)
logger.debug("setMetadata")
kw = dict(metadata=encodestring(metadata_pickled))
metadata_pickled = json.dumps(metadata).encode()
kw = dict(metadata=encodebytes(metadata_pickled).decode())
openoffice.acquire()
try:
stdout, stderr = self._callUnoConverter(*['setmetadata'], **kw)
self._callUnoConverter(*['setmetadata'], **kw)
finally:
openoffice.release()
doc_loaded = self.document.getContent()
......
cloudooo/handler/ooo/helper/unoconverter.py
View file @ d4abf3ff
......@@ -40,11 +40,6 @@ from base64 import b64encode, b64decode
from functools import partial
from getopt import getopt, GetoptError
try:
basestring
except NameError:
basestring = str
__doc__ = """
usage: unodocument [options]
......@@ -81,7 +76,17 @@ Options:
"""
class UnoDocument(object):
MARKER = []
def next(gen, default=MARKER):
try:
return gen.__next__()
except StopIteration:
if default is MARKER:
raise
return default
class UnoDocument:
"""A module to easily work with OpenOffice.org."""
def __init__(self, service_manager, document_url,
......@@ -242,9 +247,9 @@ class UnoDocument(object):
continue
property_value = getattr(container, property_name, '')
if property_value:
if isinstance(property_value, basestring):
if isinstance(property_value, str):
metadata[property_name] = property_value
elif isinstance(property_value, tuple) and isinstance(property_value[0], basestring):
elif isinstance(property_value, tuple) and isinstance(property_value[0], str):
metadata[property_name] = property_value
else:
try:
......@@ -284,7 +289,7 @@ class UnoDocument(object):
if isinstance(current_value, tuple):
if isinstance(value, list):
value = tuple(value)
elif isinstance(value, basestring):
elif isinstance(value, str):
# BBB: old ERP5 code sends Keywords as a string
# separated by a whitespace.
value = tuple(value.split(' '))
......@@ -294,7 +299,7 @@ class UnoDocument(object):
else:
new_properties.append([prop, value])
for prop, value in new_properties:
if isinstance(value, basestring):
if isinstance(value, str):
user_defined_properties.addProperty(prop, 0, '')
user_defined_properties.setPropertyValue(prop, value)
self.document_loaded.store()
......@@ -311,7 +316,7 @@ def main():
help_msg = "\nUse --help or -h\n"
try:
opt_list, arg_list = getopt(sys.argv[1:], "h", ["help", "test",
opt_list, _ = getopt(sys.argv[1:], "h", ["help", "test",
"convert", "getmetadata", "setmetadata",
"uno_path=", "office_binary_path=",
"hostname=", "port=", "source_format=",
......@@ -325,10 +330,8 @@ def main():
param_list = [tuple[0] for tuple in iter(opt_list)]
try:
import json
except ImportError:
import simplejson as json
metadata = mimemapper = script = None
hostname = port = office_binary_path = uno_path = None
document_url = destination_format = source_format = infilter = refresh = None
......
cloudooo/handler/ooo/helper/unomimemapper.py
View file @ d4abf3ff
......@@ -30,23 +30,10 @@
##############################################################################
import sys
try:
import json
except ImportError:
import simplejson as json
import json
import helper_util
from getopt import getopt, GetoptError
# python3 support
try:
basestring
except NameError:
basestring = str
try:
long
except NameError:
long = int
__doc__ = """
......@@ -65,7 +52,7 @@ Options:
"""
class UnoMimemapper(object):
class UnoMimemapper:
""" """
def __init__(self, hostname, port, uno_path=None, office_binary_path=None):
......@@ -74,7 +61,7 @@ class UnoMimemapper(object):
uno_path,
office_binary_path)
def _getElementNameByService(self, uno_service, ignore_name_list=[]):
def _getElementNameByService(self, uno_service, ignore_name_list):
"""Returns an dict with elements."""
name_list = uno_service.getElementNames()
service_dict = {}
......@@ -84,7 +71,7 @@ class UnoMimemapper(object):
for obj in iter(element_list):
if obj.Name in ignore_name_list:
continue
if not isinstance(obj.Value, (bool, int, long, basestring, tuple)):
if not isinstance(obj.Value, (bool, int, str, tuple)):
continue
element_dict[obj.Name] = obj.Value
service_dict[name] = element_dict
......
cloudooo/handler/ooo/mimemapper.py
View file @ d4abf3ff
......@@ -33,14 +33,14 @@ from re import findall
from subprocess import Popen, PIPE
from subprocess import STDOUT
from zope.interface import implementer
from filter import Filter
from .filter import Filter
from os import environ, path
from cloudooo.interfaces.mimemapper import IMimemapper
import json
@implementer(IMimemapper)
class MimeMapper(object):
class MimeMapper:
"""Load all filters from OOo. You can get the filter you want or all
filters of the specific extension.
"""
......@@ -63,7 +63,7 @@ class MimeMapper(object):
def _typeToDocumentService(self, document_type):
"""Returns the document service according to document type."""
for k, v in self._document_type_dict.iteritems():
for k, v in self._document_type_dict.items():
if k.startswith(document_type):
return v
......@@ -124,7 +124,7 @@ class MimeMapper(object):
'Text', # Use 'Text - Choose Encoding' instead
'Text (StarWriter/Web)', # Use 'Text - Choose Encoding (Writer/Web)' instead
]
for filter_name, value in filter_dict.iteritems():
for filter_name, value in filter_dict.items():
if filter_name in ooo_disable_filter_list:
continue
ui_name = value.get('UIName')
......@@ -211,7 +211,7 @@ class MimeMapper(object):
'pdf': ['com.sun.star.drawing.DrawingDocument'],
'xls': ['com.sun.star.sheet.SpreadsheetDocument'],
})
self.document_service_list = self._extension_list_by_type.keys()
self.document_service_list = list(self._extension_list_by_type.keys())
self._loaded = True
def getFilterName(self, extension, document_service):
......@@ -226,11 +226,11 @@ class MimeMapper(object):
filter_list = [filter for filter in self.getFilterList(extension) \
if filter.getDocumentService() == document_service]
if len(filter_list) > 1:
for filter in iter(filter_list):
for filter in filter_list:
if filter.isPreferred():
return filter.getName()
else:
for filter in iter(filter_list):
for filter in filter_list:
if filter.getName().endswith("Export"):
return filter.getName()
filter_list.sort(key=lambda x: x.getSortIndex())
......@@ -272,10 +272,10 @@ class MimeMapper(object):
allowed_extension_list.extend(self._extension_list_by_type.get(document_type, []))
# gets list of extensions of each document type if document_type_list isn't
# empty.
for type in iter(document_type_list):
for type in document_type_list:
# gets list of extensions with key document type
extension_list = self._extension_list_by_type.get(type)
for ext in iter(extension_list):
for ext in extension_list:
if not ext in allowed_extension_list:
allowed_extension_list.append(ext)
return tuple(allowed_extension_list)
......
cloudooo/handler/ooo/monitor/__init__.py
View file @ d4abf3ff
from request import MonitorRequest
from memory import MonitorMemory
from sleeping_time import MonitorSpleepingTime
from .request import MonitorRequest
from .memory import MonitorMemory
from .sleeping_time import MonitorSpleepingTime
from cloudooo.handler.ooo.application.openoffice import openoffice
from cloudooo.util import convertStringToBool
......
cloudooo/handler/ooo/monitor/memory.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from monitor import Monitor
from .monitor import Monitor
from multiprocessing import Process
import psutil
from cloudooo.util import logger
......@@ -54,7 +54,7 @@ class MonitorMemory(Monitor, Process):
if not hasattr(self, 'process') or \
self.process.pid != int(self.openoffice.pid()):
self.create_process()
return self.process.memory_info().rss / (1024 * 1024)
return self.process.memory_info().rss // (1024 * 1024)
except TypeError:
logger.debug("OpenOffice is stopped")
return 0
......
cloudooo/handler/ooo/monitor/monitor.py
View file @ d4abf3ff
......@@ -33,7 +33,7 @@ from cloudooo.interfaces.monitor import IMonitor
@implementer(IMonitor)
class Monitor(object):
class Monitor:
""" """
def __init__(self, openoffice, interval):
......
cloudooo/handler/ooo/monitor/request.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from monitor import Monitor
from .monitor import Monitor
from threading import Thread
from cloudooo.util import logger
from time import sleep
......
cloudooo/handler/ooo/monitor/sleeping_time.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from monitor import Monitor
from .monitor import Monitor
from threading import Thread
from cloudooo.util import logger
from time import sleep, time
......
cloudooo/handler/ooo/monitor/timeout.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from monitor import Monitor
from .monitor import Monitor
from multiprocessing import Process
from time import sleep
from cloudooo.util import logger
......
cloudooo/handler/ooo/tests/HighTestLoad.py
View file @ d4abf3ff
......@@ -31,11 +31,11 @@
import sys
from multiprocessing import Process
from os import listdir
from xmlrpclib import ServerProxy
from xmlrpc.client import ServerProxy
from os.path import join
from getopt import getopt, GetoptError
from time import ctime, time
from base64 import encodestring
from base64 import encodebytes
__doc__ = """
usage: python HighTestLoad.py [options]
......@@ -51,7 +51,7 @@ Options:
"""
class Log(object):
class Log:
"""Object to manipulate the log file"""
def __init__(self, log_path, mode='a'):
......@@ -109,7 +109,7 @@ class Client(Process):
folder_list = listdir(self.folder)[:self.number_of_request]
for filename in folder_list:
file_path = join(self.folder, filename)
data = encodestring(open(file_path).read())
data = encodebytes(open(file_path).read())
self.log.msg("Input: %s\n" % file_path)
try:
now = time()
......@@ -118,7 +118,7 @@ class Client(Process):
self.log.msg("Duration: %s\n" % response_duration)
time_list.append(response_duration)
self.log.flush()
except Exception, err:
except Exception as err:
self.log.msg("%s\n" % str(err))
self.number_of_request -= 1
......@@ -132,15 +132,15 @@ def main():
help_msg = "\nUse --help or -h"
try:
opt_list, arg_list = getopt(sys.argv[1:], "hc:n:f:s:l:", ["help"])
except GetoptError, msg:
except GetoptError as msg:
msg = msg.msg + help_msg
print >> sys.stderr, msg
print(msg, file=sys.stderr)
sys.exit(2)
kw = {}
for opt, arg in opt_list:
if opt in ('-h', '--help'):
print >> sys.stdout, __doc__
print(__doc__, file=sys.stdout)
sys.exit(2)
elif opt == '-c':
number_client = int(arg)
......
cloudooo/handler/ooo/tests/HighTestLoadGranulate.py
View file @ d4abf3ff
......@@ -31,10 +31,10 @@
import sys
from multiprocessing import Process
from os import listdir
from xmlrpclib import ServerProxy
from xmlrpc.client import ServerProxy
from os.path import join
from getopt import getopt, GetoptError
from base64 import encodestring
from base64 import encodebytes
"""
XXX - This file is duplicated and should be refactored to be used for all
......@@ -58,7 +58,7 @@ Options:
"""
class Log(object):
class Log:
"""Object to manipulate the log file"""
def __init__(self, log_path, mode='a'):
......@@ -114,13 +114,13 @@ class Client(Process):
folder_list = listdir(self.folder)[:self.number_of_request]
for filename in folder_list:
file_path = join(self.folder, filename)
data = encodestring(open(file_path).read())
data = encodebytes(open(file_path).read())
try:
proxy.getTableItemList(data, 'odt')
proxy.getChapterItemList(data, 'odt')
proxy.getImageItemList(data, 'odt')
self.log.msg("Success\n")
except Exception, err:
except Exception as err:
self.log.msg("Error - %s\n"%str(err))
self.number_of_request -= 1
self.log.close()
......@@ -131,15 +131,15 @@ def main():
help_msg = "\nUse --help or -h"
try:
opt_list, arg_list = getopt(sys.argv[1:], "hc:n:f:s:l:", ["help"])
except GetoptError, msg:
except GetoptError as msg:
msg = msg.msg + help_msg
print >> sys.stderr, msg
print(msg, file=sys.stderr)
sys.exit(2)
kw = {}
for opt, arg in opt_list:
if opt in ('-h', '--help'):
print >> sys.stdout, __doc__
print(__doc__, file=sys.stdout)
sys.exit(2)
elif opt == '-c':
number_client = int(arg)
......
cloudooo/handler/ooo/tests/testOooAllFormat.py
View file @ d4abf3ff
......@@ -62,5 +62,5 @@ class TestAllFormats(TestCase):
request = {'document_type': document_type}
extension_list = self.proxy.getAllowedExtensionList(request)
for extension in extension_list:
with self.subTest(extension):
self._testConvertFile(input_url, source_format, extension[0], None)
cloudooo/handler/ooo/tests/testOooAllFormatERP5Compatibility.py
View file @ d4abf3ff
......@@ -28,9 +28,10 @@
#
##############################################################################
from xmlrpclib import Fault
from xmlrpc.client import Fault
from cloudooo.tests.cloudoooTestCase import TestCase
from base64 import encodestring
from base64 import encodebytes
class TestAllFormatsERP5Compatibility(TestCase):
"""
......@@ -54,27 +55,21 @@ class TestAllFormatsERP5Compatibility(TestCase):
"""Test all spreadsheet formats"""
self.runTestForType('data/test.ods', 'ods', 'application/vnd.oasis.opendocument.spreadsheet')
def runTestForType(self, filename, source_format, source_mimetype):
def runTestForType(self, filename:str, source_format:str, source_mimetype:str) -> None:
"""Generic test"""
extension_list = self.proxy.getAllowedTargetItemList(source_mimetype)[1]['response_data']
fault_list = []
for extension, format in extension_list:
try:
data_output = self.proxy.run_generate(filename,
encodestring(
open(filename).read()),
with open(filename, 'rb') as f:
encoded_data = encodebytes(f.read()).decode()
for extension, _ in extension_list:
with self.subTest(extension):
_, data_output, fault = self.proxy.run_generate(filename,
encoded_data,
None,
extension,
source_mimetype)[1]
source_mimetype)
self.assertFalse(fault)
data_output = data_output['data']
file_type = self._getFileType(data_output)
if file_type.endswith(": empty"):
fault_list.append((source_format, extension, file_type))
except Fault as err:
fault_list.append((source_format, extension, err.faultString))
if fault_list:
template_message = 'input_format: %r\noutput_format: %r\n traceback:\n%s'
message = '\n'.join([template_message % fault for fault in fault_list])
self.fail('Failed Conversions:\n' + message)
self.assertNotIn(": empty", file_type)
cloudooo/handler/ooo/tests/testOooFileSystemDocument.py
View file @ d4abf3ff
......@@ -30,8 +30,8 @@
import unittest
import magic
from StringIO import StringIO
from base64 import decodestring
from io import BytesIO
from base64 import decodebytes
from os import path
from zipfile import ZipFile
from cloudooo.handler.ooo.document import FileSystemDocument
......@@ -43,7 +43,7 @@ class TestFileSystemDocument(unittest.TestCase):
def setUp(self):
"""Create data to tests and instantiated a FileSystemDocument"""
self.tmp_url = '/tmp'
self.data = decodestring("cloudooo Test")
self.data = decodebytes(b"cloudooo Test")
self.fsdocument = FileSystemDocument(self.tmp_url, self.data, 'txt')
def tearDown(self):
......@@ -59,7 +59,7 @@ class TestFileSystemDocument(unittest.TestCase):
document_test_url = path.join(self.fsdocument.directory_name,
document_filename)
with open(document_test_url, 'wb') as f:
f.write(decodestring(b"Test Document"))
f.write(decodebytes(b"Test Document"))
self.fsdocument.reload(document_test_url)
self.assertEqual(path.exists(old_document_url), False)
self.assertNotEqual(self.fsdocument.original_data,
......@@ -112,7 +112,7 @@ class TestFileSystemDocument(unittest.TestCase):
mime = magic.Magic(mime=True)
mimetype = mime.from_buffer(zip_file)
self.assertEqual(mimetype, 'application/zip')
ziptest = ZipFile(StringIO(zip_file), 'r')
ziptest = ZipFile(BytesIO(zip_file), 'r')
self.assertEqual(len(ziptest.filelist), 2)
for file in ziptest.filelist:
if file.filename.endswith("document2"):
......@@ -130,7 +130,7 @@ class TestFileSystemDocument(unittest.TestCase):
self.assertEqual(mimetype, "application/zip")
mimetype = mime.from_buffer(zipdocument.getContent())
self.assertEqual(mimetype, "text/html")
zipfile = ZipFile(StringIO(zipdocument.getContent(True)))
zipfile = ZipFile(BytesIO(zipdocument.getContent(True)))
self.assertEqual(sorted(zipfile.namelist()),
sorted(['logo.gif', 'test.htm']))
......
cloudooo/handler/ooo/tests/testOooGranulator.py
View file @ d4abf3ff
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2010 Nexedi SA and Contributors. All Rights Reserved.
......@@ -30,7 +29,7 @@
##############################################################################
from zipfile import ZipFile
from StringIO import StringIO
from io import BytesIO
from lxml import etree
from cloudooo.tests.handlerTestCase import HandlerTestCase
from cloudooo.handler.ooo.granulator import OOGranulator
......@@ -71,7 +70,7 @@ class TestOOGranulator(HandlerTestCase):
data = f.read()
oogranulator = OOGranulator(data, 'odt')
table_data_doc = oogranulator.getTable('Developers')
content_xml_str = ZipFile(StringIO(table_data_doc)).read('content.xml')
content_xml_str = ZipFile(BytesIO(table_data_doc)).read('content.xml')
content_xml = etree.fromstring(content_xml_str)
table_list = content_xml.xpath('//table:table',
namespaces=content_xml.nsmap)
......@@ -129,7 +128,7 @@ class TestOOGranulator(HandlerTestCase):
"""Test if getImage() returns the right image file successfully"""
with open('./data/granulate_test.odt', 'rb') as f:
data = f.read()
zip = ZipFile(StringIO(data))
zip = ZipFile(BytesIO(data))
image_id = '10000000000000C80000009C38276C51.jpg'
original_image = zip.read('Pictures/%s' % image_id)
geted_image = self.oogranulator.getImage(image_id)
......@@ -161,8 +160,8 @@ class TestOOGranulator(HandlerTestCase):
big_paragraph = self.oogranulator.getParagraph(5)
self.assertEqual('P8', big_paragraph[1])
self.assertTrue(big_paragraph[0].startswith(u'A prática cotidiana prova'))
self.assertTrue(big_paragraph[0].endswith(u'corresponde às necessidades.'))
self.assertTrue(big_paragraph[0].startswith('A prática cotidiana prova'))
self.assertTrue(big_paragraph[0].endswith('corresponde às necessidades.'))
def testGetParagraphItemWithoutSuccess(self):
"""Test if getParagraphItem() returns None for not existent id"""
......
cloudooo/handler/ooo/tests/testOooHandler.py
View file @ d4abf3ff
......@@ -30,7 +30,7 @@
import magic
from os import path
from base64 import encodestring, decodestring
from base64 import encodebytes, decodebytes
from cloudooo.tests.handlerTestCase import HandlerTestCase
from cloudooo.handler.ooo.handler import Handler
from cloudooo.handler.ooo.application.openoffice import openoffice
......
cloudooo/handler/ooo/tests/testOooHighLoad.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from base64 import encodestring, decodestring
from base64 import encodebytes, decodebytes
from multiprocessing import Process, Array
from cloudooo.tests.cloudoooTestCase import TestCase
import magic
......@@ -39,8 +39,8 @@ mime_decoder = magic.Magic(mime=True)
def basicTestToGenerate(id, proxy, data, source_format, destination_format,
result_list):
"""Test to use method generate of server"""
document = proxy.convertFile(encodestring(data), source_format, destination_format)
mimetype = mime_decoder.from_buffer(decodestring(document))
document = proxy.convertFile(encodebytes(data), source_format, destination_format)
mimetype = mime_decoder.from_buffer(decodebytes(document))
assert mimetype == 'application/pdf'
result_list[id] = True
......
cloudooo/handler/ooo/tests/testOooLegacyInterface.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
#
##############################################################################
from base64 import encodestring
from base64 import encodebytes
from cloudooo.tests.cloudoooTestCase import TestCase
from pkg_resources import resource_filename
......@@ -39,10 +39,11 @@ class TestLegacyInterface(TestCase):
"""Check implicit base conversion of HTML documents."""
filename = resource_filename('cloudooo.handler.ooo.tests.data',
'test_failure_conversion.html')
data = open(filename, 'r').read()
with open(filename, 'rb') as f:
data = f.read()
status, response_dict, message = self.proxy.run_convert(
filename,
encodestring(data),
encodebytes(data).decode(),
None,
None,
'text/html')
......@@ -55,9 +56,10 @@ class TestLegacyInterface(TestCase):
"""Check conversion of HTML to odt"""
filename = resource_filename('cloudooo.handler.ooo.tests.data',
'test_failure_conversion.html')
data = open(filename, 'r').read()
with open(filename, 'rb') as f:
data = f.read()
status, response_dict, message = self.proxy.run_generate(filename,
encodestring(data),
encodebytes(data).decode(),
None,
'odt',
'text/html')
......
cloudooo/handler/ooo/tests/testOooMimemapper.py
View file @ d4abf3ff
......@@ -166,17 +166,17 @@ class TestMimeMapper(HandlerTestCase):
def testIfThereIsDuplicateData(self):
"""Test if there is duplicate data."""
# XXX It can not exists multiple keys inside a dictionary
extension_list = self.mimemapper._doc_type_list_by_extension.keys()
extension_list = list(self.mimemapper._doc_type_list_by_extension.keys())
self.assertEqual(len(extension_list), len(set(extension_list)))
for type_list in self.mimemapper._doc_type_list_by_extension.values():
for type_list in list(self.mimemapper._doc_type_list_by_extension.values()):
self.assertEqual(len(type_list), len(set(type_list)))
document_type_list = self.mimemapper._document_type_dict.keys()
document_type_list = list(self.mimemapper._document_type_dict.keys())
self.assertEqual(len(document_type_list), len(set(document_type_list)))
document_service_list = self.mimemapper._document_type_dict.values()
document_service_list = list(self.mimemapper._document_type_dict.values())
self.assertEqual(len(document_service_list), len(set(document_service_list)))
document_service_list = self.mimemapper._extension_list_by_type.keys()
document_service_list = list(self.mimemapper._extension_list_by_type.keys())
self.assertEqual(len(document_service_list), len(set(document_service_list)))
extension_list = self.mimemapper._extension_list_by_type.values()
extension_list = list(self.mimemapper._extension_list_by_type.values())
for extension in extension_list:
self.assertEqual(len(extension), len(set(extension)),
"extension_list_by_type has duplicate data")
......
cloudooo/handler/ooo/tests/testOooOdfDocument.py
View file @ d4abf3ff
......@@ -58,7 +58,7 @@ class TestOdfDocument(HandlerTestCase):
def testGetNotPresentFile(self):
"""Test if the getFile method returns None for not present file request"""
requested_file = self.oodocument.getFile('not_present.xml')
self.assertEqual(requested_file, '')
self.assertEqual(requested_file, b'')
def testParseContent(self):
"""Test if the _parsed_content attribute is the parsed content.xml"""
......
cloudooo/handler/ooo/tests/testOooServer.py
View file @ d4abf3ff
##############################################################################
# coding: utf-8
#
# Copyright (c) 2009-2010 Nexedi SA and Contributors. All Rights Reserved.
# Gabriel M. Monnerat <gabriel@tiolive.com>
......@@ -31,8 +30,8 @@
from os.path import join, exists
from os import remove
from base64 import encodestring, decodestring
from StringIO import StringIO
from base64 import encodebytes, decodebytes
from io import BytesIO
from lxml import etree
from zipfile import ZipFile, is_zipfile
from cloudooo.tests.cloudoooTestCase import TestCase
......@@ -199,8 +198,10 @@ class TestGetMetadata(TestCase):
def testupdateFileMetadataUpdateSomeMetadata(self):
"""Test server using method updateFileMetadata when the same metadata is
updated"""
odf_data = self.proxy.updateFileMetadata(encodestring(
open(join('data', 'testMetadata.odt')).read()),
with open(join('data', 'testMetadata.odt'), 'rb') as f:
data = f.read()
odf_data = self.proxy.updateFileMetadata(
encodebytes(data).decode(),
'odt',
dict(Reference="testSetMetadata", Something="ABC"))
new_odf_data = self.proxy.updateFileMetadata(odf_data,
......@@ -220,7 +221,7 @@ class TestGetMetadata(TestCase):
with open(join('data', 'test.odt'), 'rb') as f:
data = f.read()
generate_result = self.proxy.run_generate('test.odt',
encodestring(data),
encodebytes(data).decode(),
None, 'pdf',
'application/vnd.oasis.opendocument.text')
response_code, response_dict, response_message = generate_result
......@@ -239,7 +240,7 @@ class TestGenerate(TestCase):
with open(join('data', 'test.ods'), 'rb') as f:
data = f.read()
generate_result = self.proxy.run_generate('test.ods',
encodestring(data),
encodebytes(data).decode(),
None, 'html',
"application/vnd.oasis.opendocument.spreadsheet")
response_code, response_dict, response_message = generate_result
......@@ -249,7 +250,7 @@ class TestGenerate(TestCase):
self.assertEqual(response_dict['mime'], 'application/zip')
output_url = join(self.tmp_url, "zip.zip")
with open(output_url, 'wb') as f:
f.write(decodestring(response_dict['data']))
f.write(decodebytes(response_dict['data'].encode()))
self.assertTrue(is_zipfile(output_url))
filename_list = [file.filename for file in ZipFile(output_url).filelist]
for filename in filename_list:
......@@ -266,7 +267,7 @@ class TestGenerate(TestCase):
with open(join('data', 'test.ods'), 'rb') as f:
data = f.read()
generate_result = self.proxy.run_generate('test.ods',
encodestring(data),
encodebytes(data).decode(),
None, 'ms.xlsx',
"application/vnd.oasis.opendocument.spreadsheet")
response_code, response_dict, response_message = generate_result
......@@ -283,7 +284,7 @@ class TestGenerate(TestCase):
with open(join('data', 'test_png.odp'), 'rb') as f:
data = f.read()
generate_result = self.proxy.run_generate('test_png.odp',
encodestring(data),
encodebytes(data).decode(),
None, 'html',
'application/vnd.oasis.opendocument.presentation')
response_code, response_dict, response_message = generate_result
......@@ -293,19 +294,16 @@ class TestGenerate(TestCase):
self.assertEqual(response_dict['mime'], 'application/zip')
output_url = join(self.tmp_url, "zip.zip")
with open(output_url, 'wb') as f:
f.write(decodestring(response_dict['data']))
f.write(decodebytes(response_dict['data'].encode()))
self.assertTrue(is_zipfile(output_url))
zipfile = ZipFile(output_url)
try:
with ZipFile(output_url) as zipfile:
png_path = join(self.tmp_url, "img0.png")
zipfile.extractall(self.tmp_url)
with open(png_path, 'rb') as f:
content_type = self._getFileType(encodestring(f.read()))
content_type = self._getFileType(encodebytes(f.read()).decode())
self.assertEqual(content_type, 'image/png')
m = magic.Magic()
self.assertIn("8-bit/color RGB", m.from_file(png_path))
finally:
zipfile.close()
if exists(output_url):
remove(output_url)
......@@ -317,7 +315,7 @@ class TestGenerate(TestCase):
with open(join('data', 'test.odp'), 'rb') as f:
data = f.read()
generate_result = self.proxy.run_generate('test.odp',
encodestring(data),
encodebytes(data).decode(),
None, 'html',
'application/vnd.oasis.opendocument.presentation')
response_code, response_dict, response_message = generate_result
......@@ -327,7 +325,7 @@ class TestGenerate(TestCase):
self.assertEqual(response_dict['mime'], 'application/zip')
output_url = join(self.tmp_url, "zip.zip")
with open(output_url, 'wb') as f:
f.write(decodestring(response_dict['data'].encode()))
f.write(decodebytes(response_dict['data'].encode()))
self.assertTrue(is_zipfile(output_url))
filename_list = [file.filename for file in ZipFile(output_url).filelist]
for filename in filename_list:
......@@ -344,9 +342,9 @@ class TestGenerate(TestCase):
@expectedFailure
def testRunGenerateMethodFailResponse(self):
"""Test run_generate method with invalid document"""
data = open(join('data', 'test.odt'), 'r').read()[:100]
data = open(join('data', 'test.odt')).read()[:100]
generate_result = self.proxy.run_generate('test.odt',
encodestring(data),
encodebytes(data).decode(),
None, 'pdf', 'application/vnd.oasis.opendocument.text')
response_code, response_dict, response_message = generate_result
self.assertEqual(response_code, 402)
......@@ -361,7 +359,7 @@ class TestSetMetadata(TestCase):
with open(join('data', 'testMetadata.odt'), 'rb') as f:
data = f.read()
setmetadata_result = self.proxy.run_setmetadata('testMetadata.odt',
encodestring(data),
encodebytes(data).decode(),
{"Title": "testSetMetadata", "Description": "Music"})
response_code, response_dict, response_message = setmetadata_result
self.assertEqual(response_code, 200)
......@@ -391,7 +389,7 @@ class TestSetMetadata(TestCase):
with open(join('data', 'testMetadata.odt'), 'rb') as f:
data = f.read()[:100]
setmetadata_result = self.proxy.run_setmetadata('testMetadata.odt',
encodestring(data),
encodebytes(data).decode(),
{"Title": "testSetMetadata", "Description": "Music"})
response_code, response_dict, response_message = setmetadata_result
self.assertEqual(response_code, 402)
......@@ -423,7 +421,7 @@ class TestGetTableItemList(TestCase):
with open("data/granulate_table_test.odt", "rb") as f:
data = f.read()
granulated_table = self.proxy.getTableItemList(
encodestring(data),
encodebytes(data).decode(),
"odt")
self.assertEqual(table_list, granulated_table)
......@@ -435,18 +433,18 @@ class TestGetTableItemList(TestCase):
with open("data/granulate_table_test.doc", "rb") as f:
data = f.read()
granulated_table = self.proxy.getTableItemList(
encodestring(data),
encodebytes(data).decode(),
"doc")
self.assertEqual(table_list, granulated_table)
def testGetTableFromOdt(self):
"""Test if getTableItemList can get a item of some granulated table from odt file"""
with open("./data/granulate_table_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
granulated_table = self.proxy.getTableItemList(data, "odt")
table_item = decodestring(self.proxy.getTable(data, granulated_table[1][0],
"odt"))
content_xml_str = ZipFile(StringIO(table_item)).read('content.xml')
table_item = decodebytes(self.proxy.getTable(data, granulated_table[1][0],
"odt").encode())
content_xml_str = ZipFile(BytesIO(table_item)).read('content.xml')
content_xml = etree.fromstring(content_xml_str)
table_list = content_xml.xpath('//table:table',
namespaces=content_xml.nsmap)
......@@ -458,12 +456,12 @@ class TestGetTableItemList(TestCase):
def testGetTableFromDoc(self):
"""Test if getTableItemList can get a item of some granulated table from doc file"""
with open("./data/granulate_table_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
granulated_table = self.proxy.getTableItemList(data, "doc")
self.proxy.getTable(data, granulated_table[1][0], "doc")
table_item = decodestring(self.proxy.getTable(data, granulated_table[1][0],
"doc"))
content_xml_str = ZipFile(StringIO(table_item)).read('content.xml')
table_item = decodebytes(self.proxy.getTable(data, granulated_table[1][0],
"doc").encode())
content_xml_str = ZipFile(BytesIO(table_item)).read('content.xml')
content_xml = etree.fromstring(content_xml_str)
table_list = content_xml.xpath('//table:table',
namespaces=content_xml.nsmap)
......@@ -475,7 +473,7 @@ class TestGetTableItemList(TestCase):
def testGetColumnItemListFromOdt(self):
"""Test if getColumnItemList can get the list of column item from odt file"""
with open("./data/granulate_table_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
columns = self.proxy.getColumnItemList(
data,
"SoccerTeams",
......@@ -485,7 +483,7 @@ class TestGetTableItemList(TestCase):
def testGetColumnItemListFromDoc(self):
"""Test if getColumnItemList can get the list of column item from doc file"""
with open("./data/granulate_table_test.doc", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
#in the doc format the tables lose their names
columns = self.proxy.getColumnItemList(
data,
......@@ -496,7 +494,7 @@ class TestGetTableItemList(TestCase):
def testGetLineItemListFromOdt(self):
"""Test if getLineItemList can get the list of lines items from odt file"""
with open("./data/granulate_table_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
line_item_list = self.proxy.getLineItemList(data, "Developers", "odt")
self.assertEqual([['Name', 'Hugo'], ['Phone', '+55 (22) 8888-8888'],
['Email', 'hugomaia@tiolive.com'], ['Name', 'Rafael'],
......@@ -506,7 +504,7 @@ class TestGetTableItemList(TestCase):
def testGetLineItemListFromDoc(self):
"""Test if getLineItemList can get the list of lines items from doc file"""
with open("./data/granulate_table_test.doc", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
line_item_list = self.proxy.getLineItemList(data, "Table1", "doc")
self.assertEqual([['Name', 'Hugo'], ['Phone', '+55 (22) 8888-8888'],
['Email', 'hugomaia@tiolive.com'], ['Name', 'Rafael'],
......@@ -518,56 +516,60 @@ class TestImagetItemList(TestCase):
def testGetImageItemListFromOdt(self):
"""Test if getImageItemList can get the list of images items from odt file"""
with open("./data/granulate_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
image_list = self.proxy.getImageItemList(data, "odt")
self.assertEqual([['10000000000000C80000009CBF079A6E41EE290C.jpg', ''],
['10000201000000C80000004E85B3F70C71E07CE8.png', 'TioLive Logo'],
['10000201000000C80000004E85B3F70C71E07CE8.png', ''],
['2000004F0000423300001370ADF6545B2997B448.svm', 'Python Logo'],
['10000201000000C80000004E85B3F70C71E07CE8.png', 'Again TioLive Logo']],
self.assertEqual(
[['10000000000000C80000009C38276C51.jpg', ''],
['10000201000000C80000004E7B947D46.png', 'TioLive Logo'],
['10000201000000C80000004E7B947D46.png', ''],
['2000004F00004233000013707E7DE37A.svm', 'Python Logo'],
['10000201000000C80000004E7B947D46.png', 'Again TioLive Logo']],
image_list)
# the image filenames are the ones from the original document
with open("./data/granulate_test.odt", "rb") as f:
self.assertIn('Pictures/10000000000000C80000009C38276C51.jpg', ZipFile(f).namelist())
def testGetImageItemListFromDoc(self):
"""Test if getImageItemList can get the list of images items from doc file"""
with open("./data/granulate_test.doc", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
image_list = self.proxy.getImageItemList(data, "doc")
self.assertEqual([['10000000000000C80000009CBF079A6E41EE290C.jpg', ''],
['10000201000000C80000004E85B3F70C71E07CE8.png', 'TioLive Logo'],
['10000201000000C80000004E85B3F70C71E07CE8.png', ''],
['2000031600004233000013702113A0E70B910778.wmf', 'Python Logo'],
['10000201000000C80000004E85B3F70C71E07CE8.png', 'Again TioLive Logo']],
image_list)
# the doc has been converted to odt, so the image ids are
# not really predictable (they seem to depend on Libreoffice version)
self.assertIn('.jpg', image_list[0][0])
self.assertIn('TioLive Logo', [i[1] for i in image_list])
def testGetImageFromOdt(self):
"""Test if getImage can get a image from odt file after zip"""
with open("./data/granulate_test.odt", "rb") as f:
data = encodestring(f.read())
zip = ZipFile(StringIO(decodestring(data)))
image_id = '10000201000000C80000004E7B947D46.png'
original_image = zip.read('Pictures/%s' % image_id)
geted_image = decodestring(self.proxy.getImage(data, image_id, "odt"))
data = encodebytes(f.read()).decode()
with ZipFile('./data/granulate_test.odt') as zip:
original_image = zip.read('Pictures/10000201000000C80000004E7B947D46.png')
geted_image = decodebytes(self.proxy.getImage(data, '10000201000000C80000004E7B947D46.png', "odt").encode())
self.assertEqual(original_image, geted_image)
def testGetImageFromDoc(self):
"""Test if getImage can get a image from doc file after zip"""
with open("./data/granulate_test.doc", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
#This conversion is necessary to catch the image from the doc file;
#so compare with the server return.
data_odt = self.proxy.convertFile(data, 'doc', 'odt', False)
zip = ZipFile(StringIO(decodestring(data_odt)))
image_id = '10000000000000C80000009CBF079A6E41EE290C.jpg'
original_image = zip.read('Pictures/%s' % image_id)
geted_image = decodestring(self.proxy.getImage(data, image_id, "doc"))
zip = ZipFile(BytesIO(decodebytes(data_odt.encode())))
png_filename = [
name for name in zip.namelist()
if name.startswith('Pictures/')
and name.endswith('.png')][0]
image_id = png_filename[len('Pictures/'):]
original_image = zip.read(png_filename)
geted_image = decodebytes(self.proxy.getImage(data, image_id, "doc").encode())
self.assertEqual(original_image, geted_image)
class TestParagraphItemList(TestCase):
def testGetParagraphItemList(self):
"""Test if getParagraphItemList can get paragraphs correctly from document"""
with open("./data/granulate_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
paragraph_list = self.proxy.getParagraphItemList(data, "odt")
self.assertEqual([[0, 'P3'], [1, 'P1'], [2, 'P12'], [3, 'P6'], [4, 'P7'],
[5, 'P8'], [6, 'P6'], [7, 'P6'], [8, 'P13'], [9, 'P9'],
......@@ -580,7 +582,7 @@ class TestParagraphItemList(TestCase):
def testGetParagraphItem(self):
"""Test if getParagraph can get a paragraph"""
with open("./data/granulate_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
paragraph = self.proxy.getParagraph(data, 1, "odt")
self.assertEqual(['', 'P1'], paragraph)
......@@ -589,7 +591,7 @@ class TestChapterItemList(TestCase):
def testGetChapterItemList(self):
"""Test if getChapterItemList can get the chapters list correctly from document"""
with open("./data/granulate_chapters_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
chapter_list = self.proxy.getChapterItemList(data, "odt")
self.assertEqual([[0, 'Title 0'], [1, 'Title 1'], [2, 'Title 2'],
[3, 'Title 3'], [4, 'Title 4'], [5, 'Title 5'],
......@@ -599,7 +601,7 @@ class TestChapterItemList(TestCase):
def testGetChapterItem(self):
"""Test if getChapterItem can get a chapter"""
with open("./data/granulate_chapters_test.odt", "rb") as f:
data = encodestring(f.read())
data = encodebytes(f.read()).decode()
chapter = self.proxy.getChapterItem(1, data, "odt")
self.assertEqual(['Title 1', 1], chapter)
......@@ -611,43 +613,43 @@ class TestCSVEncoding(TestCase):
"""
def test_decode_ascii(self):
with open("./data/csv_ascii.csv", "rb") as f:
data = encodestring(f.read())
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = encodebytes(f.read()).decode()
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
["test", "1234"],
[x.text for x in tree.getroot().find('.//tr[1]').iterdescendants() if x.text])
def test_decode_utf8(self):
with open("./data/csv_utf8.csv", "rb") as f:
data = encodestring(f.read())
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = encodebytes(f.read()).decode()
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
[u"Jérome", u"ジェローム"],
["Jérome", "ジェローム"],
[x.text for x in tree.getroot().find('.//tr[1]').iterdescendants() if x.text])
self.assertEqual(
[u"नमस्ते", u"여보세요"],
["नमस्ते", "여보세요"],
[x.text for x in tree.getroot().find('.//tr[2]').iterdescendants() if x.text])
def test_decode_latin9(self):
with open("./data/csv_latin9.csv", "rb") as f:
data = encodestring(f.read())
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = encodebytes(f.read()).decode()
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
[u"Jérome", u"1€"],
["Jérome", "1€"],
[x.text for x in tree.getroot().find('.//tr[1]').iterdescendants() if x.text])
def test_separator_semicolon(self):
with open("./data/csv_semicolon.csv", "rb") as f:
data = encodestring(f.read())
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = encodebytes(f.read()).decode()
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
['a a', '1'],
[x.text for x in tree.getroot().find('.//tr[1]').iterdescendants() if x.text])
......@@ -657,10 +659,10 @@ class TestCSVEncoding(TestCase):
def test_separator_tab(self):
with open("./data/tsv.tsv", "rb") as f:
data = encodestring(f.read())
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = encodebytes(f.read()).decode()
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
['a', 'b'],
[x.text for x in tree.getroot().find('.//tr[1]').iterdescendants() if x.text])
......@@ -669,10 +671,10 @@ class TestCSVEncoding(TestCase):
[x.text for x in tree.getroot().find('.//tr[2]').iterdescendants() if x.text])
def test_empty_csv(self):
data = encodestring("")
converted = decodestring(self.proxy.convertFile(data, "csv", "html"))
data = ''
converted = decodebytes(self.proxy.convertFile(data, "csv", "html").encode())
parser = etree.HTMLParser()
tree = etree.parse(StringIO(converted), parser)
tree = etree.parse(BytesIO(converted), parser)
self.assertEqual(
[],
[x.text for x in tree.getroot().findall('.//td')])
cloudooo/handler/ooo/tests/testOooUnoConverter.py
View file @ d4abf3ff
......@@ -79,6 +79,7 @@ class TestUnoConverter(HandlerTestCase):
"--source_format=%s" % "odt",
"--mimemapper=%s" % mimemapper_pickled]
stdout, stderr = Popen(command,
text=True,
stdout=PIPE,
stderr=PIPE).communicate()
self.assertEqual(stderr, '')
......
cloudooo/handler/ooo/tests/testOooUnoMimemapper.py
View file @ d4abf3ff
......@@ -67,7 +67,8 @@ class TestUnoMimeMapper(HandlerTestCase):
"--port=%s" % self.openoffice_port]
stdout, stderr = Popen(command,
stdout=PIPE,
stderr=PIPE).communicate()
stderr=PIPE,
text=True).communicate()
self.assertEqual(stderr, '')
filter_dict, type_dict = json.loads(stdout)
self.assertNotEqual(filter_dict.get('writer8'), None)
......@@ -86,7 +87,8 @@ class TestUnoMimeMapper(HandlerTestCase):
"--port=%s" % self.openoffice_port]
stdout, stderr = Popen(command,
stdout=PIPE,
stderr=PIPE).communicate()
stderr=PIPE,
text=True).communicate()
self.assertEqual(stderr, '')
filter_dict, type_dict = json.loads(stdout)
self.assertNotEqual(filter_dict.get('writer8'), None)
......@@ -113,7 +115,8 @@ class TestUnoMimeMapper(HandlerTestCase):
for i in range(10):
stdout, stderr = Popen(command,
stdout=PIPE,
stderr=PIPE).communicate()
stderr=PIPE,
text=True).communicate()
self.assertEqual(stdout, '')
self.assertIn(error_msg, stderr)
openoffice.start()
......
cloudooo/handler/pdf/handler.py
View file @ d4abf3ff
......@@ -79,6 +79,7 @@ class Handler:
stdout=PIPE,
stderr=PIPE,
close_fds=True,
text=True,
env=self.environment).communicate()
info_list = [_f for _f in stdout.split("\n") if _f]
metadata = {}
......
cloudooo/handler/wkhtmltopdf/handler.py
View file @ d4abf3ff
......@@ -40,7 +40,7 @@ def keyNameToOption(key_name, prefix=""):
return "--" + prefix + key_name.replace("_", "-")
@implementer(IHandler)
class Handler(object):
class Handler:
"""ImageMagic Handler is used to handler images."""
def __init__(self, base_folder_url, data, source_format, **kw):
......@@ -272,10 +272,12 @@ class Handler(object):
return option_list
def makeDataPathArgumentOptionList(self, *args, **kw):
return self.makeDataUrlArgumentOptionList(*args, url_type="path", **kw)
kw['url_type'] = "path"
return self.makeDataUrlArgumentOptionList(*args, **kw)
def makeDataFileArgumentOptionList(self, *args, **kw):
return self.makeDataUrlArgumentOptionList(*args, url_type="file", **kw)
kw['url_type'] = "file"
return self.makeDataUrlArgumentOptionList(*args, **kw)
def makeRepeatableDataUrlArgumentOptionList(self, allowed_option_list,
option_dict, **kw):
......
cloudooo/handler/x2t/handler.py
View file @ d4abf3ff
......@@ -157,7 +157,7 @@ yformat_tuple = ("docy", "xlsy", "ppty")
@implementer(IHandler)
class Handler(object):
class Handler:
"""
X2T Handler is used to convert Microsoft Office 2007 documents to OnlyOffice
documents.
......@@ -198,7 +198,7 @@ class Handler(object):
output_data = None
if source_format in yformat_tuple:
if self._data.startswith("PK\x03\x04"):
if self._data.startswith(b"PK\x03\x04"):
os.mkdir(input_dir)
unzip(self.file.getUrl(), input_dir)
input_file_name = os.path.join(input_dir, "body.txt")
......@@ -227,7 +227,7 @@ class Handler(object):
root = ElementTree.Element('root')
for key, value in config.items():
ElementTree.SubElement(root, key).text = value
ElementTree.ElementTree(root).write(config_file, encoding='utf-8', xml_declaration=True,
ElementTree.ElementTree(root).write(config_file, encoding='unicode', xml_declaration=True,
default_namespace=None, method="xml")
# run convertion binary
......@@ -280,7 +280,7 @@ class Handler(object):
def getMetadata(self, base_document=False):
r"""Returns a dictionary with all metadata of document.
"""
if self._source_format in yformat_tuple and self._data.startswith("PK\x03\x04"):
if self._source_format in yformat_tuple and self._data.startswith(b"PK\x03\x04"):
if base_document:
openxml_format = yformat_map[self._source_format]
data = self.convert(yformat_map[self._source_format])
......@@ -305,7 +305,7 @@ class Handler(object):
"""
if metadata is None:
metadata = {}
if self._source_format in yformat_tuple and self._data.startswith("PK\x03\x04"):
if self._source_format in yformat_tuple and self._data.startswith(b"PK\x03\x04"):
root_dir = self.file.directory_name
output_file_name = os.path.join(root_dir, "tmp")
try:
......@@ -330,7 +330,7 @@ class Handler(object):
return OOoHandler(self.base_folder_url, self._data, self._source_format, **self._init_kw).setMetadata(metadata)
@staticmethod
def getAllowedConversionFormatList(source_mimetype):
def getAllowedConversionFormatList(source_mimetype:str):
"""Returns a list content_type and their titles which are supported
by enabled handlers.
......
cloudooo/handler/x2t/tests/testX2tHandler.py
View file @ d4abf3ff
......@@ -30,7 +30,7 @@
from zipfile import ZipFile
from cStringIO import StringIO
from io import BytesIO
from cloudooo.handler.x2t.handler import Handler
from cloudooo.tests.handlerTestCase import HandlerTestCase
......@@ -44,80 +44,94 @@ class TestHandler(HandlerTestCase):
def testConvertXlsx(self):
"""Test conversion of xlsx to xlsy and back"""
y_data = Handler(self.tmp_url, open("data/test.xlsx").read(), "xlsx", **self.kw).convert("xlsy")
y_body_data = ZipFile(StringIO(y_data)).open("body.txt").read()
self.assertTrue(y_body_data.startswith("XLSY;v10;0;"), "%r... does not start with 'XLSY;v10;0;'" % (y_body_data[:20],))
with open("data/test.xlsx", "rb") as f:
data = f.read()
y_data = Handler(self.tmp_url, data, "xlsx", **self.kw).convert("xlsy")
y_body_data = ZipFile(BytesIO(y_data)).open("body.txt").read()
self.assertTrue(y_body_data.startswith(b"XLSY;v10;0;"), "{!r}... does not start with 'XLSY;v10;0;'".format(y_body_data[:20]))
x_data = Handler(self.tmp_url, y_data, "xlsy", **self.kw).convert("xlsx")
# magic inspired by https://github.com/minad/mimemagic/pull/19/files
self.assertIn("xl/", x_data[:2000])
self.assertIn(b"xl/", x_data[:2000])
def testConvertXlsy(self):
"""Test conversion of xlsy to xlsx and back"""
x_data = Handler(self.tmp_url, open("data/test_body.xlsy").read(), "xlsy", **self.kw).convert("xlsx")
self.assertIn("xl/", x_data[:2000])
x_data = Handler(self.tmp_url, open("data/test.xlsy").read(), "xlsy", **self.kw).convert("xlsx")
self.assertIn("xl/", x_data[:2000])
with open("data/test_body.xlsy", "rb") as f:
data = f.read()
x_data = Handler(self.tmp_url, data, "xlsy", **self.kw).convert("xlsx")
self.assertIn(b"xl/", x_data[:2000])
with open("data/test.xlsy", "rb") as f:
data = f.read()
x_data = Handler(self.tmp_url, data, "xlsy", **self.kw).convert("xlsx")
self.assertIn(b"xl/", x_data[:2000])
y_data = Handler(self.tmp_url, x_data, "xlsx", **self.kw).convert("xlsy")
y_body_data = ZipFile(StringIO(y_data)).open("body.txt").read()
self.assertTrue(y_body_data.startswith("XLSY;v10;0;"), "%r... does not start with 'XLSY;v10;0;'" % (y_body_data[:20],))
y_body_data = ZipFile(BytesIO(y_data)).open("body.txt").read()
self.assertTrue(y_body_data.startswith(b"XLSY;v10;0;"), "{!r}... does not start with 'XLSY;v10;0;'".format(y_body_data[:20]))
def testConvertDocx(self):
"""Test conversion of docx to docy and back"""
y_data = Handler(self.tmp_url, open("data/test_with_image.docx").read(), "docx", **self.kw).convert("docy")
y_zip = ZipFile(StringIO(y_data))
with open("data/test_with_image.docx", "rb") as f:
data = f.read()
y_data = Handler(self.tmp_url, data, "docx", **self.kw).convert("docy")
y_zip = ZipFile(BytesIO(y_data))
y_body_data = y_zip.open("body.txt").read()
self.assertTrue(y_body_data.startswith("DOCY;v10;0;"), "%r... does not start with 'DOCY;v10;0;'" % (y_body_data[:20],))
self.assertTrue(y_body_data.startswith(b"DOCY;v10;0;"), "{!r}... does not start with 'DOCY;v10;0;'".format(y_body_data[:20]))
y_zip.open("media/image1.png")
x_data = Handler(self.tmp_url, y_data, "docy", **self.kw).convert("docx")
# magic inspired by https://github.com/minad/mimemagic/pull/19/files
self.assertIn("word/", x_data[:2000])
self.assertIn(b"word/", x_data[:2000])
def testConvertDocy(self):
"""Test conversion of docy to docx and back"""
x_data = Handler(self.tmp_url, open("data/test_with_image.docy").read(), "docy", **self.kw).convert("docx")
self.assertIn("word/", x_data[:2000])
with open("data/test_with_image.docy", "rb") as f:
data = f.read()
x_data = Handler(self.tmp_url, data, "docy", **self.kw).convert("docx")
self.assertIn(b"word/", x_data[:2000])
y_data = Handler(self.tmp_url, x_data, "docx", **self.kw).convert("docy")
y_zip = ZipFile(StringIO(y_data))
y_zip = ZipFile(BytesIO(y_data))
y_body_data = y_zip.open("body.txt").read()
self.assertTrue(y_body_data.startswith("DOCY;v10;0;"), "%r... does not start with 'DOCY;v10;0;'" % (y_body_data[:20],))
self.assertTrue(y_body_data.startswith(b"DOCY;v10;0;"), "{!r}... does not start with 'DOCY;v10;0;'".format(y_body_data[:20]))
y_zip.open("media/image1.png")
def testgetMetadata(self):
"""Test getMetadata from yformats"""
handler = Handler(self.tmp_url, "", "xlsy", **self.kw)
handler = Handler(self.tmp_url, b"", "xlsy", **self.kw)
self.assertEqual(handler.getMetadata(), {
u'CreationDate': u'00/00/0000 00:00:00',
u'ImplementationName': u'com.sun.star.comp.comphelper.OPropertyBag',
u'MIMEType': u'text/plain',
u'ModificationDate': u'00/00/0000 00:00:00',
u'PrintDate': u'00/00/0000 00:00:00',
u'TemplateDate': u'00/00/0000 00:00:00',
'CreationDate': '00/00/0000 00:00:00',
'ImplementationName': 'com.sun.star.comp.comphelper.OPropertyBag',
'MIMEType': 'text/plain',
'ModificationDate': '00/00/0000 00:00:00',
'PrintDate': '00/00/0000 00:00:00',
'TemplateDate': '00/00/0000 00:00:00',
})
handler = Handler(self.tmp_url, open("data/test_with_metadata.xlsy").read(), "xlsy", **self.kw)
with open("data/test_with_metadata.xlsy", "rb") as f:
data = f.read()
handler = Handler(self.tmp_url, data, "xlsy", **self.kw)
self.assertEqual(handler.getMetadata(), {
u'CreationDate': u'31/01/2018 21:09:10',
u'Keywords': [u'\u0442\u0435\u0441\u0442', u'\u0441\u0430\u0431\u0436\u0435\u043a\u0442'],
'CreationDate': '31/01/2018 21:09:10',
'Keywords': ['\u0442\u0435\u0441\u0442', '\u0441\u0430\u0431\u0436\u0435\u043a\u0442'],
'MIMEType': 'application/x-asc-spreadsheet',
u'ModificationDate': u'31/01/2018 21:22:36',
u'PrintDate': u'00/00/0000 00:00:00',
u'Subject': u'\u0432\u044b\u043a\u043b\u044e\u0447\u0438 \u0442\u0435\u043b\u0435\u0432\u0438\u0437\u043e\u0440',
u'TemplateDate': u'00/00/0000 00:00:00',
u'Title': u'kesha'})
'ModificationDate': '31/01/2018 21:22:36',
'PrintDate': '00/00/0000 00:00:00',
'Subject': '\u0432\u044b\u043a\u043b\u044e\u0447\u0438 \u0442\u0435\u043b\u0435\u0432\u0438\u0437\u043e\u0440',
'TemplateDate': '00/00/0000 00:00:00',
'Title': 'kesha'})
def testsetMetadata(self):
"""Test setMetadata for yformats"""
handler = Handler(self.tmp_url, open("data/test_with_metadata.xlsy").read(), "xlsy", **self.kw)
with open("data/test_with_metadata.xlsy", "rb") as f:
data = f.read()
handler = Handler(self.tmp_url, data, "xlsy", **self.kw)
new_mime_data = handler.setMetadata({
"Title": "test title",
"Subject": "test subject",
"Keywords": "test keywords",
})
handler = Handler(self.tmp_url, new_mime_data, "xlsy", **self.kw)
self.assertEqual(handler.getMetadata(), {u'Keywords': u'test keywords', 'MIMEType': 'application/x-asc-spreadsheet', u'Title': u'test title', u'Subject': u'test subject'})
self.assertEqual(handler.getMetadata(), {'Keywords': 'test keywords', 'MIMEType': 'application/x-asc-spreadsheet', 'Title': 'test title', 'Subject': 'test subject'})
def testGetAllowedConversionFormatList(self):
"""Test all combination of mimetype
......
cloudooo/handler/x2t/tests/testX2tServer.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@
##############################################################################
import io
import zipfile
from base64 import decodestring, encodestring
from base64 import decodebytes, encodebytes
from os.path import join
from cloudooo.tests.cloudoooTestCase import TestCase
......@@ -62,30 +62,30 @@ class TestServer(TestCase):
return scenario_list
def test_xlsx_to_xlsy(self):
with open(join('data', 'test.xlsx')) as f:
with open(join('data', 'test.xlsx'), 'rb') as f:
xlsx_data = f.read()
xlsy_data = self.proxy.convertFile(
encodestring(xlsx_data),
encodebytes(xlsx_data).decode(),
'xlsx',
'xlsy',
False
)
self.assertEqual(
sorted(zipfile.ZipFile(io.BytesIO(decodestring(xlsy_data))).namelist()),
sorted(zipfile.ZipFile(io.BytesIO(decodebytes(xlsy_data.encode()))).namelist()),
sorted(['Editor.xlsx', 'body.txt', 'metadata.json'])
)
def test_docx_to_docy(self):
with open(join('data', 'test_with_image.docx')) as f:
with open(join('data', 'test_with_image.docx'), 'rb') as f:
docx_data = f.read()
docy_data = self.proxy.convertFile(
encodestring(docx_data),
encodebytes(docx_data).decode(),
'docx',
'docy',
False
)
self.assertEqual(
sorted(zipfile.ZipFile(io.BytesIO(decodestring(docy_data))).namelist()),
sorted(zipfile.ZipFile(io.BytesIO(decodebytes(docy_data.encode()))).namelist()),
sorted(['body.txt', 'media/image1.png', 'metadata.json'])
)
cloudooo/manager.py
View file @ d4abf3ff
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2009-2010 Nexedi SA and Contributors. All Rights Reserved.
......@@ -30,10 +29,12 @@
##############################################################################
import mimetypes
import xmlrpc
from mimetypes import guess_type, guess_extension
from base64 import encodestring, decodestring
from binascii import a2b_base64
from base64 import encodebytes
from zope.interface import implementer
from interfaces.manager import IManager, IERP5Compatibility
from .interfaces.manager import IManager, IERP5Compatibility
from cloudooo.util import logger, parseContentType
from cloudooo.interfaces.granulate import ITableGranulator
from cloudooo.interfaces.granulate import IImageGranulator
......@@ -106,8 +107,8 @@ class Manager(object):
self.mimetype_registry = kw.pop("mimetype_registry")
self.handler_dict = kw.pop("handler_dict")
def convertFile(self, file, source_format, destination_format, zip=False,
refresh=False, conversion_kw={}):
def convertFile(self, file:str, source_format:str, destination_format:str, zip=False,
refresh=False, conversion_kw={}) -> str:
"""Returns the converted file in the given format.
Keywords arguments:
file -- File as string in base64
......@@ -136,43 +137,43 @@ class Manager(object):
self.mimetype_registry,
self.handler_dict)
handler = handler_class(self._path_tmp_dir,
decodestring(file),
a2b_base64(file),
source_format,
**kw)
decode_data = handler.convert(destination_format, **conversion_kw)
return encodestring(decode_data)
return encodebytes(decode_data).decode()
def updateFileMetadata(self, file, source_format, metadata_dict):
def updateFileMetadata(self, file:str, source_format:str, metadata_dict:dict) -> str:
"""Receives the string of document and a dict with metadatas. The metadata
is added in document.
e.g
self.updateFileMetadata(data = encodestring(data), metadata = \
self.updateFileMetadata(data = encodebytes(data), metadata = \
{"title":"abc","description":...})
return encodestring(document_with_metadata)
return encodebytes(document_with_metadata)
"""
handler_class = getHandlerClass(source_format,
None,
self.mimetype_registry,
self.handler_dict)
handler = handler_class(self._path_tmp_dir,
decodestring(file),
a2b_base64(file),
source_format,
**self.kw)
metadata_dict = dict([(key.capitalize(), value) \
for key, value in metadata_dict.iteritems()])
metadata_dict = {key.capitalize(): value \
for key, value in metadata_dict.items()}
decode_data = handler.setMetadata(metadata_dict)
return encodestring(decode_data)
return encodebytes(decode_data).decode()
def getFileMetadataItemList(self, file, source_format, base_document=False):
"""Receives the string of document as encodestring and returns a dict with
def getFileMetadataItemList(self, file:str, source_format:str, base_document=False) -> dict[str, str]:
"""Receives the string of document as encodebytes and returns a dict with
metadatas.
e.g.
self.getFileMetadataItemList(data = encodestring(data))
self.getFileMetadataItemList(data = encodebytes(data).decode())
return {'Title': 'abc','Description': 'comments', 'Data': None}
If converted_data is True, the ODF of data is added in dictionary.
e.g
self.getFileMetadataItemList(data = encodestring(data), True)
self.getFileMetadataItemList(data = encodebytes(data).decode(), True)
return {'Title': 'abc','Description': 'comments', 'Data': string_ODF}
Note that all keys of the dictionary have the first word in uppercase.
......@@ -182,11 +183,11 @@ class Manager(object):
self.mimetype_registry,
self.handler_dict)
handler = handler_class(self._path_tmp_dir,
decodestring(file),
a2b_base64(file),
source_format,
**self.kw)
metadata_dict = handler.getMetadata(base_document)
metadata_dict['Data'] = encodestring(metadata_dict.get('Data', ''))
metadata_dict['Data'] = encodebytes(metadata_dict.get('Data', b'')).decode()
return metadata_dict
def getAllowedExtensionList(self, request_dict={}):
......@@ -217,7 +218,7 @@ class Manager(object):
else:
return [('', '')]
def getAllowedConversionFormatList(self, source_mimetype):
def getAllowedConversionFormatList(self, source_mimetype:str) -> list:
r"""Returns a list content_type and their titles which are supported
by enabled handlers.
......@@ -354,7 +355,7 @@ class Manager(object):
logger.error('Error in generate from %s to %s', orig_format, extension, exc_info=True)
return (402, response_dict, str(e))
def getAllowedTargetItemList(self, content_type):
def getAllowedTargetItemList(self, content_type:str):
"""Wrapper getAllowedExtensionList but returns a dict.
This is a Backwards compatibility provided for ERP5 Project, in order to
keep compatibility with OpenOffice.org Daemon.
......@@ -373,25 +374,27 @@ class Manager(object):
logger.error('Error in getting target item list from %s', content_type, exc_info=True)
return (402, {}, e.args[0])
def _getOOGranulator(self, data, source_format="odt"):
def _getOOGranulator(self, data:str, source_format="odt"):
"""Returns an instance of the handler OOGranulator after convert the
data to 'odt'"""
data to 'odt'
data is a str with the actual data encoded in base64
"""
GRANULATABLE_FORMAT_LIST = ("odt",)
if source_format not in GRANULATABLE_FORMAT_LIST:
data = self.convertFile(data, source_format,
GRANULATABLE_FORMAT_LIST[0], zip=False)
return OOGranulator(decodestring(data), GRANULATABLE_FORMAT_LIST[0])
return OOGranulator(a2b_base64(data), GRANULATABLE_FORMAT_LIST[0])
def getTableItemList(self, data, source_format="odt"):
"""Returns the list of table IDs in the form of (id, title)."""
document = self._getOOGranulator(data, source_format)
return document.getTableItemList()
def getTable(self, data, id, source_format="odt"):
def getTable(self, data, id, source_format="odt") -> str:
"""Returns the table into a new 'format' file."""
document = self._getOOGranulator(data, source_format)
#the file will be convert; so, the source_format will be always 'odt'
return encodestring(document.getTable(id, 'odt'))
return encodebytes(document.getTable(id, 'odt')).decode()
def getColumnItemList(self, data, table_id, source_format):
"""Return the list of columns in the form of (id, title)."""
......@@ -403,17 +406,16 @@ class Manager(object):
document = self._getOOGranulator(data, source_format)
return document.getLineItemList(table_id)
def getImageItemList(self, data, source_format):
def getImageItemList(self, data:str, source_format:str):
"""Return the list of images in the form of (id, title)."""
data = self.convertFile(data, source_format, 'odt', zip=False)
document = self._getOOGranulator(data, source_format)
return document.getImageItemList()
def getImage(self, data, image_id, source_format, format=None,
resolution=None, **kw):
def getImage(self, data:str, image_id:str, source_format:str, format=None,
resolution=None, **kw) -> str:
"""Return the given image."""
document = self._getOOGranulator(data, source_format)
return encodestring(document.getImage(image_id, format, resolution, **kw))
return encodebytes(document.getImage(image_id, format, resolution, **kw)).decode()
def getParagraphItemList(self, data, source_format):
"""Returns the list of paragraphs in the form of (id, class) where class
......
cloudooo/paster_application.py
View file @ d4abf3ff
......@@ -51,7 +51,7 @@ def application(global_config, **local_config):
"""
prefix = 'env-'
environment_dict = {}
for parameter_name, value in local_config.iteritems():
for parameter_name, value in local_config.items():
if parameter_name.startswith(prefix):
value = value or ''
variable_name = parameter_name[len(prefix):]
......@@ -81,7 +81,7 @@ def application(global_config, **local_config):
mimetype_registry = local_config.get("mimetype_registry", "")
local_config["mimetype_registry"] = handler_mapping_list = \
filter(None, mimetype_registry.split("\n"))
[_f for _f in mimetype_registry.split("\n") if _f]
ooo_disable_filter_list = []
for filter_name in local_config.get("ooo_disable_filter_list", "").split("\n"):
......@@ -109,6 +109,6 @@ def application(global_config, **local_config):
handler_dict[handler] = module.Handler
local_config['handler_dict'] = handler_dict
from manager import Manager
from .manager import Manager
cloudooo_manager = Manager(cloudooo_path_tmp_dir, **local_config)
return WSGIXMLRPCApplication(instance=cloudooo_manager)
cloudooo/tests/cloudoooTestCase.py
View file @ d4abf3ff
......@@ -56,16 +56,20 @@ class TestCase(unittest.TestCase):
#create temporary path for some files
self.working_path = config.get("app:main", "working_path")
self.tmp_url = path.join(self.working_path, "tmp")
self.proxy = ServerProxy(("http://%s:%s/RPC2" % (self.hostname, self.port)),\
self.proxy = ServerProxy(("http://{}:{}/RPC2".format(self.hostname, self.port)),\
allow_none=True)
self.addCleanup(self.proxy('close'))
self.afterSetUp()
def afterSetUp(self):
"""Must be overwrite into subclass in case of need """
def _getFileType(self, output_data):
def _getFileType(self, output_data:str) -> str:
"""get file type of `output_data`
output_data is base64
"""
mime = Magic(mime=True)
mimetype = mime.from_buffer(decodestring(output_data))
mimetype = mime.from_buffer(decodebytes(output_data.encode()))
return mimetype
def _testConvertFile(self, input_url, source_format, destination_format,
......@@ -73,7 +77,10 @@ class TestCase(unittest.TestCase):
""" Generic test for converting file """
fault_list = []
try:
output_data = self.proxy.convertFile(encodestring(open(input_url).read()),
with open(input_url, 'rb') as f:
data = f.read()
output_data = self.proxy.convertFile(
encodebytes(data).decode(),
source_format,
destination_format,
zip)
......@@ -90,45 +97,49 @@ class TestCase(unittest.TestCase):
message = '\n'.join([template_message % fault for fault in fault_list])
self.fail('Failed Conversions:\n' + message)
def _testFaultConversion(self, data, source_format, destination_format):
def _testFaultConversion(self, data:bytes, source_format:str, destination_format:str):
""" Generic test for fail converting"""
self.assertRaises(Fault, self.proxy.convertFile, (data,
source_format,
destination_format))
def _testGetMetadata(self, input_url, source_format, expected_metadata,
def _testGetMetadata(self, input_url:str, source_format:str, expected_metadata:dict,
base_document=False):
""" Generic tes for getting metadata file"""
with open(input_url, 'rb') as f:
input_data = f.read()
metadata_dict = self.proxy.getFileMetadataItemList(
encodestring(open(input_url).read()),
encodebytes(input_data).decode(),
source_format,
base_document)
for key,value in expected_metadata.iteritems():
for key,value in expected_metadata.items():
self.assertEqual(metadata_dict[key], value)
def _testFaultGetMetadata(self, data, source_format):
def _testFaultGetMetadata(self, data:bytes, source_format:str):
""" Generic test for fail converting"""
self.assertRaises(Fault, self.proxy.getFileMetadataItemList, (data,
source_format))
def _testUpdateMetadata(self, input_url, source_format, metadata_dict):
def _testUpdateMetadata(self, input_url:str, source_format:str, metadata_dict:dict):
""" Generic test for setting metadata for file """
output_data = self.proxy.updateFileMetadata(encodestring(open(input_url).read()),
with open(input_url, 'rb') as f:
input_data = f.read()
output_data = self.proxy.updateFileMetadata(encodebytes(input_data).decode(),
source_format,
metadata_dict)
new_metadata_dict = self.proxy.getFileMetadataItemList(
output_data,
source_format,
False)
for key,value in metadata_dict.iteritems():
for key, value in metadata_dict.items():
self.assertEqual(new_metadata_dict[key], value)
def _testRunConvert(self, filename, data, expected_response_code,
response_dict_data,response_dict_keys,
def _testRunConvert(self, filename:str, data:bytes, expected_response_code:int,
response_dict_data, response_dict_keys:list[str],
expected_response_message, response_dict_meta=None):
"""Generic test for run_convert"""
response_code, response_dict, response_message = \
self.proxy.run_convert(filename, encodestring(data))
self.proxy.run_convert(filename, encodebytes(data).decode())
self.assertEqual(response_code, expected_response_code)
self.assertIsInstance(response_dict, dict)
if expected_response_code == 402:
......@@ -150,6 +161,7 @@ class TestCase(unittest.TestCase):
def runConversionList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testConvertFile(*scenario)
def GetMetadataScenarioList(self):
......@@ -161,6 +173,7 @@ class TestCase(unittest.TestCase):
def runGetMetadataList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testGetMetadata(*scenario)
def UpdateMetadataScenarioList(self):
......@@ -172,6 +185,7 @@ class TestCase(unittest.TestCase):
def runUpdateMetadataList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testUpdateMetadata(*scenario)
def FaultConversionScenarioList(self):
......@@ -183,6 +197,7 @@ class TestCase(unittest.TestCase):
def runFaultConversionList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testFaultConversion(*scenario)
def FaultGetMetadataScenarioList(self):
......@@ -194,6 +209,7 @@ class TestCase(unittest.TestCase):
def runFaultGetMetadataList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testFaultGetMetadata(*scenario)
def ConvertScenarioList(self):
......@@ -205,5 +221,6 @@ class TestCase(unittest.TestCase):
def runConvertScenarioList(self, scenarios):
for scenario in scenarios:
with self.subTest(scenario):
self._testRunConvert(*scenario)
cloudooo/tests/handlerTestCase.py
View file @ d4abf3ff
......@@ -32,7 +32,7 @@
import unittest
import sys
from os import environ, path, mkdir, putenv
from ConfigParser import ConfigParser
from configparser import ConfigParser
from cloudooo.handler.ooo.application.openoffice import openoffice
from cloudooo.handler.ooo.mimemapper import mimemapper
......@@ -42,7 +42,7 @@ config = ConfigParser()
def make_suite(test_case):
"""Function is used to run all tests together"""
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(test_case))
suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(test_case))
return suite
......@@ -84,8 +84,8 @@ def startFakeEnvironment(start_openoffice=True, conf_path=None):
if start_openoffice:
default_language = config.get('app:main',
'openoffice_user_interface_language', False,
{'openoffice_user_interface_language': 'en'})
'openoffice_user_interface_language', raw=False,
vars={'openoffice_user_interface_language': 'en'})
openoffice.loadSettings(hostname,
openoffice_port,
working_path,
......
cloudooo/tests/runHandlerUnitTest.py
View file @ d4abf3ff
......@@ -6,7 +6,7 @@ import logging
import unittest
from time import sleep
from subprocess import Popen
from ConfigParser import ConfigParser
from configparser import ConfigParser
from argparse import ArgumentParser
from os import chdir, path, environ, curdir, remove
from glob import glob
......
cloudooo/tests/runTestSuite.py
View file @ d4abf3ff
......@@ -39,7 +39,7 @@ from erp5.util.testsuite import SubprocessError
from erp5.util import taskdistribution
def _parsingErrorHandler(data, _):
print >> sys.stderr, 'Error parsing data:', repr(data)
print('Error parsing data:', repr(data), file=sys.stderr)
taskdistribution.patchRPCParser(_parsingErrorHandler)
class TestSuite(BaseTestSuite):
......@@ -115,7 +115,7 @@ def run():
if test_result is not None:
assert revision == test_result.revision, (revision, test_result.revision)
while suite.acquire():
test = test_result.start(suite.running.keys())
test = test_result.start(list(suite.running.keys()))
if test is not None:
suite.start(test.name, lambda status_dict, __test=test:
__test.stop(**status_dict))
......
cloudooo/util.py
View file @ d4abf3ff
......@@ -96,7 +96,7 @@ convertStringToBool = ('false', 'true').index
def zipTree(destination, *tree_path_list):
"""
destination may be a path or a StringIO
destination may be a path or a file-like
tree_path_list is a list that may contain a path or a couple(path, archive_root)
"""
......
cloudooo/wsgixmlrpcapplication.py
View file @ d4abf3ff
......@@ -13,7 +13,7 @@
# limitations under the License.
import logging
from SimpleXMLRPCServer import SimpleXMLRPCDispatcher
from xmlrpc.server import SimpleXMLRPCDispatcher
logger = logging.getLogger(__name__)
......@@ -22,7 +22,7 @@ class ErrorLoggingXMLRPCDispatcher(SimpleXMLRPCDispatcher):
"""
def _dispatch(self, method, params):
try:
return SimpleXMLRPCDispatcher._dispatch(self, method, params)
return super()._dispatch(method, params)
except:
logger.exception("Error calling %s", method)
raise
......@@ -48,7 +48,7 @@ class WSGIXMLRPCApplication:
return self.handle_POST(environ, start_response)
else:
start_response("400 Bad request", [('Content-Type', 'text/plain')])
return ['']
return [b'']
def handle_POST(self, environ, start_response):
"""Handles the HTTP POST request.
......@@ -74,7 +74,7 @@ class WSGIXMLRPCApplication:
response = self.dispatcher._marshaled_dispatch(
data, getattr(self.dispatcher, '_dispatch', None)
)
response += '\n'
response += b'\n'
except: # This should only happen if the module is buggy
# internal error, report as HTTP server error
logger.exception("Error serving request")
......
setup.py
View file @ d4abf3ff
......@@ -28,7 +28,7 @@ setup(name='cloudooo',
"Topic :: System :: Networking",
"Topic :: System :: Operating System Kernels :: Linux",
"Topic :: Internet :: WWW/HTTP :: WSGI",
"Programming Language :: Python :: 2.6",
"Programming Language :: Python :: 3",
"Natural Language :: English",
"Framework :: Paste"],
keywords='xmlrpc openoffice wsgi paste python',
......@@ -40,6 +40,7 @@ setup(name='cloudooo',
include_package_data=True,
zip_safe=False,
install_requires=install_requires,
python_requires='>=3.8',
entry_points="""
[paste.app_factory]
main = cloudooo.paster_application:application
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7