Repository: Quansight-Labs/pytest-run-parallel (language: python)

Issue:
Probably parallel execution does not work with unittest
I'm trying this tool to run tests using multi-threaded context:
Here is the code I using for a check of multi-threaded execution:
```python
import unittest


def test_failed1():
    import threading

    print("test_failed1+ident:", threading.current_thread().ident)
    assert False


class TestABC(unittest.TestCase):

    def test_failed2(self):
        import threading

        print("TestABC+test_failed2+ident:", threading.current_thread().ident)
        assert False
```

and execute it with the following command:
```bash
pytest --parallel-threads=3 tests/test_abc.py -v
```
and I see the following output:
```
=============================================================================================== FAILURES ===============================================================================================
_____________________________________________________________________________________________ test_failed1 _____________________________________________________________________________________________

    def test_failed1():
        import threading
    
        print("test_failed1+ident:", threading.current_thread().ident)
>       assert False
E       assert False

tests/test_abc.py:8: AssertionError
----------------------------------------------------------------------------------------- Captured stdout call -----------------------------------------------------------------------------------------
test_failed1+ident:test_failed1+ident:test_failed1+ident:  139805692208704139805700601408
 
139805708994112
_________________________________________________________________________________________ TestABC.test_failed2 _________________________________________________________________________________________

self = <tests.test_abc.TestABC testMethod=test_failed2>

    def test_failed2(self):
        import threading
    
        print("TestABC+test_failed2+ident:", threading.current_thread().ident)
>       assert False
E       assert False

tests/test_abc.py:18: AssertionError
----------------------------------------------------------------------------------------- Captured stdout call -----------------------------------------------------------------------------------------
TestABC+test_failed2+ident: 139806255169536
======================================================================================= short test summary info ========================================================================================
FAILED tests/test_abc.py::test_failed1 - assert False
FAILED tests/test_abc.py::TestABC::test_failed2 - assert False
========================================================================================== 2 failed in 0.29s ===========================================================================================
```

We can see that `test_failed1` printed 3 times `test_failed1+ident` and thread ident but the `TestABC::test_failed2` printed current thread ident only once, so I supposed that it ran the test in one thread only and not 3 threads as required in the command.

@andfoy can you please check this? Thanks!



Discussion / hints:
@vfdev-5 thanks for opening this issue, simultaneous printing under free threaded Python leads to stdout to not respect printing output order, if you detail the output trace that you shared, you can see that all the output is being concatenated:

```
139805692208704139805700601408 139805708994112
```

This is something that's occurring without this plugin. From what I've experienced, this is the current behavior in CPython.
With respect to the second test executions, it seems like a legit issue. Let me check how pytest collection works

Fix patch:
diff --git a/.gitignore b/.gitignore
index 9b6398d..45a4626 100644
--- a/.gitignore
+++ b/.gitignore
@@ -50,6 +50,7 @@ htmlcov/
 nosetests.xml
 coverage.xml
 *.cover
+*.lcov
 *.py,cover
 .hypothesis/
 .pytest_cache/


Test patch:
diff --git a/src/pytest_run_parallel/plugin.py b/src/pytest_run_parallel/plugin.py
index 3acbac7..2830a4a 100644
--- a/src/pytest_run_parallel/plugin.py
+++ b/src/pytest_run_parallel/plugin.py
@@ -2,6 +2,8 @@
 import threading
 import functools
 
+from _pytest.outcomes import Skipped, Failed
+
 
 def pytest_addoption(parser):
     group = parser.getgroup('run-parallel')
@@ -22,26 +24,27 @@ def pytest_configure(config):
         'using `n` threads.')
 
 
-@pytest.hookimpl(trylast=True)
-def pytest_generate_tests(metafunc):
-    n_workers = metafunc.config.option.parallel_threads
-    m = metafunc.definition.get_closest_marker('parallel_threads')
-    if m is not None:
-        n_workers = int(m.args[0])
-    setattr(metafunc.function, '_n_workers', n_workers)
-
-
 def wrap_function_parallel(fn, n_workers=10):
     barrier = threading.Barrier(n_workers)
     @functools.wraps(fn)
     def inner(*args, **kwargs):
         errors = []
+        skip = None
+        failed = None
         def closure(*args, **kwargs):
             barrier.wait()
             try:
                 fn(*args, **kwargs)
+            except Warning as w:
+                pass
             except Exception as e:
                 errors.append(e)
+            except Skipped as s:
+                nonlocal skip
+                skip = s.msg
+            except Failed as f:
+                nonlocal failed
+                failed = f
 
         workers = []
         for _ in range(0, n_workers):
@@ -56,14 +59,21 @@ def closure(*args, **kwargs):
         for worker in workers:
             worker.join()
 
-        if len(errors) > 0:
+        if skip is not None:
+            pytest.skip(skip)
+        elif failed is not None:
+            raise failed
+        elif len(errors) > 0:
             raise errors[0]
+
     return inner
 
 
-@pytest.hookimpl(wrapper=True)
-def pytest_pyfunc_call(pyfuncitem):
-    n_workers = getattr(pyfuncitem.obj, '_n_workers', None)
+@pytest.hookimpl(trylast=True)
+def pytest_itemcollected(item):
+    n_workers = item.config.option.parallel_threads
+    m = item.get_closest_marker('parallel_threads')
+    if m is not None:
+        n_workers = int(m.args[0])
     if n_workers is not None and n_workers > 1:
-        pyfuncitem.obj = wrap_function_parallel(pyfuncitem.obj, n_workers)
-    return (yield)
+        item.obj = wrap_function_parallel(item.obj, n_workers)
diff --git a/tests/test_run_parallel.py b/tests/test_run_parallel.py
index 8221fd1..a530d81 100644
--- a/tests/test_run_parallel.py
+++ b/tests/test_run_parallel.py
@@ -84,7 +84,7 @@ def test_check_thread_count(counter):
 
         @pytest.mark.order(2)
         @pytest.mark.parallel_threads(1)
-        def test_check_thread_count(counter2):
+        def test_check_thread_count2(counter2):
             assert counter2._count == 5
     """)
 
@@ -97,6 +97,65 @@ def test_check_thread_count(counter2):
     # fnmatch_lines does an assertion internally
     result.stdout.fnmatch_lines([
         '*::test_check_thread_count PASSED*',
+        '*::test_check_thread_count2 PASSED*',
+    ])
+
+    # make sure that we get a '0' exit code for the testsuite
+    assert result.ret == 0
+
+
+def test_unittest_compat(pytester):
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+        import unittest
+        from threading import Lock
+
+        class Counter:
+            def __init__(self):
+                self._count = 0
+                self._lock = Lock()
+
+            def increase(self):
+                with self._lock:
+                    self._count += 1
+
+        class TestExample(unittest.TestCase):
+            @classmethod
+            def setUpClass(cls):
+                cls.counter = Counter()
+                cls.counter2 = Counter()
+
+            @pytest.mark.order(1)
+            def test_example_1(self):
+                self.counter.increase()
+
+            @pytest.mark.order(1)
+            @pytest.mark.parallel_threads(5)
+            def test_example_2(self):
+                self.counter2.increase()
+
+            @pytest.mark.order(2)
+            @pytest.mark.parallel_threads(1)
+            def test_check_thread_count(self):
+                assert self.counter._count == 10
+
+            @pytest.mark.order(2)
+            @pytest.mark.parallel_threads(1)
+            def test_check_thread_count2(self):
+                assert self.counter2._count == 5
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest(
+        '--parallel-threads=10',
+        '-v'
+    )
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines([
+        '*::test_check_thread_count PASSED*',
+        '*::test_check_thread_count2 PASSED*',
     ])
 
     # make sure that we get a '0' exit code for the testsuite
@@ -114,3 +173,80 @@ def test_help_message(pytester):
         # '             Set the number of threads used to execute each test concurrently.',
     ])
 
+
+def test_skip(pytester):
+    """Make sure that pytest accepts our fixture."""
+
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+
+        def test_skipped():
+            pytest.skip('Skip propagation')
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest(
+        '--parallel-threads=10',
+        '-v'
+    )
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines([
+        '*::test_skipped SKIPPED*',
+    ])
+
+    # make sure that we get a '0' exit code for the testsuite
+    assert result.ret == 0
+
+
+def test_fail(pytester):
+    """Make sure that pytest accepts our fixture."""
+
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+
+        def test_should_fail():
+            pytest.fail()
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest(
+        '--parallel-threads=10',
+        '-v'
+    )
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines([
+        '*::test_should_fail FAILED*',
+    ])
+
+    # make sure that we get a '0' exit code for the testsuite
+    assert result.ret != 0
+
+
+def test_exception(pytester):
+    """Make sure that pytest accepts our fixture."""
+
+    # create a temporary pytest test module
+    pytester.makepyfile("""
+        import pytest
+
+        def test_should_fail():
+            raise ValueError('Should raise')
+    """)
+
+    # run pytest with the following cmd args
+    result = pytester.runpytest(
+        '--parallel-threads=10',
+        '-v'
+    )
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines([
+        '*::test_should_fail FAILED*',
+    ])
+
+    # make sure that we get a '0' exit code for the testsuite
+    assert result.ret != 0