1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
|
Author: Santiago Vila <sanvila@debian.org>
Last-Update: 2023-10-28
Bug-Debian: https://bugs.debian.org/1030885
Description: Skip parallel tests on single-cpu systems
--- a/tests/test_app/test_evo.py
+++ b/tests/test_app/test_evo.py
@@ -1,8 +1,9 @@
import pathlib
+import multiprocessing
from os.path import dirname, join
from tempfile import TemporaryDirectory
-from unittest import TestCase, main
+from unittest import TestCase, main, skipIf
from unittest.mock import MagicMock
from numpy.testing import assert_allclose, assert_raises
@@ -854,6 +855,7 @@
# correct message being relayed
self.assertTrue("ValueError: '-' at" in result.message)
+ @skipIf(multiprocessing.cpu_count() == 1, "Does not work on single-cpu systems")
def test_bstrap_parallel(self):
"""exercising bootstrap with parallel"""
aln = load_aligned_seqs(join(data_dir, "brca1.fasta"), moltype="dna")
--- a/tests/test_app/test_io_new.py
+++ b/tests/test_app/test_io_new.py
@@ -4,8 +4,10 @@
import os
import pathlib
import pickle
+import multiprocessing
from pathlib import Path
+from unittest import skipIf
import numpy
import pytest
@@ -479,6 +481,7 @@
assert isinstance(writer.data_store.summary_not_completed, Table)
+@skipIf(multiprocessing.cpu_count() == 1, "Does not work on single-cpu systems")
def test_write_db_parallel(tmp_dir, fasta_dir):
"""writing with overwrite in parallel should reset db"""
dstore = open_data_store(fasta_dir, suffix="fasta")
--- a/tests/test_util/test_parallel.py
+++ b/tests/test_util/test_parallel.py
@@ -35,6 +35,7 @@
class ParallelTests(TestCase):
+ @skipIf(multiprocessing.cpu_count() == 1, "Does not work on single-cpu systems")
def test_create_processes(self):
"""Procressor pool should create multiple distingue processes"""
max_worker_count = multiprocessing.cpu_count() - 1
@@ -45,6 +46,7 @@
self.assertEqual(sorted(list(result_values)), index)
self.assertEqual(len(set(result_processes)), max_worker_count)
+ @skipIf(multiprocessing.cpu_count() == 1, "Does not work on single-cpu systems")
def test_random_seeding(self):
"""Random seed should be set every function call"""
# On Windows process ids are not guaranteed to be sequential(1,2,3,4...)
@@ -56,6 +58,7 @@
self.assertEqual(result1[0], result2[0])
self.assertNotEqual(result1, result2)
+ @skipIf(multiprocessing.cpu_count() == 1, "Does not work on single-cpu systems")
@skipIf(sys.version_info[1] < 7, "method exclusive to Python 3.7 and above")
def test_is_master_process(self):
"""
|
