Andrew Hamilton
feb0d0a68c
ProcessPoolExecutor relied on private API, lost os.setpriority and auto-restart. Revert to the subprocess worker but spawn the current executable with --worker instead of a separate eris-worker console script, keeping a single entry point for shiv/pyinstaller. Bump .python-version to 3.14.
42 lines
1.2 KiB
Python
Executable file
42 lines
1.2 KiB
Python
Executable file
|
|
import asyncio
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
import eris.tools as tools
|
|
import eris.worker as worker
|
|
|
|
|
|
class WorkerTestCase(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.original_working_dir = os.getcwd()
|
|
os.chdir(self.temp_dir)
|
|
os.mkdir(tools.CACHE_PATH)
|
|
open("foo", "w").close()
|
|
|
|
def tearDown(self):
|
|
if hasattr(self, "worker_"):
|
|
self.worker_.kill()
|
|
shutil.rmtree(self.temp_dir)
|
|
os.chdir(self.original_working_dir)
|
|
|
|
def test_run_job(self):
|
|
loop = asyncio.new_event_loop()
|
|
compression = "none"
|
|
self.worker_ = worker.Worker(compression)
|
|
command = [sys.executable, "-m", "eris.worker", "--worker"]
|
|
loop.run_until_complete(self.worker_.create_process(command=command))
|
|
future = self.worker_.run_tool("foo", tools.metadata)
|
|
status = loop.run_until_complete(future)
|
|
self.assertEqual(status, tools.Status.ok)
|
|
result_path = os.path.join(tools.CACHE_PATH, "foo-metadata")
|
|
self.assertTrue(os.path.exists(result_path))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|