Coverage for dak/process_commands.py: 74%

1#! /usr/bin/env python3 

2# 

3# Copyright (C) 2012, Ansgar Burchardt <ansgar@debian.org> 

4# 

5# This program is free software; you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation; either version 2 of the License, or 

8# (at your option) any later version. 

9# 

10# This program is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

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

16# with this program; if not, write to the Free Software Foundation, Inc., 

17# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 

18 

19import datetime 

20import os 

21import sys 

22import time 

23 

24import apt_pkg 

25 

26from daklib.command import CommandError, CommandFile 

27from daklib.config import Config 

28from daklib.daklog import Logger 

29from daklib.fstransactions import FilesystemTransaction 

30from daklib.utils import find_next_free 

31 

32 

33def usage() -> None: 

34 print( 

35 """Usage: dak process-commands [-d <directory>] [<command-file>...] 

36 

37process command files 

38""" 

39 ) 

40 

41 

42def main(argv: list[str] | None = None): 

43 if argv is None: 43 ↛ 46line 43 didn't jump to line 46

44 argv = sys.argv 

45 

46 arguments = [ 

47 ("h", "help", "Process-Commands::Options::Help"), 

48 ("d", "directory", "Process-Commands::Options::Directory", "HasArg"), 

49 ] 

50 

51 cnf = Config() 

52 cnf["Process-Commands::Options::Dummy"] = "" 

53 filenames = apt_pkg.parse_commandline(cnf.Cnf, arguments, argv) # type: ignore[attr-defined] 

54 options = cnf.subtree("Process-Commands::Options") 

55 

56 if "Help" in options or (len(filenames) == 0 and "Directory" not in options): 

57 usage() 

58 sys.exit(0) 

59 

60 log = Logger("command") 

61 

62 donedir = os.path.join( 

63 cnf["Dir::Done"], datetime.datetime.now().strftime("%Y/%m/%d") 

64 ) 

65 rejectdir = cnf["Dir::Reject"] 

66 

67 if len(filenames) == 0: 67 ↛ 75line 67 didn't jump to line 75 because the condition on line 67 was always true

68 cdir = options["Directory"] 

69 filenames = [ 

70 os.path.join(cdir, fn) 

71 for fn in os.listdir(cdir) 

72 if fn.endswith(".dak-commands") 

73 ] 

74 

75 for fn in filenames: 

76 basename = os.path.basename(fn) 

77 if not fn.endswith(".dak-commands"): 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 log.log(["unexpected filename", basename]) 

79 continue 

80 

81 with open(fn, "rb") as fh: 

82 data = fh.read() 

83 

84 try: 

85 command = CommandFile(basename, data, log) 

86 command.evaluate() 

87 except CommandError as e: 

88 created = os.stat(fn).st_mtime 

89 too_new = time.time() - created < int(cnf.get("Dinstall::SkipTime", "60")) 

90 if too_new: 

91 log.log(["skipped (too new)"]) 

92 continue 

93 log.log(["reject", basename, e]) 

94 except Exception as e: 

95 log.log_traceback("Exception while processing %s:" % (basename), e) 

96 dst = find_next_free(os.path.join(rejectdir, basename)) 

97 else: 

98 log.log(["done", basename]) 

99 dst = find_next_free(os.path.join(donedir, basename)) 

100 

101 with FilesystemTransaction() as fs: 

102 fs.unlink(fn) 

103 fs.create(dst, mode=0o644, text=False).write(data) 

104 fs.commit() 

105 

106 log.close() 

107 

108 

109if __name__ == "__main__": 

110 main()