@@ -1059,232 +1059,6 @@ def test_funcarg_not_active(testdir):
1059
1059
assert result .ret == 0
1060
1060
1061
1061
1062
- @pytest .mark .skipif ("sys.version_info[0] < 3" , reason = "no context manager api on Python 2" )
1063
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1064
- @pytest .mark .skipif ('platform.python_implementation() == "PyPy"' , reason = "often deadlocks on PyPy" )
1065
- @pytest .mark .parametrize ('method' , ['fork' , 'spawn' ])
1066
- @pytest .mark .xfail
1067
- def test_multiprocessing_pool (pytester , testdir , method ):
1068
- pytest .importorskip ('multiprocessing.util' )
1069
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1070
- script = testdir .makepyfile ('''
1071
- import multiprocessing
1072
-
1073
- def target_fn(a):
1074
- {}se: # pragma: nocover
1075
- return None
1076
-
1077
- def test_run_target():
1078
- multiprocessing.set_start_method({!r})
1079
- for i in range(33):
1080
- with multiprocessing.Pool(3) as p:
1081
- p.map(target_fn, [i * 3 + j for j in range(3)])
1082
- p.join()
1083
- ''' .format ('' .join ('''if a == %r:
1084
- return a
1085
- el''' % i for i in range (99 )), method ))
1086
-
1087
- result = testdir .runpytest ('-v' ,
1088
- '--cov=%s' % script .dirpath (),
1089
- '--cov-report=term-missing' ,
1090
- script )
1091
-
1092
- assert "Doesn't seem to be a coverage.py data file" not in result .stdout .str ()
1093
- assert "Doesn't seem to be a coverage.py data file" not in result .stderr .str ()
1094
- result .stdout .fnmatch_lines ([
1095
- '*- coverage: platform *, python * -*' ,
1096
- 'test_multiprocessing_pool* 100%*' ,
1097
- '*1 passed*'
1098
- ])
1099
- assert not testdir .tmpdir .listdir (".coverage.*" )
1100
- assert result .ret == 0
1101
-
1102
-
1103
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1104
- @pytest .mark .skipif ('platform.python_implementation() == "PyPy"' , reason = "often deadlocks on PyPy" )
1105
- @pytest .mark .parametrize ('method' , ['fork' , 'spawn' ])
1106
- def test_multiprocessing_pool_terminate (pytester , testdir , method ):
1107
- pytest .importorskip ('multiprocessing.util' )
1108
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1109
- script = testdir .makepyfile ('''
1110
- import multiprocessing
1111
-
1112
- def target_fn(a):
1113
- {}se: # pragma: nocover
1114
- return None
1115
-
1116
- def test_run_target():
1117
- multiprocessing.set_start_method({!r})
1118
-
1119
- for i in range(33):
1120
- p = multiprocessing.Pool(3)
1121
- try:
1122
- p.map(target_fn, [i * 3 + j for j in range(3)])
1123
- finally:
1124
- p.terminate()
1125
- p.join()
1126
- ''' .format ('' .join ('''if a == %r:
1127
- return a
1128
- el''' % i for i in range (99 )), method ))
1129
-
1130
- result = testdir .runpytest ('-v' ,
1131
- '--cov=%s' % script .dirpath (),
1132
- '--cov-report=term-missing' ,
1133
- script )
1134
-
1135
- assert "Doesn't seem to be a coverage.py data file" not in result .stdout .str ()
1136
- assert "Doesn't seem to be a coverage.py data file" not in result .stderr .str ()
1137
- result .stdout .fnmatch_lines ([
1138
- '*- coverage: platform *, python * -*' ,
1139
- 'test_multiprocessing_pool* 100%*' ,
1140
- '*1 passed*'
1141
- ])
1142
- assert not testdir .tmpdir .listdir (".coverage.*" )
1143
- assert result .ret == 0
1144
-
1145
-
1146
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1147
- @pytest .mark .skipif ('sys.version_info[0] > 2 and platform.python_implementation() == "PyPy"' , reason = "broken on PyPy3" )
1148
- @pytest .mark .parametrize ('method' , ['fork' , 'spawn' ])
1149
- def test_multiprocessing_pool_close (pytester , testdir , method ):
1150
- pytest .importorskip ('multiprocessing.util' )
1151
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1152
- script = testdir .makepyfile ('''
1153
- import multiprocessing
1154
-
1155
- def target_fn(a):
1156
- {}se: # pragma: nocover
1157
- return None
1158
-
1159
- def test_run_target():
1160
- multiprocessing.set_start_method({!r})
1161
- for i in range(33):
1162
- p = multiprocessing.Pool(3)
1163
- try:
1164
- p.map(target_fn, [i * 3 + j for j in range(3)])
1165
- finally:
1166
- p.close()
1167
- p.join()
1168
- ''' .format ('' .join ('''if a == %r:
1169
- return a
1170
- el''' % i for i in range (99 )), method ))
1171
-
1172
- result = testdir .runpytest ('-v' ,
1173
- '--cov=%s' % script .dirpath (),
1174
- '--cov-report=term-missing' ,
1175
- script )
1176
- assert "Doesn't seem to be a coverage.py data file" not in result .stdout .str ()
1177
- assert "Doesn't seem to be a coverage.py data file" not in result .stderr .str ()
1178
- result .stdout .fnmatch_lines ([
1179
- '*- coverage: platform *, python * -*' ,
1180
- 'test_multiprocessing_pool* 100%*' ,
1181
- '*1 passed*'
1182
- ])
1183
- # assert not testdir.tmpdir.listdir(".coverage.*")
1184
- assert result .ret == 0
1185
-
1186
-
1187
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1188
- @pytest .mark .parametrize ('method' , ['fork' , 'spawn' ])
1189
- def test_multiprocessing_process (pytester , testdir , method ):
1190
- pytest .importorskip ('multiprocessing.util' )
1191
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1192
- script = testdir .makepyfile ('''
1193
- import multiprocessing
1194
-
1195
- def target_fn():
1196
- a = True
1197
- return a
1198
-
1199
- def test_run_target():
1200
- multiprocessing.set_start_method({!r})
1201
- p = multiprocessing.Process(target=target_fn)
1202
- p.start()
1203
- p.join()
1204
- ''' .format (method ))
1205
-
1206
- result = testdir .runpytest ('-v' ,
1207
- '--cov=%s' % script .dirpath (),
1208
- '--cov-report=term-missing' ,
1209
- script )
1210
-
1211
- result .stdout .fnmatch_lines ([
1212
- '*- coverage: platform *, python * -*' ,
1213
- 'test_multiprocessing_process* 9 * 100%*' ,
1214
- '*1 passed*'
1215
- ])
1216
- assert result .ret == 0
1217
-
1218
-
1219
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1220
- def test_multiprocessing_process_no_source (pytester , testdir ):
1221
- pytest .importorskip ('multiprocessing.util' )
1222
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1223
- script = testdir .makepyfile ('''
1224
- import multiprocessing
1225
-
1226
- def target_fn():
1227
- a = True
1228
- return a
1229
-
1230
- def test_run_target():
1231
- p = multiprocessing.Process(target=target_fn)
1232
- p.start()
1233
- p.join()
1234
- ''' )
1235
-
1236
- result = testdir .runpytest ('-v' ,
1237
- '--cov' ,
1238
- '--cov-report=term-missing' ,
1239
- script )
1240
-
1241
- result .stdout .fnmatch_lines ([
1242
- '*- coverage: platform *, python * -*' ,
1243
- 'test_multiprocessing_process* 8 * 100%*' ,
1244
- '*1 passed*'
1245
- ])
1246
- assert result .ret == 0
1247
-
1248
-
1249
- @pytest .mark .skipif ('sys.platform == "win32"' , reason = "multiprocessing support is broken on Windows" )
1250
- def test_multiprocessing_process_with_terminate (pytester , testdir ):
1251
- pytest .importorskip ('multiprocessing.util' )
1252
- pytester .path .joinpath (".coveragerc" ).write_text (MP_COVERAGERC )
1253
- script = testdir .makepyfile ('''
1254
- import multiprocessing
1255
- import time
1256
- from pytest_cov.embed import cleanup_on_sigterm
1257
- cleanup_on_sigterm()
1258
-
1259
- event = multiprocessing.Event()
1260
-
1261
- def target_fn():
1262
- a = True
1263
- event.set()
1264
- time.sleep(5)
1265
-
1266
- def test_run_target():
1267
- p = multiprocessing.Process(target=target_fn)
1268
- p.start()
1269
- time.sleep(0.5)
1270
- event.wait(1)
1271
- p.terminate()
1272
- p.join()
1273
- ''' )
1274
-
1275
- result = testdir .runpytest ('-v' ,
1276
- '--cov=%s' % script .dirpath (),
1277
- '--cov-report=term-missing' ,
1278
- script )
1279
-
1280
- result .stdout .fnmatch_lines ([
1281
- '*- coverage: platform *, python * -*' ,
1282
- 'test_multiprocessing_process* 16 * 100%*' ,
1283
- '*1 passed*'
1284
- ])
1285
- assert result .ret == 0
1286
-
1287
-
1288
1062
@pytest .mark .skipif ('sys.platform == "win32"' , reason = "SIGTERM isn't really supported on Windows" )
1289
1063
def test_cleanup_on_sigterm (testdir ):
1290
1064
script = testdir .makepyfile ('''
0 commit comments