test: distribute tests in different processes to avoid conflicts

pyproject.toml

@@ -38,6 +38,7 @@ pytest = "^7"
 pytest-asyncio = "^0.23"
 pytest-cov = "^4"
 pytest-xdist = "^3"
+pytest-forked = "^1.6"
 # This also installs scipy and wcwidth
 datar = { version = "^0.15", extras = ["pandas"] }
 

tests/test_utils.py

@@ -24,13 +24,13 @@
 
 HERE = Path(__file__).parent.resolve()
 
-
+@pytest.mark.forked
 def test_get_logger(caplog):
     logger = get_logger("test", "info")
     logger.debug("debug message")
     assert "debug message" not in caplog.text
 
-
+@pytest.mark.forked
 def test_plugin_context():
     with pytest.raises(ConfigurationError):
         get_plugin_context(["a", "no:b"])
@@ -43,18 +43,18 @@ def test_plugin_context():
 
     assert plugin.get_plugin("core").enabled is True
 
-
+@pytest.mark.forked
 def test_brief_list():
     assert brief_list([1]) == "1"
     assert brief_list([1, 2, 3]) == "1-3"
 
-
+@pytest.mark.forked
 def test_get_mtime_dir():
     package_dir = Path(pipen.__file__).parent
     mtime = get_mtime(package_dir, 2)
     assert mtime > 0
 
-
+@pytest.mark.forked
 def test_desc_from_docstring():
     class Base:
         ...
@@ -70,7 +70,7 @@ class Obj1(Base):
     desc = desc_from_docstring(Obj1, Base)
     assert desc == "abc def"
 
-
+@pytest.mark.forked
 def test_update_dict():
     assert update_dict(None, None) is None
     assert update_dict({}, None) == {}
@@ -100,31 +100,31 @@ def test_update_dict():
         depth=1,
     ) == {"a": {"b1": {"c": 2}}}
 
-
+@pytest.mark.forked
 def test_strsplit():
     assert strsplit("a ,b ", ",", trim=None) == ["a ", "b "]
     assert strsplit("a , b", ",", trim="left") == ["a ", "b"]
     assert strsplit("a , b", ",", trim="right") == ["a", " b"]
 
-
+@pytest.mark.forked
 def test_get_shebang():
     assert get_shebang("") is None
     assert get_shebang("#!bash") == "bash"
     assert get_shebang("#!bash \n") == "bash"
 
-
+@pytest.mark.forked
 def test_ignore_firstline_dedent():
     text = """
 
     a
     """
     assert ignore_firstline_dedent(text) == "a\n"
 
-
+@pytest.mark.forked
 def test_truncate_text():
     assert truncate_text("abcd", 2) == "a…"
 
-
+@pytest.mark.forked
 def test_mark():
     @mark(a=1)
     class P1(pipen.Proc):
@@ -157,7 +157,7 @@ class Z(Y):
     # Marks inherited, as Y/Z are not Proc nor ProcGroup
     assert get_marked(Z, "a", None) == 1
 
-
+@pytest.mark.forked
 def test_get_obj_from_spec():
     with pytest.raises(ValueError):
         _get_obj_from_spec("a.b.c")
@@ -169,6 +169,7 @@ def test_get_obj_from_spec():
     assert obj is pipen.Pipen
 
 
+@pytest.mark.forked
 @pytest.mark.asyncio
 async def test_load_pipeline(tmp_path):
     with pytest.raises(TypeError):
@@ -199,13 +200,13 @@ class P1(Proc):
     assert len(pipeline.procs) == 1
 
 
-@pytest.mark.xdist_group(name="separated")
+@pytest.mark.forked
 # To avoid: Another plugin named simpleplugin has already been registered.
 @pytest.mark.asyncio
 async def test_is_load_pipeline_with_help(tmp_path):
     pipeline = await load_pipeline(
         f"{HERE}/helpers.py:PipenIsLoading",
-        "pipen",  # not @pipen
+        "_",  # not @pipen
         ["--help"],
     )
     assert pipeline.name == "PipenIsLoading"

tests/test_xqute_pars.py

@@ -3,10 +3,11 @@
 import pytest
 
 import time
-from .helpers import RetryProc, pipen
+from .helpers import RetryProc, pipen  # noqa: F401
 
 
-def test_retry(caplog, pipen):
+@pytest.mark.forked
+def test_retry(caplog, pipen):  # noqa: F811
     proc = Proc.from_proc(RetryProc)
     rc = pipen.set_starts(proc).set_data([time.time()]).run()
     assert "Retrying" in caplog.text