Coverage for .tox/coverage/lib/python3.13/site-packages/wuttjamaican/cli/runner.py: 100%

33 statements  

« prev     ^ index     » next       coverage.py v7.16.1, created at 2026-10-02 16:58 -0500

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# WuttJamaican -- Base package for Wutta Framework 

5# Copyright © 2023-2026 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24See also: :ref:`wutta-runner` 

25""" 

26 

27import logging 

28import subprocess 

29import sys 

30 

31try: 

32 from typing import Annotated 

33except ImportError: # pragma: no cover 

34 from typing_extensions import Annotated 

35 

36import humanize 

37import typer 

38 

39from .base import wutta_typer 

40 

41log = logging.getLogger(__name__) 

42 

43 

44@wutta_typer.command() 

45def runner( 

46 ctx: typer.Context, 

47 subject: Annotated[ 

48 str, 

49 typer.Option("--subject", "-S", help="Override value for email subject."), 

50 ] = None, 

51 skip_if_empty: Annotated[ 

52 bool, 

53 typer.Option( 

54 "--skip-if-empty", 

55 help="Skip sending email if the command generates no output.", 

56 ), 

57 ] = False, 

58 keep_exit_code: Annotated[ 

59 bool, 

60 typer.Option( 

61 "--keep-exit-code", 

62 help="Exit with same return code as subprocess. " 

63 "If not specified, exits with code 0 despite command failure.", 

64 ), 

65 ] = False, 

66 command: Annotated[ 

67 str, 

68 typer.Argument(help="Command to run, the result of which will be emailed."), 

69 ] = ..., 

70): 

71 """ 

72 Run a command as subprocess, and email the result/output 

73 """ 

74 config = ctx.parent.wutta_config 

75 app = config.get_app() 

76 

77 # prep command 

78 cmd = config.parse_list(command) 

79 log.info("running command as subprocess: %s", cmd) 

80 began = app.make_utc() 

81 

82 # run command 

83 try: 

84 output = subprocess.check_output(cmd, stderr=subprocess.STDOUT) 

85 retcode = 0 

86 log.info("command completed okay") 

87 except subprocess.CalledProcessError as error: 

88 output = error.output 

89 retcode = error.returncode 

90 log.warning("command failed with code %s", retcode) 

91 

92 # capture timing 

93 ended = app.make_utc() 

94 runtime = ended - began 

95 runtime_display = humanize.naturaldelta(runtime) 

96 

97 # convert output for readability 

98 output = output.decode(errors="replace") 

99 

100 # maybe send email 

101 if skip_if_empty and not output: 

102 log.info("command had no output, so will skip sending email") 

103 else: 

104 app.send_email( 

105 "runner_result", 

106 { 

107 "cmd": cmd, 

108 "retcode": retcode, 

109 "output": output, 

110 "runtime": runtime, 

111 "runtime_display": runtime_display, 

112 }, 

113 subject=subject, 

114 ) 

115 

116 # exit with failure code if applicable 

117 if retcode and keep_exit_code: 

118 sys.exit(retcode)