separate test and coverage

Author: cdiener

.github/workflows/main.yml

@@ -35,7 +35,7 @@ jobs:
         python -m pip install tox tox-gh-actions
     - name: Test with tox
       run:
-        tox -- --cov-report=xml --benchmark-skip
+        tox -- --benchmark-skip
       timeout-minutes: 60
     - name: Report coverage
       shell: bash

src/cobra/test/test_flux_analysis/test_deletion.py

@@ -5,6 +5,7 @@
 from typing import Callable, List
 
 import numpy as np
+import
 import pandas as pd
 import pytest
 
@@ -288,6 +289,7 @@ def test_double_gene_deletion_benchmark(
     benchmark(double_gene_deletion, large_model, gene_list1=genes, processes=1)
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_double_gene_deletion(model: Model) -> None:
     """Test double gene deletion."""
     genes = [
@@ -365,18 +367,17 @@ def test_double_gene_deletion(model: Model) -> None:
             "b4025": 0.863,
         },
     }
-    if __name__ == "__main__":
-        solution = double_gene_deletion(model, gene_list1=genes, processes=3)
-        solution_one_process = double_gene_deletion(
-            model, gene_list1=genes, processes=1
-        )
+    solution = double_gene_deletion(model, gene_list1=genes, processes=3)
+    solution_one_process = double_gene_deletion(
+        model, gene_list1=genes, processes=1
+    )
 
-        for rxn_a, sub in growth_dict.items():
-            for rxn_b, growth in sub.items():
-                sol = solution.knockout[{rxn_a, rxn_b}]
-                sol_one = solution_one_process.knockout[{rxn_a, rxn_b}]
-                assert np.isclose(sol.growth, growth, atol=1e-3)
-                assert np.isclose(sol_one.growth, growth, atol=1e-3)
+    for rxn_a, sub in growth_dict.items():
+        for rxn_b, growth in sub.items():
+            sol = solution.knockout[{rxn_a, rxn_b}]
+            sol_one = solution_one_process.knockout[{rxn_a, rxn_b}]
+            assert np.isclose(sol.growth, growth, atol=1e-3)
+            assert np.isclose(sol_one.growth, growth, atol=1e-3)
 
 
 def test_double_gene_knockout_bug(large_model: Model) -> None:
@@ -415,6 +416,7 @@ def test_double_reaction_deletion_benchmark(
     benchmark(double_reaction_deletion, large_model, reaction_list1=reactions)
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_double_reaction_deletion(model: Model) -> None:
     """Test double reaction deletion."""
     reactions = ["FBA", "ATPS4r", "ENO", "FRUpts2"]
@@ -424,23 +426,22 @@ def test_double_reaction_deletion(model: Model) -> None:
         "ENO": {"FRUpts2": 0.0},
     }
 
-    if __name__ == "__main__":
-        solution = double_reaction_deletion(
-            model, reaction_list1=reactions, processes=3
-        )
-        solution_one_process = double_reaction_deletion(
-            model, reaction_list1=reactions, processes=1
-        )
-        for (rxn_a, sub) in growth_dict.items():
-            for rxn_b, growth in sub.items():
-                sol = solution.knockout[{rxn_a, rxn_b}]
-                sol_one = solution_one_process.knockout[{rxn_a, rxn_b}]
-                if math.isnan(growth):
-                    assert math.isnan(sol.growth)
-                    assert math.isnan(sol_one.growth)
-                else:
-                    assert np.isclose(sol.growth, growth, atol=1e-3)
-                    assert np.isclose(sol_one.growth, growth, atol=1e-3)
+    solution = double_reaction_deletion(
+        model, reaction_list1=reactions, processes=3
+    )
+    solution_one_process = double_reaction_deletion(
+        model, reaction_list1=reactions, processes=1
+    )
+    for (rxn_a, sub) in growth_dict.items():
+        for rxn_b, growth in sub.items():
+            sol = solution.knockout[{rxn_a, rxn_b}]
+            sol_one = solution_one_process.knockout[{rxn_a, rxn_b}]
+            if math.isnan(growth):
+                assert math.isnan(sol.growth)
+                assert math.isnan(sol_one.growth)
+            else:
+                assert np.isclose(sol.growth, growth, atol=1e-3)
+                assert np.isclose(sol_one.growth, growth, atol=1e-3)
 
 
 def test_deletion_accessor(small_model: Model) -> None:

src/cobra/test/test_flux_analysis/test_variability.py

@@ -3,6 +3,7 @@
 from typing import Callable, List
 
 import numpy as np
+import os
 import pandas as pd
 import pytest
 
@@ -84,11 +85,11 @@ def test_flux_variability(
     assert np.allclose(fva_out, fva_results)
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_parallel_flux_variability(
     model: Model, fva_results: pd.DataFrame, all_solvers: List[str]
 ) -> None:
     """Test parallel FVA."""
-    if __name__ == "__main__":
         model.solver = all_solvers
         fva_out = flux_variability_analysis(model, processes=2)
         fva_out.sort_index(inplace=True)

src/cobra/test/test_sampling/test_optgp.py

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING, Callable
 
 import numpy as np
+import os
 import pytest
 
 from cobra.sampling import OptGPSampler
@@ -23,10 +24,10 @@ def optgp(model: "Model") -> OptGPSampler:
     return sampler
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_optgp_init_benchmark(model: "Model", benchmark: Callable) -> None:
     """Benchmark inital OptGP sampling."""
-    if __name__ == "__main__":
-        benchmark(lambda: OptGPSampler(model, processes=2))
+    benchmark(lambda: OptGPSampler(model, processes=2))
 
 
 def test_optgp_sample_benchmark(optgp: "Model", benchmark: Callable) -> None:

src/cobra/test/test_sampling/test_sampling.py

@@ -1,6 +1,7 @@
 """Test functionalities of flux sampling methods."""
 
 import numpy as np
+import os
 import pytest
 
 from cobra.core import Metabolite, Model, Reaction
@@ -20,11 +21,12 @@ def test_single_optgp(model: Model) -> None:
     assert s.shape == (10, len(model.reactions))
 
 
-def test_multi_optgp(model: Model) -> None:
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
+def test_multi_optgp(model: Model) -> None:  # pragma: no cover
     """Test OptGP sampling (multi sample)."""
-    if __name__ == "__main__":
-        s = sample(model, 10, processes=2)
-        assert s.shape == (10, len(model.reactions))
+    s = sample(model, 10, processes=2)
+    assert s.shape == (10, len(model.reactions))
+    raise ValueError(__name__)
 
 
 def test_wrong_method(model: Model) -> None:

src/cobra/test/test_util/test_process_pool.py

@@ -3,6 +3,7 @@
 
 from typing import Iterable, Tuple
 
+import os
 import pytest
 from pytest_mock import MockerFixture
 
@@ -24,6 +25,7 @@ def summation(*args: Iterable[int]) -> int:
     return sum(args)
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 @pytest.mark.parametrize(
     "attributes",
     [
@@ -36,93 +38,92 @@ def summation(*args: Iterable[int]) -> int:
 )
 def test_init(attributes: dict) -> None:
     """Test that a process pool can be initialized with each of its arguments."""
-    if __name__ == "__main__":
-        ProcessPool(**attributes)
+    ProcessPool(**attributes)
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_close(mocker: MockerFixture) -> None:
     """Test that the composed pool is closed as well."""
-    if __name__ == "__main__":
-        pool = ProcessPool(processes=3)
-        mock = mocker.patch.object(pool, "_pool", autospec=True)
-        pool.close()
-        mock.close.assert_called_once()
+    pool = ProcessPool(processes=3)
+    mock = mocker.patch.object(pool, "_pool", autospec=True)
+    pool.close()
+    mock.close.assert_called_once()
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_with_context(mocker: MockerFixture) -> None:
     """Test that the composed pool's context is managed as well."""
-    if __name__ == "__main__":
-        pool = ProcessPool(processes=3)
-        mock = mocker.patch.object(pool, "_pool", autospec=True)
-        with pool:
-            pass
-        mock.__enter__.assert_called_once()
-        mock.__exit__.assert_called_once()
+    pool = ProcessPool(processes=3)
+    mock = mocker.patch.object(pool, "_pool", autospec=True)
+    with pool:
+        pass
+    mock.__enter__.assert_called_once()
+    mock.__exit__.assert_called_once()
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_apply() -> None:
     """Test that a function can be applied."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            assert pool.apply(square, (3,)) == 9
+    with ProcessPool(processes=3) as pool:
+        assert pool.apply(square, (3,)) == 9
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_apply_async() -> None:
     """Test that a function can be applied asynchronously."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            assert pool.apply_async(square, (3,)).get() == 9
+    with ProcessPool(processes=3) as pool:
+        assert pool.apply_async(square, (3,)).get() == 9
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_map() -> None:
     """Test that a function can be mapped over an iterable of values."""
-    if __name__ == "__main":
-        with ProcessPool(processes=3) as pool:
-            assert sum(pool.map(square, [2] * 6)) == 24
+    with ProcessPool(processes=3) as pool:
+        assert sum(pool.map(square, [2] * 6)) == 24
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_map_async() -> None:
     """Test that a function can be mapped over an iterable of values asynchronously."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            assert sum(pool.map_async(square, [2] * 6).get()) == 24
+    with ProcessPool(processes=3) as pool:
+        assert sum(pool.map_async(square, [2] * 6).get()) == 24
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_imap() -> None:
     """Test that mapped function results can be iterated."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            total = 0
-            for result in pool.imap(square, [2] * 6):
-                total += result
-            assert total == 24
+    with ProcessPool(processes=3) as pool:
+        total = 0
+        for result in pool.imap(square, [2] * 6):
+            total += result
+        assert total == 24
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_imap_unordered() -> None:
     """Test that mapped function results can be iterated in any order."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            assert sum(pool.imap_unordered(square, [2] * 6)) == 24
+    with ProcessPool(processes=3) as pool:
+        assert sum(pool.imap_unordered(square, [2] * 6)) == 24
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_starmap() -> None:
     """Test that a function can be starmapped over many iterables."""
-    if __name__ == "__main__":
-        with ProcessPool(processes=3) as pool:
-            assert (
-                sum(pool.starmap(summation, [range(10), range(10), range(10)])) == 135
-            )
+    with ProcessPool(processes=3) as pool:
+        assert (
+            sum(pool.starmap(summation, [range(10), range(10), range(10)])) == 135
+        )
 
 
+@pytest.mark.skipif("SKIP_MP" in os.environ, reason="unsafe for parallel execution")
 def test_starmap_async() -> None:
-    """Test that a function can be starmapped over many iterables asynchronously."""
-    if __name__ == "__main":
-        with ProcessPool(processes=3) as pool:
-            assert (
-                sum(
-                    pool.starmap_async(
-                        summation, [range(10), range(10), range(10)]
-                    ).get()
-                )
-                == 135
+"""Test that a function can be starmapped over many iterables asynchronously."""
+    with ProcessPool(processes=3) as pool:
+        assert (
+            sum(
+                pool.starmap_async(
+                    summation, [range(10), range(10), range(10)]
+                ).get()
             )
+            == 135
+        )

tox.ini

@@ -6,10 +6,24 @@ python =
     3.6: install, py36
     3.7: install, py37
     3.8: install, py38
-    3.9: install, py39
+    3.9: install, py39, coverage
     3.10: install, py310
 
 [testenv]
+extras =
+    array
+deps=
+    jsonschema
+    osqp~=0.6 ; python_version < "3.10"
+    pytest
+    pytest-benchmark
+    pytest-mock
+    pytest-raises
+commands =
+    pytest {posargs: --benchmark-skip}
+
+[testenv:coverage]
+setenv = SKIP_MP = 1
 extras =
     array
 deps=
@@ -21,7 +35,7 @@ deps=
     pytest-mock
     pytest-raises
 commands =
-    pytest --cov=cobra --cov-report=term {posargs: --benchmark-skip}
+    pytest --cov=cobra --cov-report=xml {posargs: --benchmark-skip}
 
 [testenv:isort]
 skip_install = True