Exitcode skips

HISTORY.rst

@@ -12,6 +12,9 @@ version 1.7.0-dev
 + When the ``--git-aware`` flag is used a submodule check is performed in order
   to assert that all submodules are properly checked out. This prevents
   unclear copying errors.
++ Tests are now skipped if the workflow does not exit with the desired exit
+  code, except for the exit code tests, to reduce visual clutter when reporting
+  failing tests.
 + Tests for checking file content are now skipped when the file does not exist
   in order to reduce visual clutter when reporting failing tests.
 + Test and support for Python 3.11.

src/pytest_workflow/content_tests.py

@@ -244,6 +244,9 @@ def runtest(self):
         were we are looking for multiple words (variants / sequences). """
         # Wait for thread to complete.
         self.parent.thread.join()
+        if not self.parent.workflow.matching_exitcode():
+            pytest.skip(f"'{self.parent.workflow.name}' did not exit with"
+                        f"desired exit code.")
         if self.parent.file_not_found:
             pytest.skip(f"'{self.content_name}' was not found so cannot be "
                         f"searched")

src/pytest_workflow/file_tests.py

@@ -102,6 +102,9 @@ def runtest(self):
         # Wait for the workflow process to finish before checking if the file
         # exists.
         self.workflow.wait()
+        if not self.workflow.matching_exitcode():
+            pytest.skip(f"'{self.parent.workflow.name}' did not exit with"
+                        f"desired exit code.")
         assert self.file.exists() == self.should_exist
 
     def repr_failure(self, excinfo, style=None):
@@ -134,6 +137,9 @@ def __init__(self, parent: pytest.Collector, filepath: Path,
     def runtest(self):
         # Wait for the workflow to finish before we check the md5sum of a file.
         self.workflow.wait()
+        if not self.workflow.matching_exitcode():
+            pytest.skip(f"'{self.parent.workflow.name}' did not exit with"
+                        f"desired exit code.")
         self.observed_md5sum = file_md5sum(self.filepath)
         assert self.observed_md5sum == self.expected_md5sum
 

src/pytest_workflow/plugin.py

@@ -406,7 +406,8 @@ def queue_workflow(self):
         # Create a workflow and make sure it runs in the tempdir
         workflow = Workflow(command=self.workflow_test.command,
                             cwd=tempdir,
-                            name=self.workflow_test.name)
+                            name=self.workflow_test.name,
+                            desired_exit_code=self.workflow_test.exit_code)
 
         # Add the workflow to the workflow queue.
         self.config.workflow_queue.put(workflow)
@@ -446,17 +447,16 @@ def collect(self):
         # Below structure makes it easy to append tests
         tests = []
 
-        tests += [
-            FileTestCollector.from_parent(
-                parent=self, filetest=filetest, workflow=workflow)
-            for filetest in self.workflow_test.files]
-
         tests += [ExitCodeTest.from_parent(
             parent=self,
-            desired_exit_code=self.workflow_test.exit_code,
             workflow=workflow,
             stderr_bytes=self.config.getoption("stderr_bytes"))]
 
+        tests += [
+            FileTestCollector.from_parent(
+                parent=self, filetest=filetest, workflow=workflow)
+            for filetest in self.workflow_test.files]
+
         tests += [ContentTestCollector.from_parent(
             name="stdout", parent=self,
             filepath=workflow.stdout_file,
@@ -476,17 +476,15 @@ def collect(self):
 
 class ExitCodeTest(pytest.Item):
     def __init__(self, parent: pytest.Collector,
-                 desired_exit_code: int,
                  workflow: Workflow, stderr_bytes: int):
-        name = f"exit code should be {desired_exit_code}"
+        name = f"exit code should be {workflow.desired_exit_code}"
         super().__init__(name, parent=parent)
         self.stderr_bytes = stderr_bytes
         self.workflow = workflow
-        self.desired_exit_code = desired_exit_code
 
     def runtest(self):
         # workflow.exit_code waits for workflow to finish.
-        assert self.workflow.exit_code == self.desired_exit_code
+        assert self.workflow.matching_exitcode()
 
     def repr_failure(self, excinfo, style=None):
         standerr = self.workflow.stderr_file
@@ -499,7 +497,7 @@ def repr_failure(self, excinfo, style=None):
                 standout_file.seek(-self.stderr_bytes, os.SEEK_END)
             message = (f"'{self.workflow.name}' exited with exit code " +
                        f"'{self.workflow.exit_code}' instead of "
-                       f"'{self.desired_exit_code}'.\nstderr: "
+                       f"'{self.workflow.desired_exit_code}'.\nstderr: "
                        f"{standerr_file.read().strip().decode('utf-8')}"
                        f"\nstdout: "
                        f"{standout_file.read().strip().decode('utf-8')}")

src/pytest_workflow/workflow.py

@@ -35,7 +35,8 @@ class Workflow(object):
     def __init__(self,
                  command: str,
                  cwd: Optional[Path] = None,
-                 name: Optional[str] = None):
+                 name: Optional[str] = None,
+                 desired_exit_code: int = 0):
         """
         Initiates a workflow object
         :param command: The string that represents the command to be run
@@ -69,6 +70,7 @@ def __init__(self,
         self._started = False
         self.errors: List[Exception] = []
         self.start_lock = threading.Lock()
+        self.desired_exit_code = desired_exit_code
 
     def start(self):
         """Runs the workflow in a subprocess in the background.
@@ -137,6 +139,12 @@ def wait(self, timeout_secs: Optional[float] = None,
             # workflow
             pass
 
+    def matching_exitcode(self) -> bool:
+        """Checks if the workflow exited with the desired exit code"""
+        # This is done in the workflow object to reduce redundancy in the rest
+        # of the code.
+        return self.exit_code == self.desired_exit_code
+
     @property
     def stdout(self) -> bytes:
         self.wait()

tests/test_fail_messages.py

@@ -94,17 +94,17 @@
      "it should not be there"),
     ("""\
     - name: fail_test
-      command: grep
+      command: bash -c 'echo "" >&2'
       stderr:
         contains:
           - "No arguments?"
     """,
      "'No arguments?' was not found in 'fail_test': stderr "
      "while it should be there"),
     ("""\
-    - name: fail_test
-      command: grep
-      stderr:
+   - name: fail_test
+     command: bash -c 'echo "grep --help" >&2'
+     stderr:
         must_not_contain:
           - "grep --help"
     """,

tests/test_skip_tests.py

@@ -28,8 +28,9 @@
         - "halloumi"
       must_not_contain:
         - "gorgonzola"
-
-- name: wall3_test
+""")
+SKIP_TESTS2 = textwrap.dedent("""
+- name: wall2_test
   command: bash -c "echo 'testing' > test3.txt"
   files:
     - path: test3.txt
@@ -38,9 +39,32 @@
       must_not_contain:
         - "testing"
 """)
+SKIP_TESTS3 = textwrap.dedent("""
+- name: wall3_test
+  command: bash -c "exit 12"
+  files:
+    - path: test4.txt
+      contains:
+        - "content"
+      must_not_contain:
+        - "no_content"
+      md5sum: e583af1f8b00b53cda87ae9ead880224
+""")
 
 
 def test_skips(pytester):
     pytester.makefile(".yml", test=SKIP_TESTS)
     result = pytester.runpytest("-v").parseoutcomes()
-    assert {"failed": 4, "passed": 3, "skipped": 3} == result
+    assert {"failed": 2, "passed": 1, "skipped": 3} == result
+
+
+def test_skips2(pytester):
+    pytester.makefile(".yml", test=SKIP_TESTS2)
+    result = pytester.runpytest("-v").parseoutcomes()
+    assert {"failed": 2, "passed": 2} == result
+
+
+def test_skips3(pytester):
+    pytester.makefile(".yml", test=SKIP_TESTS3)
+    result = pytester.runpytest("-v").parseoutcomes()
+    assert {"failed": 1, "skipped": 4} == result

tests/test_workflow.py

@@ -96,3 +96,12 @@ def test_workflow_name():
 def test_workflow_name_inferred():
     workflow = Workflow("echo moo")
     assert workflow.name == "echo"
+
+
+def test_workflow_matching_exit_code():
+    workflow = Workflow("echo moo")
+    workflow.run()
+    assert workflow.matching_exitcode()
+    workflow2 = Workflow("grep", desired_exit_code=2)
+    workflow2.run()
+    assert workflow2.matching_exitcode()