Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| import logging | |
| def console(soffice='soffice'): | |
| import uno | |
| import unokit.contexts | |
| import unokit.services | |
| import oxt_tool.remote | |
| with oxt_tool.remote.new_remote_context(soffice=soffice) as context: | |
| desktop = unokit.services.css.frame.Desktop() | |
| def new_textdoc(): | |
| return desktop.loadComponentFromURL('private:factory/swriter', | |
| '_blank', 0, tuple()) | |
| from unokit.util import dump, dumpdir | |
| local = dict(uno=uno, context=context, | |
| css=unokit.services.css, dump=dump, dumpdir=dumpdir, | |
| desktop=desktop, new_textdoc=new_textdoc) | |
| __import__('code').interact(banner='oxt-console', local=local) | |
| def dict_to_namedvalue(d): | |
| import uno | |
| nv = list() | |
| for n, v in d.items(): | |
| if isinstance(v, dict): | |
| v = dict_to_namedvalue(v) | |
| item = uno.createUnoStruct('com.sun.star.beans.NamedValue') | |
| item.Name = n | |
| item.Value = v | |
| nv.append(item) | |
| return tuple(nv) | |
| def test_remotely(soffice, discover_dirs, extra_path, logconf_path): | |
| import sys | |
| import os | |
| import os.path | |
| import discover | |
| import oxt_tool.remote | |
| logger = logging.getLogger('unokit') | |
| logger.addHandler(logging.StreamHandler()) | |
| logger.setLevel(logging.INFO) | |
| logfmt = logging.Formatter(('frontend %5d ' % os.getpid()) | |
| +'%(message)s') | |
| logchn = logging.StreamHandler() | |
| logchn.setFormatter(logfmt) | |
| logger = logging.getLogger('frontend') | |
| logger.addHandler(logchn) | |
| logger.setLevel(logging.INFO) | |
| working_dir = sys.argv[1] | |
| working_dir = os.path.abspath(working_dir) | |
| for path in sys.path: | |
| logger.info('sys.path: %s', path) | |
| if logconf_path: | |
| logconf_path = os.path.abspath(logconf_path) | |
| backend_path = sys.modules['oxt_tool'].__file__ | |
| backend_path = os.path.dirname(backend_path) | |
| backend_path = os.path.join(backend_path, 'backend.py') | |
| backend_name = 'backend.TestRunnerJob' | |
| tss = [] | |
| for d in discover_dirs: | |
| d = os.path.abspath(d) | |
| logger.info('discover tests: %s', d) | |
| testloader = discover.DiscoveringTestLoader() | |
| testsuite = testloader.discover(d) | |
| tss.append(testsuite) | |
| import unittest | |
| testsuite = unittest.TestSuite(tss) | |
| with oxt_tool.remote.new_remote_context(soffice=soffice) as context: | |
| logger.info('remote context created') | |
| factory = load_component(backend_path, backend_name) | |
| if factory: | |
| backendjob = factory.createInstanceWithContext(context) | |
| if backendjob: | |
| import cPickle | |
| from unokit.adapters import OutputStreamToFileLike | |
| pickled_testsuite = cPickle.dumps(testsuite) | |
| outputstream = OutputStreamToFileLike(sys.stderr) | |
| logstream = OutputStreamToFileLike(sys.stderr) | |
| args = dict(outputstream=outputstream, | |
| logstream=logstream, | |
| pickled_testsuite=pickled_testsuite, | |
| extra_path=tuple(extra_path), | |
| logconf_path=logconf_path, | |
| working_dir=working_dir) | |
| args = dict_to_namedvalue(args) | |
| result = backendjob.execute(args) | |
| result = str(result) | |
| result = cPickle.loads(result) | |
| return 0 if result['successful'] else 1 | |
| return -1 | |
| def load_component(component_path, component_name): | |
| import os | |
| import os.path | |
| import uno | |
| from unokit.services import css | |
| loader = css.loader.Python() | |
| if loader: | |
| component_path = os.path.abspath(component_path) | |
| component_url = uno.systemPathToFileUrl(component_path) | |
| return loader.activate(component_name, '', | |
| component_url, None) | |
| def console_in_proc(soffice='soffice'): | |
| import os | |
| import unohelper | |
| from com.sun.star.task import XJob | |
| import oxt_tool.remote | |
| logfmt = logging.Formatter(('frontend %5d ' % os.getpid()) | |
| +'%(message)s') | |
| logchn = logging.StreamHandler() | |
| logchn.setFormatter(logfmt) | |
| logger = logging.getLogger('frontend') | |
| logger.addHandler(logchn) | |
| logger.setLevel(logging.INFO) | |
| class ConsoleInput(unohelper.Base, XJob): | |
| def __init__(self, context): | |
| self.context = context | |
| def execute(self, arguments): | |
| prompt, = arguments | |
| try: | |
| return raw_input(prompt.Value) | |
| except EOFError: | |
| return None | |
| import sys | |
| import os.path | |
| backend_path = sys.modules['oxt_tool'].__file__ | |
| backend_path = os.path.dirname(backend_path) | |
| backend_path = os.path.join(backend_path, 'backend.py') | |
| backend_name = 'backend.ConsoleJob' | |
| with oxt_tool.remote.new_remote_context(soffice=soffice) as context: | |
| logger.info('remote context created') | |
| factory = load_component(backend_path, backend_name) | |
| if factory: | |
| backendjob = factory.createInstanceWithContext(context) | |
| if backendjob: | |
| from unokit.adapters import OutputStreamToFileLike | |
| outstream = OutputStreamToFileLike(sys.stderr) | |
| args = dict(inp=ConsoleInput(context), | |
| outstream=outstream) | |
| args = dict_to_namedvalue(args) | |
| backendjob.execute(args) | |
| class LoEnviron(object): | |
| def __init__(self, program_dir): | |
| self.program_dir = program_dir | |
| def rc_ext(self): | |
| import sys | |
| if sys.platform == 'win32': | |
| return '.ini' | |
| else: | |
| return 'rc' | |
| def fundamentalrc(self): | |
| import os.path | |
| filename = 'fundamental' + self.rc_ext | |
| return os.path.join(self.program_dir, filename) | |
| def ure_bootstrap(self): | |
| return 'vnd.sun.star.pathname:' + self.fundamentalrc | |
| def run_in_lo(soffice='soffice'): | |
| import os | |
| import sys | |
| import os.path | |
| uno_pythonpath = os.environ['UNO_PYTHONPATH'].split(os.pathsep) | |
| sys.path.extend(uno_pythonpath) | |
| loenv = LoEnviron(os.environ['LO_PROGRAM']) | |
| os.environ['URE_BOOTSTRAP'] = loenv.ure_bootstrap | |
| import oxt_tool.remote | |
| logger = logging.getLogger('unokit') | |
| logger.addHandler(logging.StreamHandler()) | |
| logger.setLevel(logging.INFO) | |
| logger = logging.getLogger('oxt_tool') | |
| logger.addHandler(logging.StreamHandler()) | |
| logger.setLevel(logging.INFO) | |
| logfmt = logging.Formatter(('frontend %5d ' % os.getpid()) | |
| +'%(message)s') | |
| logchn = logging.StreamHandler() | |
| logchn.setFormatter(logfmt) | |
| logger = logging.getLogger('frontend') | |
| logger.addHandler(logchn) | |
| logger.setLevel(logging.INFO) | |
| for path in sys.path: | |
| logger.info('sys.path: %s', path) | |
| working_dir = os.getcwd() | |
| working_dir = os.path.abspath(working_dir) | |
| argv = list(sys.argv[1:]) | |
| if argv[0].startswith('--logfile='): | |
| logfile = argv[0][len('--logfile='):] | |
| argv = argv[1:] | |
| else: | |
| logfile = None | |
| argv[0] = os.path.abspath(argv[0]) | |
| print argv | |
| backend_path = sys.modules['oxt_tool'].__file__ | |
| backend_path = os.path.dirname(backend_path) | |
| backend_path = os.path.join(backend_path, 'backend.py') | |
| backend_name = 'backend.RemoteRunJob' | |
| with oxt_tool.remote.new_remote_context(soffice=soffice) as context: | |
| logger.info('remote context created') | |
| factory = load_component(backend_path, backend_name) | |
| if factory: | |
| backendjob = factory.createInstanceWithContext(context) | |
| if backendjob: | |
| import cPickle | |
| from unokit.adapters import OutputStreamToFileLike | |
| from unokit.adapters import InputStreamFromFileLike | |
| stdin = InputStreamFromFileLike(sys.stdin) | |
| stdout = OutputStreamToFileLike(sys.stdout) | |
| stderr = OutputStreamToFileLike(sys.stderr) | |
| path = cPickle.dumps(sys.path) | |
| argv = cPickle.dumps(argv) | |
| args = dict(logfile=logfile, | |
| working_dir=working_dir, | |
| path=path, | |
| argv=argv, | |
| stdin=stdin, | |
| stdout=stdout, | |
| stderr=stderr) | |
| args = dict_to_namedvalue(args) | |
| return backendjob.execute(args) | |
| return -1 | |
| def install(unopkg='unopkg'): | |
| import sys | |
| cmd = [unopkg, 'add', '-v', '-s'] | |
| cmd.extend(sys.argv[1:]) | |
| cmd = [('"%s"' % x) if ' ' not in x else x | |
| for x in cmd] | |
| cmd = ' '.join(cmd) | |
| import os | |
| return os.system(cmd) | |
| class Console(object): | |
| def __init__(self, buildout, name, options): | |
| import os.path | |
| self.__name = name | |
| self.__logger = logging.getLogger(name) | |
| self.__bindir = buildout['buildout']['bin-directory'] | |
| soffice = options.get('soffice', 'soffice').strip() | |
| if not os.path.exists(soffice): | |
| self.__skip = True | |
| self.__logger.info('soffice not found at: %s', soffice) | |
| self.__logger.info('installation will be skipped') | |
| return | |
| in_proc = options.get('in_proc', 'true').strip().lower() | |
| in_proc = in_proc in ['true', 'yes', '1'] | |
| self.__python = options.get('python', '').strip() | |
| self.__soffice = soffice | |
| self.__in_proc = in_proc | |
| self.__skip = False | |
| def install(self): | |
| if self.__skip: | |
| self.__logger.info('skipped') | |
| return [] | |
| from zc.buildout import easy_install | |
| import pkg_resources | |
| import sys | |
| ws = [pkg_resources.get_distribution(dist) | |
| for dist in ['unokit', 'oxt.tool']] | |
| if self.__in_proc: | |
| func = 'console_in_proc' | |
| else: | |
| func = 'console' | |
| entrypoints = [(self.__name, 'oxt_tool', func)] | |
| arguments = '%r' % self.__soffice | |
| if self.__python: | |
| python = self.__python | |
| else: | |
| python = sys.executable | |
| return easy_install.scripts(entrypoints, | |
| ws, python, self.__bindir, | |
| arguments=arguments) | |
| update = install | |
| class TestRunner(object): | |
| def __init__(self, buildout, name, options): | |
| import os.path | |
| self.__name = name | |
| self.__logger = logging.getLogger(name) | |
| self.__bindir = buildout['buildout']['bin-directory'] | |
| self.__skip = False | |
| self.__python = options.get('python', '').strip() | |
| self.__soffice = options.get('soffice', 'soffice').strip() | |
| if not os.path.exists(self.__soffice): | |
| self.__skip = True | |
| self.__logger.info('soffice not found at: %s', self.__soffice) | |
| self.__logger.info('installation will be skipped') | |
| return | |
| self.__discover = options['discover'].split() | |
| self.__extra_path = options['extra_path'].split() | |
| self.__logconf_path = options.get('logconf_path') | |
| def install(self): | |
| if self.__skip: | |
| self.__logger.info('skipped') | |
| return [] | |
| from zc.buildout import easy_install | |
| import pkg_resources | |
| import sys | |
| ws = [pkg_resources.get_distribution(dist) | |
| for dist in ['unokit', 'oxt.tool', 'discover']] | |
| entrypoints = [(self.__name, 'oxt_tool', 'test_remotely')] | |
| arguments = '%r, %r, %r, %r' | |
| arguments = arguments % (self.__soffice, | |
| self.__discover, | |
| self.__extra_path, | |
| self.__logconf_path) | |
| if self.__python: | |
| python = self.__python | |
| else: | |
| python = sys.executable | |
| return easy_install.scripts(entrypoints, | |
| ws, python, self.__bindir, | |
| arguments=arguments) | |
| update = install | |
| class Installer(object): | |
| def __init__(self, buildout, name, options): | |
| import os.path | |
| self.__name = name | |
| self.__logger = logging.getLogger(name) | |
| self.__bindir = buildout['buildout']['bin-directory'] | |
| unopkg = options.get('unopkg', 'unopkg').strip() | |
| if not os.path.exists(unopkg): | |
| self.__skip = True | |
| self.__logger.info('unopkg not found at: %s', unopkg) | |
| self.__logger.info('installation will be skipped') | |
| return | |
| self.__unopkg = unopkg | |
| self.__skip = False | |
| def install(self): | |
| if self.__skip: | |
| self.__logger.info('skipped') | |
| return [] | |
| from zc.buildout import easy_install | |
| import pkg_resources | |
| import sys | |
| ws = [pkg_resources.get_distribution(dist) | |
| for dist in ['unokit', 'oxt.tool']] | |
| entrypoints = [(self.__name, 'oxt_tool', 'install')] | |
| arguments = '%r' % self.__unopkg | |
| return easy_install.scripts(entrypoints, | |
| ws, sys.executable, self.__bindir, | |
| arguments=arguments) | |
| update = install | |