Coverage for dak/process_commands.py: 74%

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

2# 

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

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

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

6# (at your option) any later version. 

7# 

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

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

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

11# GNU General Public License for more details. 

12# 

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

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

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

16 

17import datetime 

18import os 

19import sys 

20import time 

21 

22import apt_pkg 

23 

24from daklib.command import CommandError, CommandFile 

25from daklib.config import Config 

26from daklib.daklog import Logger 

27from daklib.fstransactions import FilesystemTransaction 

28from daklib.utils import find_next_free 

29 

30 

31def usage() -> None: 

32 print( 

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

34 

35process command files 

36""" 

37 ) 

38 

39 

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

41 if argv is None: 41 ↛ 44line 41 didn't jump to line 44

42 argv = sys.argv 

43 

44 arguments = [ 

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

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

47 ] 

48 

49 cnf = Config() 

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

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

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

53 

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

55 usage() 

56 sys.exit(0) 

57 

58 log = Logger("command") 

59 

60 donedir = os.path.join( 

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

62 ) 

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

64 

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

66 cdir = options["Directory"] 

67 filenames = [ 

68 os.path.join(cdir, fn) 

69 for fn in os.listdir(cdir) 

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

71 ] 

72 

73 for fn in filenames: 

74 basename = os.path.basename(fn) 

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

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

77 continue 

78 

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

80 data = fh.read() 

81 

82 try: 

83 command = CommandFile(basename, data, log) 

84 command.evaluate() 

85 except CommandError as e: 

86 created = os.stat(fn).st_mtime 

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

88 if too_new: 

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

90 continue 

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

92 except Exception as e: 

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

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

95 else: 

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

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

98 

99 with FilesystemTransaction() as fs: 

100 fs.unlink(fn) 

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

102 fs.commit() 

103 

104 log.close()