| | |
| | 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 |
| |
|
| | @property |
| | def rc_ext(self): |
| | import sys |
| | if sys.platform == 'win32': |
| | return '.ini' |
| | else: |
| | return 'rc' |
| |
|
| |
|
| | @property |
| | def fundamentalrc(self): |
| | import os.path |
| | filename = 'fundamental' + self.rc_ext |
| | return os.path.join(self.program_dir, filename) |
| |
|
| |
|
| | @property |
| | 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 |
| |
|
