| |
| 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 |
|
|
