Coverage for arguably/__main__.py: 77%

1#!/usr/bin/env python3 

2import multiprocessing 

3import queue 

4import sys 

5from pathlib import Path 

6from typing import List 

7 

8import arguably 

9from ._util import load_and_run, LoadAndRunResult, run_redirected_io 

10 

11args_for_file: List[str] = [] 

12argv_cut_index = 2 

13 

14 

15@arguably.command 

16def main(file: Path, *args: str, debug: bool = False, no_warn: bool = False) -> None: 

17 """ 

18 run functions from any python file 

19 

20 Args: 

21 file: the file to load 

22 *args: the function to run, as well as any arguments 

23 debug: if set, will show a debug log for how argparse is set up and for how functions are called 

24 no_warn: if set, will not show warnings 

25 """ 

26 

27 # Check that the user-specified file exists 

28 file = file.expanduser() 

29 if not file.exists(): 

30 arguably.error(f"file {str(file)} does not exist") 

31 

32 # Prepare argv for the invocation of arguably in the subprocess 

33 del sys.argv[:1] # Remove argv[0] - the filename becomes argv[0]. 

34 if debug: 

35 sys.argv.remove("--debug") 

36 if no_warn: 

37 sys.argv.remove("--no-warn") 

38 

39 # Run `load_and_run` on the file in a spawned process 

40 mp_ctx = multiprocessing.get_context("spawn") 

41 results_queue: multiprocessing.Queue[LoadAndRunResult] = mp_ctx.Queue() 

42 run_redirected_io(mp_ctx, load_and_run, (results_queue, file, args_for_file, debug, no_warn)) 

43 

44 # Check the results 

45 try: 

46 result = results_queue.get(timeout=0) 

47 except queue.Empty: 

48 arguably.error("no results from launched process") 

49 else: 

50 if result.error: 

51 arguably.error(result.error) 

52 elif result.exception: 

53 raise result.exception 

54 

55 

56if __name__ == "__main__": 

57 # We strip off the rest of the arguments - if we were doing things normally, this wouldn't be required - however, 

58 # we're effectively adding a subcommand without telling argparse, which will cause issues if there are any --options 

59 # passed in. 

60 if "--debug" in sys.argv[1 : argv_cut_index + 1]: 

61 argv_cut_index += 1 

62 if "--no-warn" in sys.argv[1 : argv_cut_index + 1]: 

63 argv_cut_index += 1 

64 if len(sys.argv) > argv_cut_index: 

65 args_for_file = sys.argv[argv_cut_index:] 

66 del sys.argv[argv_cut_index:] 

67 

68 # Run it 

69 arguably.run(name="arguably")