Skip to content

Commit

Permalink
Improved multiprocessing test to terminate the pool and process, and …
Browse files Browse the repository at this point in the history
…to fix issue with old pytest and python.
  • Loading branch information
Sylvain MARIE committed Jan 7, 2022
1 parent 71c2d6b commit 795130c
Show file tree
Hide file tree
Showing 2 changed files with 55 additions and 34 deletions.
24 changes: 20 additions & 4 deletions tests/cases/issues/test_issue_242.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,22 @@
# + All contributors to <https://github.com/smarie/python-pyfields>
#
# License: 3-clause BSD, <https://github.com/smarie/python-pyfields/blob/master/LICENSE>
import pytest
from distutils.version import LooseVersion

import sys

from pytest_cases import parametrize_with_cases
from multiprocessing import Pool, Process

from functools import partial


PYTEST_VERSION = LooseVersion(pytest.__version__)
PYTEST3_OR_GREATER = PYTEST_VERSION >= LooseVersion('3.0.0')
PY3 = sys.version_info >= (3,)


class TestCases:
def case_A(self):
return 2, 4
Expand All @@ -30,11 +40,17 @@ def test_f_xy(x, y):
p = Process(target=partial(f), args=(x, y, True))
p.start()
p.join()
p.terminate()

# in a pool
pool = Pool(processes=2)
pool.starmap(partial(f), [(x, y, False), (x, y, True)])
if PY3:
# in a pool
pool = Pool(processes=2)
pool.starmap(partial(f), [(x, y, False), (x, y, True)])
pool.terminate()


def test_synthesis(module_results_dct):
assert list(module_results_dct) == ["test_f_xy[A]", "test_f_xy[B]"]
if PYTEST3_OR_GREATER:
assert list(module_results_dct) == ["test_f_xy[A]", "test_f_xy[B]"]
else:
assert list(module_results_dct) == ['test_f_xy[A[0]-A[1]]', 'test_f_xy[B[0]-B[1]]']
65 changes: 35 additions & 30 deletions tests/cases/issues/test_issue_246.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,45 +2,50 @@
# + All contributors to <https://github.com/smarie/python-pyfields>
#
# License: 3-clause BSD, <https://github.com/smarie/python-pyfields/blob/master/LICENSE>
from distutils.version import LooseVersion

import pytest
from pytest_cases import parametrize_with_cases

PYTEST_VERSION = LooseVersion(pytest.__version__)
PYTEST3_OR_GREATER = PYTEST_VERSION >= LooseVersion('3.0.0')

@pytest.mark.foo
class MarkedCases:
@pytest.mark.bar
def case_instance(self):
return 1
if PYTEST3_OR_GREATER:
@pytest.mark.foo
class MarkedCases:
@pytest.mark.bar
def case_instance(self):
return 1

@staticmethod
@pytest.mark.bar
def case_static():
return 2
@staticmethod
@pytest.mark.bar
def case_static():
return 2

@classmethod
@pytest.mark.bar
def case_classmethod(cls):
return 3
@classmethod
@pytest.mark.bar
def case_classmethod(cls):
return 3


@pytest.mark.foo
class TestNominal:
@pytest.mark.bar
def test_pytest_nominal(self, request):
# make sure the mark has been propagated from class to test
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'bar', 'foo'}
@pytest.mark.foo
class TestNominal:
@pytest.mark.bar
def test_pytest_nominal(self, request):
# make sure the mark has been propagated from class to test
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'bar', 'foo'}


@parametrize_with_cases('a', cases=MarkedCases)
def test_pytest_cases(a, request):
# make sure the mark has been propagated from case class to test
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'parametrize', 'foo', 'bar'}
@parametrize_with_cases('a', cases=MarkedCases)
def test_pytest_cases(a, request):
# make sure the mark has been propagated from case class to test
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'parametrize', 'foo', 'bar'}


@parametrize_with_cases('b', cases=MarkedCases)
def test_pytest_cases2(b, request):
# make sure the mark has been propagated from case class to test, but not a second time
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'parametrize', 'foo', 'bar'}
@parametrize_with_cases('b', cases=MarkedCases)
def test_pytest_cases2(b, request):
# make sure the mark has been propagated from case class to test, but not a second time
all_marks = tuple(m.name for m in request.node.iter_markers())
assert set(all_marks) == {'parametrize', 'foo', 'bar'}

0 comments on commit 795130c

Please sign in to comment.