gunittest: handle process manually since pygrass would raise an exception (introduced in https://trac.osgeo.org/grass/changeset/61229, alternative would be to use try-except, SimpleModule doctest is failing), clean up some module calls (use SimpleModule), improve CalledModuleError creation in pygrass

git-svn-id: https://svn.osgeo.org/grass/grass/trunk@61231 15284696-431f-4ddb-bdfa-cd5b030d7da7

lib/python/gunittest/case.py

@@ -14,13 +14,14 @@ for details.
 
 import os
 import subprocess
+import time
 import unittest
 from unittest.util import safe_repr
 
 from grass.pygrass.modules import Module
 from grass.exceptions import CalledModuleError
 
-from .gmodules import call_module
+from .gmodules import call_module, SimpleModule
 from .checkers import (check_text_ellipsis,
                        text_to_keyvalue, keyvalue_equals, diff_keyvalue,
                        file_md5, files_equal_md5)
@@ -444,25 +445,19 @@ class TestCase(unittest.TestCase):
         :raises CalledModuleError: if the module failed
         """
         module = _module_from_parameters(module, **kwargs)
+        _check_module_run_parameters(module)
 
-        if module.run_:
-            raise ValueError('Do not run the module manually, set run_=False')
-        if not module.finish_:
-            raise ValueError('This function will always finish module run,'
-                             ' set finish_=None or finish_=True.')
-        # we expect most of the usages with stdout=PIPE
-        # TODO: in any case capture PIPE always?
-        if module.stdout_ is None:
-            module.stdout_ = subprocess.PIPE
-        elif module.stdout_ != subprocess.PIPE:
-            raise ValueError('stdout_ can be only PIPE or None')
-        if module.stderr_ is None:
-            module.stderr_ = subprocess.PIPE
-        elif module.stderr_ != subprocess.PIPE:
-            raise ValueError('stderr_ can be only PIPE or None')
-            # because we want to capture it
+        # do what module.run() with finish_=True would do
+        start = time.time()
         module.run()
-        if module.popen.returncode:
+        stdout, stderr = module.popen.communicate(input=module.stdin)
+        module.outputs['stdout'].value = stdout if stdout else ''
+        module.outputs['stderr'].value = stderr if stderr else ''
+        module.time = time.time() - start
+        if module.popen.poll():
+            # here exception raised by run() with finish_=True would be
+            # almost enough but we want some additional info to be included
+            # in the test report
             errors = module.outputs['stderr'].value
             # provide diagnostic at least in English locale
             # TODO: standardized error code would be handy here
@@ -507,28 +502,19 @@ class TestCase(unittest.TestCase):
         non-zero return code.
         """
         module = _module_from_parameters(module, **kwargs)
+        _check_module_run_parameters(module)
 
-        # TODO: merge stderr to stdout? if caller gives PIPE, for sure not
-        if module.run_:
-            raise ValueError('Do not run the module manually, set run_=False')
-        if not module.finish_:
-            raise ValueError('This function will always finish module run,'
-                             ' set finish_=None or finish_=True.')
-        if module.stdout_ is None:
-            module.stdout_ = subprocess.PIPE
-        elif module.stdout_ != subprocess.PIPE:
-            raise ValueError('stdout_ can be only PIPE or None')
-            # because we want to capture it
-        if module.stderr_ is None:
-            module.stderr_ = subprocess.PIPE
-        elif module.stderr_ != subprocess.PIPE:
-            raise ValueError('stderr_ can be only PIPE or None')
-            # because we want to capture it
-
+        # do what module.run() with finish_=True would do
+        start = time.time()
         module.run()
+        stdout, stderr = module.popen.communicate(input=module.stdin)
+        module.outputs['stdout'].value = stdout if stdout else ''
+        module.outputs['stderr'].value = stderr if stderr else ''
+        module.time = time.time() - start
+        # TODO: these two lines should go to report in some better way
         print module.outputs['stdout'].value
         print module.outputs['stderr'].value
-        if module.popen.returncode:
+        if module.popen.poll():
             # TODO: message format
             # TODO: stderr?
             stdmsg = ('Running <{m.name}> module ended'
@@ -554,27 +540,19 @@ class TestCase(unittest.TestCase):
         Works like `assertModule()` but expects module to fail.
         """
         module = _module_from_parameters(module, **kwargs)
+        _check_module_run_parameters(module)
 
-        if module.run_:
-            raise ValueError('Do not run the module manually, set run_=False')
-        if not module.finish_:
-            raise ValueError('This function will always finish module run,'
-                             ' set finish_=None or finish_=True.')
-        if module.stdout_ is None:
-            module.stdout_ = subprocess.PIPE
-        elif module.stdout_ != subprocess.PIPE:
-            raise ValueError('stdout_ can be only PIPE or None')
-            # because we want to capture it
-        if module.stderr_ is None:
-            module.stderr_ = subprocess.PIPE
-        elif module.stderr_ != subprocess.PIPE:
-            raise ValueError('stderr_ can be only PIPE or None')
-            # because we want to capture it
-
+        # do what module.run() with finish_=True would do
+        start = time.time()
         module.run()
+        stdout, stderr = module.popen.communicate(input=module.stdin)
+        module.outputs['stdout'].value = stdout if stdout else ''
+        module.outputs['stderr'].value = stderr if stderr else ''
+        module.time = time.time() - start
+        # TODO: these two lines should go to report in some better way
         print module.outputs['stdout'].value
         print module.outputs['stderr'].value
-        if not module.popen.returncode:
+        if not module.popen.poll():
             stdmsg = ('Running <%s> ended with zero (successful) return code'
                       ' when expecting module to fail' % module.get_python())
             self.fail(self._formatMessage(msg, stdmsg))
@@ -592,5 +570,25 @@ def _module_from_parameters(module, **kwargs):
             # allow to pass all parameters in one dictionary called parameters
         if kwargs.keys() == ['parameters']:
             kwargs = kwargs['parameters']
-        module = Module(module, run_=False, **kwargs)
+        module = SimpleModule(module, **kwargs)
     return module
+
+
+def _check_module_run_parameters(module):
+    # in this case module already run and we would start it again
+    if module.run_:
+        raise ValueError('Do not run the module manually, set run_=False')
+    if module.finish_:
+        raise ValueError('This function will always finish module run,'
+                         ' set finish_=None or finish_=False.')
+    # we expect most of the usages with stdout=PIPE
+    # TODO: in any case capture PIPE always?
+    if module.stdout_ is None:
+        module.stdout_ = subprocess.PIPE
+    elif module.stdout_ != subprocess.PIPE:
+        raise ValueError('stdout_ can be only PIPE or None')
+    if module.stderr_ is None:
+        module.stderr_ = subprocess.PIPE
+    elif module.stderr_ != subprocess.PIPE:
+        raise ValueError('stderr_ can be only PIPE or None')
+        # because we want to capture it

lib/python/gunittest/gmodules.py

@@ -50,7 +50,7 @@ class SimpleModule(Module):
                                  ', it would be overriden' % banned)
         kargs['stdout_'] = subprocess.PIPE
         kargs['stderr_'] = subprocess.PIPE
-        kargs['finish_'] = True
+        kargs['finish_'] = False
         kargs['run_'] = False
 
         Module.__init__(self, cmd, *args, **kargs)

lib/python/gunittest/testsuite/test_assertions.py

@@ -91,7 +91,7 @@ class TestAssertCommandKeyValue(grass.gunittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.use_temp_region()
-        cls.runModule(Module('g.region', rast='elevation', run_=False))
+        cls.runModule(SimpleModule('g.region', rast='elevation'))
 
     @classmethod
     def tearDownClass(cls):
@@ -99,7 +99,8 @@ class TestAssertCommandKeyValue(grass.gunittest.TestCase):
 
     def test_pygrass_module(self):
         """Test syntax with Module as module"""
-        module = Module('r.info', map='elevation', flags='gr', run_=False)
+        module = Module('r.info', map='elevation', flags='gr',
+                        run_=False, finish_=False)
         self.assertCommandKeyValue(module,
                                    reference=dict(min=55.58, max=156.33),
                                    precision=0.01, sep='=')
@@ -132,7 +133,7 @@ class TestRasterMapAssertations(grass.gunittest.TestCase):
     def setUpClass(cls):
         cls.use_temp_region()
         # TODO: here we should actually not call self.runModule but call_module
-        cls.runModule(Module('g.region', rast='elevation', run_=False))
+        cls.runModule(SimpleModule('g.region', rast='elevation'))
 
     @classmethod
     def tearDownClass(cls):

lib/python/gunittest/testsuite/test_module_assertions.py

@@ -4,32 +4,38 @@ import copy
 import subprocess
 
 from grass.pygrass.modules import Module
+from grass.gunittest.gmodules import SimpleModule
 
 import grass.gunittest
 from grass.gunittest.gmodules import CalledModuleError
 
 
 class TestModuleAssertions(grass.gunittest.TestCase):
+    """Test assertions using PyGRASS Module"""
     # pylint: disable=R0904
 
     def setUp(self):
+        """Create two Module instances one correct and one with wrong map"""
         self.rinfo = Module('r.info', map='elevation', flags='g',
-                       stdout_=subprocess.PIPE, run_=False)
+                            stdout_=subprocess.PIPE, run_=False, finish_=False)
         self.rinfo_wrong = copy.deepcopy(self.rinfo)
         self.wrong_map = 'does_not_exists'
         self.rinfo_wrong.inputs['map'].value = self.wrong_map
 
     def test_runModule(self):
+        """Module used in runModule()"""
         self.runModule(self.rinfo)
         self.assertTrue(self.rinfo.outputs['stdout'].value)
         self.assertRaises(CalledModuleError, self.runModule, self.rinfo_wrong)
 
     def test_assertModule(self):
+        """Module used in assertModule()"""
         self.assertModule(self.rinfo)
         self.assertTrue(self.rinfo.outputs['stdout'].value)
         self.assertRaises(self.failureException, self.assertModule, self.rinfo_wrong)
 
     def test_assertModuleFail(self):
+        """Module used in assertModuleFail()"""
         self.assertModuleFail(self.rinfo_wrong)
         stderr = self.rinfo_wrong.outputs['stderr'].value
         self.assertTrue(stderr)
@@ -37,5 +43,39 @@ class TestModuleAssertions(grass.gunittest.TestCase):
         self.assertRaises(self.failureException, self.assertModuleFail, self.rinfo)
 
 
+class TestSimpleModuleAssertions(grass.gunittest.TestCase):
+    """Test assertions using SimpleModule"""
+    # pylint: disable=R0904
+
+    def setUp(self):
+        """Create two SimpleModule instances one correct and one with wrong map
+        """
+        self.rinfo = SimpleModule('r.info', map='elevation', flags='g')
+        self.rinfo_wrong = copy.deepcopy(self.rinfo)
+        self.wrong_map = 'does_not_exists'
+        self.rinfo_wrong.inputs['map'].value = self.wrong_map
+
+    def test_runModule(self):
+        """SimpleModule used in runModule()"""
+        self.runModule(self.rinfo)
+        self.assertTrue(self.rinfo.outputs['stdout'].value)
+        self.assertRaises(CalledModuleError, self.runModule, self.rinfo_wrong)
+
+    def test_assertModule(self):
+        """SimpleModule used in assertModule()"""
+        self.assertModule(self.rinfo)
+        self.assertTrue(self.rinfo.outputs['stdout'].value)
+        self.assertRaises(self.failureException, self.assertModule, self.rinfo_wrong)
+
+    def test_assertModuleFail(self):
+        """SimpleModule used in assertModuleFail()"""
+        self.assertModuleFail(self.rinfo_wrong)
+        stderr = self.rinfo_wrong.outputs['stderr'].value
+        self.assertTrue(stderr)
+        self.assertIn(self.wrong_map, stderr)
+        self.assertRaises(self.failureException, self.assertModuleFail, self.rinfo)
+
+
+
 if __name__ == '__main__':
     grass.gunittest.test()

lib/python/pygrass/modules/interface/module.py

@@ -564,8 +564,9 @@ class Module(object):
             self.outputs['stderr'].value = stderr if stderr else ''
             self.time = time.time() - start
             if self.popen.poll():
-                raise CalledModuleError(self.popen.returncode, self.get_bash(),
-                                        {}, stderr)
+                raise CalledModuleError(returncode=self.popen.returncode,
+                                        code=self.get_bash(),
+                                        module=self.name, errors=stderr)
         return self
 
 ###############################################################################