Merge pull request #160 from dmitry-lipetsk/D20241207_001--remote_op-expect_error

RemoteOperations::exec_command must not raise an exception when 'expect_error' is True (#159)

Author: dmitry-lipetsk

testgres/operations/remote_ops.py

@@ -83,26 +83,26 @@ def exec_command(self, cmd, wait_exit=False, verbose=False, expect_error=False,
 
         exit_status = process.returncode
 
-        if encoding:
-            result = result.decode(encoding)
-            error = error.decode(encoding)
-
-        if expect_error:
-            raise Exception(result, error)
+        assert type(result) == bytes  # noqa: E721
+        assert type(error) == bytes  # noqa: E721
 
         if not error:
-            error_found = 0
+            error_found = False
         else:
-            error = normalize_error(error)
             error_found = exit_status != 0 or any(
-                marker in error for marker in ['error', 'Permission denied', 'fatal', 'No such file or directory']
+                marker in error for marker in [b'error', b'Permission denied', b'fatal', b'No such file or directory']
             )
 
-        if not ignore_errors and error_found:
-            if isinstance(error, bytes):
-                message = b"Utility exited with non-zero code. Error: " + error
-            else:
-                message = f"Utility exited with non-zero code. Error: {error}"
+        assert type(error_found) == bool  # noqa: E721
+
+        if encoding:
+            result = result.decode(encoding)
+            error = error.decode(encoding)
+
+        if not ignore_errors and error_found and not expect_error:
+            error = normalize_error(error)
+            assert type(error) == str  # noqa: E721
+            message = "Utility exited with non-zero code. Error: " + error
             raise ExecUtilException(message=message, command=cmd, exit_code=exit_status, out=result)
 
         if verbose:

tests/__init__.py

@@ -0,0 +1,11 @@
+import pytest
+import platform
+
+
+class RunConditions:
+    # It is not a test kit!
+    __test__ = False
+
+    def skip_if_windows():
+        if platform.system().lower() == "windows":
+            pytest.skip("This test does not support Windows.")

tests/helpers/__init__.py

@@ -0,0 +1,54 @@
+import pytest
+
+from testgres import ExecUtilException
+from testgres import LocalOperations
+
+from .helpers.run_conditions import RunConditions
+
+
+class TestLocalOperations:
+
+    @pytest.fixture(scope="function", autouse=True)
+    def setup(self):
+        self.operations = LocalOperations()
+
+    def test_exec_command_success(self):
+        """
+        Test exec_command for successful command execution.
+        """
+        RunConditions.skip_if_windows()
+
+        cmd = "python3 --version"
+        response = self.operations.exec_command(cmd, wait_exit=True, shell=True)
+
+        assert b'Python 3.' in response
+
+    def test_exec_command_failure(self):
+        """
+        Test exec_command for command execution failure.
+        """
+        RunConditions.skip_if_windows()
+
+        cmd = "nonexistent_command"
+        while True:
+            try:
+                self.operations.exec_command(cmd, wait_exit=True, shell=True)
+            except ExecUtilException as e:
+                error = e.message
+                break
+            raise Exception("We wait an exception!")
+        assert error == "Utility exited with non-zero code. Error `b'/bin/sh: 1: nonexistent_command: not found\\n'`"
+
+    def test_exec_command_failure__expect_error(self):
+        """
+        Test exec_command for command execution failure.
+        """
+        RunConditions.skip_if_windows()
+
+        cmd = "nonexistent_command"
+
+        exit_status, result, error = self.operations.exec_command(cmd, verbose=True, wait_exit=True, shell=True, expect_error=True)
+
+        assert error == b'/bin/sh: 1: nonexistent_command: not found\n'
+        assert exit_status == 127
+        assert result == b''

tests/helpers/run_conditions.py

@@ -30,12 +30,27 @@ def test_exec_command_failure(self):
         Test exec_command for command execution failure.
         """
         cmd = "nonexistent_command"
-        try:
-            exit_status, result, error = self.operations.exec_command(cmd, verbose=True, wait_exit=True)
-        except ExecUtilException as e:
-            error = e.message
+        while True:
+            try:
+                self.operations.exec_command(cmd, verbose=True, wait_exit=True)
+            except ExecUtilException as e:
+                error = e.message
+                break
+            raise Exception("We wait an exception!")
         assert error == b'Utility exited with non-zero code. Error: bash: line 1: nonexistent_command: command not found\n'
 
+    def test_exec_command_failure__expect_error(self):
+        """
+        Test exec_command for command execution failure.
+        """
+        cmd = "nonexistent_command"
+
+        exit_status, result, error = self.operations.exec_command(cmd, verbose=True, wait_exit=True, shell=True, expect_error=True)
+
+        assert error == b'bash: line 1: nonexistent_command: command not found\n'
+        assert exit_status == 127
+        assert result == b''
+
     def test_is_executable_true(self):
         """
         Test is_executable for an existing executable.