Coverage for melissa/utility/process.py: 81%

1#!/usr/bin/python3 

2 

3# Copyright (c) 2022, Institut National de Recherche en Informatique et en Automatique (Inria) 

4# All rights reserved. 

5# 

6# Redistribution and use in source and binary forms, with or without 

7# modification, are permitted provided that the following conditions are met: 

8# 

9# * Redistributions of source code must retain the above copyright notice, this 

10# list of conditions and the following disclaimer. 

11# 

12# * Redistributions in binary form must reproduce the above copyright notice, 

13# this list of conditions and the following disclaimer in the documentation 

14# and/or other materials provided with the distribution. 

15# 

16# * Neither the name of the copyright holder nor the names of its 

17# contributors may be used to endorse or promote products derived from 

18# this software without specific prior written permission. 

19# 

20# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 

21# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 

22# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 

23# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 

24# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 

25# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 

26# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER 

27# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, 

28# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 

29# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 

30 

31"""This module provides helper functions for launching new processes with `subprocess`.""" 

32 

33__all__ = ["ArgumentList", "CompletedProcess", "Environment", "Process"] 

34 

35import subprocess 

36from typing import List, Dict, Union, TextIO 

37import os 

38 

39ArgumentList = List[str] 

40Environment = Union[os._Environ, Dict[str, str]] 

41Process = subprocess.Popen 

42 

43 

44class CompletedProcess: 

45 """This class provides attributes for retaining standard output and error for `Scheduler`""" 

46 def __init__(self, exit_status: int, stdout: str, stderr: str) -> None: 

47 self.exit_status = exit_status 

48 self.stdout = stdout 

49 self.stderr = stderr 

50 

51 

52def launch( 

53 args: ArgumentList, env: Environment, stdout: TextIO, stderr: TextIO 

54) -> "Process[str]": 

55 """Launches a new subprocess with a given argument list and environment.""" 

56 process = subprocess.Popen( 

57 args=args, 

58 env=env, 

59 stdin=subprocess.DEVNULL, 

60 stdout=stdout, 

61 stderr=stderr, 

62 universal_newlines=True 

63 ) 

64 process.env = env # type: ignore 

65 return process