Coverage for dak/export.py: 50%

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 sys 

18 

19import apt_pkg 

20from sqlalchemy import select 

21 

22from daklib.config import Config 

23from daklib.dbconn import DBChange, DBConn, PolicyQueue, PolicyQueueUpload 

24from daklib.policy import UploadCopy 

25 

26 

27def usage(): 

28 print( 

29 """Usage: dak export -q <queue> [options] -a|--all|<source...> 

30 

31Export uploads from policy queues, that is the changes files for the given 

32source package and all other files associated with that. 

33 

34 -a --all export all uploads 

35 -c --copy copy files instead of symlinking them 

36 -d <directory> target directory to export packages to 

37 default: current directory 

38 -q <queue> queue to grab uploads from 

39 <source> source package name to export 

40""" 

41 ) 

42 

43 

44def main(argv=None): 

45 if argv is None: 45 ↛ 48line 45 didn't jump to line 48

46 argv = sys.argv 

47 

48 arguments = [ 

49 ("h", "help", "Export::Options::Help"), 

50 ("a", "all", "Export::Options::All"), 

51 ("c", "copy", "Export::Options::Copy"), 

52 ("d", "directory", "Export::Options::Directory", "HasArg"), 

53 ("q", "queue", "Export::Options::Queue", "HasArg"), 

54 ] 

55 

56 cnf = Config() 

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

58 options = cnf.subtree("Export::Options") 

59 

60 if "Help" in options or "Queue" not in options: 60 ↛ 64line 60 didn't jump to line 64 because the condition on line 60 was always true

61 usage() 

62 sys.exit(0) 

63 

64 session = DBConn().session() 

65 

66 queue = session.scalars( 

67 select(PolicyQueue).filter_by(queue_name=options["Queue"]).limit(1) 

68 ).first() 

69 if queue is None: 

70 print("Unknown queue '{0}'".format(options["Queue"])) 

71 sys.exit(1) 

72 uploads = select(PolicyQueueUpload).filter_by(policy_queue=queue) 

73 if "All" not in options: 

74 uploads = uploads.join(PolicyQueueUpload.changes).where( 

75 DBChange.source.in_(source_names) 

76 ) 

77 directory = options.get("Directory", ".") 

78 symlink = "Copy" not in options 

79 

80 for u in session.scalars(uploads): 

81 UploadCopy(u).export(directory, symlink=symlink, ignore_existing=True)