MCPcopy Index your code
hub / github.com/RustPython/RustPython / _runtest

Method _runtest

Lib/test/libregrtest/run_workers.py:356–388  ·  view source on GitHub ↗
(self, test_name: TestName)

Source from the content-addressed store, hash-verified

354 return (result, stdout)
355
356 def _runtest(self, test_name: TestName) -> MultiprocessResult:
357 with contextlib.ExitStack() as stack:
358 stdout_file = self.create_stdout(stack)
359 json_file, json_tmpfile = self.create_json_file(stack)
360 worker_runtests = self.create_worker_runtests(test_name, json_file)
361
362 retcode: str | int | None
363 retcode, tmp_files = self.run_tmp_files(worker_runtests,
364 stdout_file.fileno())
365
366 stdout = self.read_stdout(stdout_file)
367
368 if retcode is None:
369 raise WorkerError(self.test_name, stdout=stdout,
370 err_msg=None,
371 state=State.TIMEOUT)
372 if retcode != 0:
373 name = support.get_signal_name(retcode)
374 if name:
375 retcode = f"{retcode} ({name})"
376 raise WorkerError(self.test_name, f"Exit code {retcode}", stdout,
377 state=State.WORKER_FAILED)
378
379 result, stdout = self.read_json(json_file, json_tmpfile, stdout)
380
381 if tmp_files:
382 msg = (f'\n\n'
383 f'Warning -- {test_name} leaked temporary files '
384 f'({len(tmp_files)}): {", ".join(sorted(tmp_files))}')
385 stdout += msg
386 result.set_env_changed()
387
388 return MultiprocessResult(result, stdout)
389
390 def run(self) -> None:
391 fail_fast = self.runtests.fail_fast

Callers 1

runMethod · 0.95

Calls 13

create_stdoutMethod · 0.95
create_json_fileMethod · 0.95
run_tmp_filesMethod · 0.95
read_stdoutMethod · 0.95
read_jsonMethod · 0.95
WorkerErrorClass · 0.85
lenFunction · 0.85
sortedFunction · 0.85
MultiprocessResultClass · 0.85
set_env_changedMethod · 0.80
filenoMethod · 0.45

Tested by

no test coverage detected