explosion/spaCy
1
1
Fork 0
mirror of https://github.com/explosion/spaCy.git synced 2025-08-26 23:14:55 +03:00
Code Issues Packages Projects Releases Wiki Activity

Use mkdtemp()

Browse Source
This commit is contained in:
richardpaulhudson 2022-10-04 17:34:52 +02:00
parent b8a299f349
commit 2cc2cc1a15
1 changed files with 50 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
spacy/tests/test_parallel.py
View File

@ -1,6 +1,7 @@
import os
import shutil
from time import time
from tempfile import TemporaryDirectory
from tempfile import mkdtemp
from pathlib import Path
import pytest
import srsly
@ -220,56 +221,59 @@ def test_project_run_multiprocessing_failure(failing_command: str):
with a non-zero return code, the other two commands after several seconds with
zero return codes. Measuring the execution length for the whole group shows
whether or not the sleeping processes were successfully terminated."""
with TemporaryDirectory(ignore_cleanup_errors=True) as d: # avoid errors in Windows CI
dirpath = Path(d)
pscript = """
d = mkdtemp() # avoid errors in Windows CI that occur with make_tempdir()
dirpath = Path(d)
pscript = """
import sys
from time import sleep
_, sleep_secs, rc = sys.argv
sleep(int(sleep_secs))
sys.exit(int(rc))
"""
"""
pscript_loc = os.sep.join((str(d), "pscript.py"))
with open(pscript_loc, "w") as pscript_file:
pscript_file.write(pscript)
os.chmod(pscript_loc, 0o777)
project = {
"commands": [
{
"name": "commandA",
"script": [" ".join(("python", pscript_loc, "25", "0"))],
},
{
"name": "commandB",
"script": [" ".join((failing_command, pscript_loc, "0", "1"))],
},
{
"name": "commandC",
"script": [" ".join(("python", pscript_loc, "20", "0"))],
},
],
"workflows": {
"all": [
{"parallel": ["commandA", "commandB", "commandC"]},
"commandC",
]
pscript_loc = os.sep.join((d, "pscript.py"))
with open(pscript_loc, "w") as pscript_file:
pscript_file.write(pscript)
os.chmod(pscript_loc, 0o777)
project = {
"commands": [
{
"name": "commandA",
"script": [" ".join(("python", pscript_loc, "25", "0"))],
},
}
srsly.write_yaml(dirpath / "project.yml", project)
load_project_config(dirpath)
start = time()
with pytest.raises(SystemExit) as rc_e:
project_run(dirpath, "all")
if os.name == "nt":
# because in Windows the terminated process has rc=15 rather than rc=-15,
# 15 is the highest rc rather than 1.
assert rc_e.value.code == 15
else:
assert rc_e.value.code == 1
time_taken = time() - start
assert (
time_taken < 15
), "Test took {time_taken}, subprocess seems not to have been terminated"
{
"name": "commandB",
"script": [" ".join((failing_command, pscript_loc, "0", "1"))],
},
{
"name": "commandC",
"script": [" ".join(("python", pscript_loc, "20", "0"))],
},
],
"workflows": {
"all": [
{"parallel": ["commandA", "commandB", "commandC"]},
"commandC",
]
},
}
srsly.write_yaml(dirpath / "project.yml", project)
load_project_config(dirpath)
start = time()
with pytest.raises(SystemExit) as rc_e:
project_run(dirpath, "all")
if os.name == "nt":
# because in Windows the terminated process has rc=15 rather than rc=-15,
# 15 is the highest rc rather than 1.
assert rc_e.value.code == 15
else:
assert rc_e.value.code == 1
time_taken = time() - start
assert (
time_taken < 15
), "Test took {time_taken}, subprocess seems not to have been terminated"
try:
shutil.rmtree(d)
except: # sometimes fails in Windows CI owing to CI-specific race condition
pass