Source code for gunittest.gmodules
# -*- coding: utf-8 -*-
"""Specialized interfaces for invoking modules for testing framework
Copyright (C) 2014 by the GRASS Development Team
This program is free software under the GNU General Public
License (>=v2). Read the file COPYING that comes with GRASS GIS
for details.
:authors: Vaclav Petras, Soeren Gebbert
"""
import subprocess
from grass.script.core import start_command
from grass.script.utils import encode, decode
from grass.exceptions import CalledModuleError
from grass.pygrass.modules import Module
from .utils import do_doctest_gettext_workaround
[docs]class SimpleModule(Module):
"""Simple wrapper around pygrass.modules.Module to make sure that
run\_, finish\_, stdout and stderr are set correctly.
>>> mapcalc = SimpleModule('r.mapcalc', expression='test_a = 1',
... overwrite=True)
>>> mapcalc.run()
Module('r.mapcalc')
>>> mapcalc.returncode
0
>>> colors = SimpleModule('r.colors',
... map='test_a', rules='-', stdin_='1 red')
>>> colors.run()
Module('r.colors')
>>> colors.returncode
0
>>> str(colors.inputs.stdin)
'1 red'
>>> str(colors.outputs.stdout)
''
>>> colors.outputs.stderr.strip()
"Color table for raster map <test_a> set to 'rules'"
"""
def __init__(self, cmd, *args, **kargs):
for banned in ['stdout_', 'stderr_', 'finish_', 'run_']:
if banned in kargs:
raise ValueError('Do not set %s parameter'
', it would be overriden' % banned)
kargs['stdout_'] = subprocess.PIPE
kargs['stderr_'] = subprocess.PIPE
kargs['finish_'] = True
kargs['run_'] = False
Module.__init__(self, cmd, *args, **kargs)
[docs]def call_module(module, stdin=None,
merge_stderr=False, capture_stdout=True, capture_stderr=True,
**kwargs):
r"""Run module with parameters given in `kwargs` and return its output.
>>> print (call_module('g.region', flags='pg')) # doctest: +ELLIPSIS
projection=...
zone=...
n=...
s=...
w=...
>>> call_module('m.proj', flags='i', input='-', stdin="50.0 41.5")
'8642890.65|6965155.61|0.00\n'
>>> call_module('g.region', aabbbccc='notexist') # doctest: +IGNORE_EXCEPTION_DETAIL
Traceback (most recent call last):
...
CalledModuleError: Module run g.region ... ended with error
If `stdin` is not set and `kwargs` contains ``input`` with value set
to ``-`` (dash), the function raises an error.
Note that ``input`` nor ``output`` parameters are used by this
function itself, these are usually module parameters which this
function just passes to it. However, when ``input`` is in parameters
the function checks if its values is correct considering value of
``stdin`` parameter.
:param str module: module name
:param stdin: string to be used as module standard input (stdin) or `None`
:param merge_stderr: if the standard error output should be merged with stdout
:param kwargs: module parameters
:returns: module standard output (stdout) as string or None if apture_stdout is False
:raises CalledModuleError: if module return code is non-zero
:raises ValueError: if the parameters are not correct
.. note::
The data read is buffered in memory, so do not use this method
if the data size is large or unlimited.
"""
# TODO: remove this:
do_doctest_gettext_workaround()
# implementation inspired by subprocess.check_output() function
if stdin:
if 'input' in kwargs and kwargs['input'] != '-':
raise ValueError(_("input='-' must be used when stdin is specified"))
if stdin == subprocess.PIPE:
raise ValueError(_("stdin must be string or buffer, not PIPE"))
kwargs['stdin'] = subprocess.PIPE # to be able to send data to stdin
elif 'input' in kwargs and kwargs['input'] == '-':
raise ValueError(_("stdin must be used when input='-'"))
if merge_stderr and not (capture_stdout and capture_stderr):
raise ValueError(_("You cannot merge stdout and stderr and not capture them"))
if 'stdout' in kwargs:
raise TypeError(_("stdout argument not allowed, it could be overridden"))
if 'stderr' in kwargs:
raise TypeError(_("stderr argument not allowed, it could be overridden"))
if capture_stdout:
kwargs['stdout'] = subprocess.PIPE
if capture_stderr:
if merge_stderr:
kwargs['stderr'] = subprocess.STDOUT
else:
kwargs['stderr'] = subprocess.PIPE
process = start_command(module, **kwargs)
# input=None means no stdin (our default)
# for no stdout, output is None which is out interface
# for stderr=STDOUT or no stderr, errors is None
# which is fine for CalledModuleError
output, errors = process.communicate(input=encode(decode(stdin)) if stdin else None)
returncode = process.poll()
if returncode:
raise CalledModuleError(returncode, module, kwargs, errors)
return decode(output) if output else None