diff --git a/pyproject.toml b/pyproject.toml index 926b62045c..37e4161032 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,7 @@ warn_unused_configs = true [[tool.mypy.overrides]] module = [ + "src.archivematicaCommon.lib.executeOrRunSubProcess", "src.MCPClient.lib.client.*", "src.MCPClient.lib.clientScripts.characterize_file", "src.MCPClient.lib.clientScripts.has_packages", diff --git a/src/archivematicaCommon/lib/executeOrRunSubProcess.py b/src/archivematicaCommon/lib/executeOrRunSubProcess.py index b23bcfaedc..b7aff8d3c5 100644 --- a/src/archivematicaCommon/lib/executeOrRunSubProcess.py +++ b/src/archivematicaCommon/lib/executeOrRunSubProcess.py @@ -21,18 +21,29 @@ import subprocess import sys import tempfile +from typing import Dict +from typing import List +from typing import Optional +from typing import Tuple +from typing import Union from archivematicaFunctions import escape +Arguments = List[str] +Input = Union[str, bytes, io.IOBase] +Environment = Dict[str, str] +Command = Union[str, List[str]] +Result = Tuple[int, str, str] + def launchSubProcess( - command, - stdIn="", - printing=True, - arguments=None, - env_updates=None, - capture_output=False, -): + command: Command, + stdIn: Input = "", + printing: bool = True, + arguments: Optional[Arguments] = None, + env_updates: Optional[Environment] = None, + capture_output: bool = False, +) -> Result: """ Launches a subprocess using ``command``, where ``command`` is either: a) a single string containing a commandline statement, or @@ -91,7 +102,7 @@ def launchSubProcess( stdin_pipe = subprocess.PIPE communicate_input = stdIn elif isinstance(stdIn, io.IOBase): - stdin_pipe = stdIn + stdin_pipe = stdIn.fileno() communicate_input = None else: raise Exception("stdIn must be a string or a file object") @@ -141,8 +152,13 @@ def launchSubProcess( def createAndRunScript( - text, stdIn="", printing=True, arguments=None, env_updates=None, capture_output=True -): + text: Command, + stdIn: Input = "", + printing: bool = True, + arguments: Optional[Arguments] = None, + env_updates: Optional[Environment] = None, + capture_output: bool = True, +) -> Result: if arguments is None: arguments = [] if env_updates is None: @@ -152,7 +168,10 @@ def createAndRunScript( encoding="utf-8", mode="wt", delete=False ) as tmpfile: os.chmod(tmpfile.name, 0o770) - tmpfile.write(text) + if isinstance(text, str): + tmpfile.write(text) + else: + tmpfile.write(" ".join(text)) tmpfile.close() cmd = [tmpfile.name] cmd.extend(arguments) @@ -170,14 +189,14 @@ def createAndRunScript( def executeOrRun( - type, - text, - stdIn="", - printing=True, - arguments=None, - env_updates=None, - capture_output=True, -): + type: str, + text: Command, + stdIn: Input = "", + printing: bool = True, + arguments: Optional[Arguments] = None, + env_updates: Optional[Environment] = None, + capture_output: bool = True, +) -> Result: """ Attempts to run the provided command on the shell, with the text of "stdIn" passed as standard input if provided. The type parameter @@ -222,7 +241,7 @@ def executeOrRun( capture_output=capture_output, ) if type == "bashScript": - text = "#!/bin/bash\n" + text + text = rf"#!/bin/bash\n{text}" return createAndRunScript( text, stdIn=stdIn, @@ -232,7 +251,7 @@ def executeOrRun( capture_output=capture_output, ) if type == "pythonScript": - text = "#!/usr/bin/env python\n" + text + text = rf"#!/usr/bin/env python\n{text}" return createAndRunScript( text, stdIn=stdIn, @@ -250,3 +269,4 @@ def executeOrRun( env_updates=env_updates, capture_output=capture_output, ) + raise ValueError(f"unknown type {type}")