Coverage for daklib/ls.py: 52%

1"""List packages according to various criteria 

2 

3@copyright: 2014, Ansgar Burchardt <ansgar@debian.org> 

4@license: GPL-2+ 

5""" 

6 

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

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

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

10# (at your option) any later version. 

11# 

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

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

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

15# GNU General Public License for more details. 

16# 

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

18# along with this program; if not, write to the Free Software 

19# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 

20 

21from collections import defaultdict 

22 

23import sqlalchemy.dialects.postgresql as pgsql 

24import sqlalchemy.sql as sql 

25 

26from daklib.dbconn import DBConn 

27 

28 

29def list_packages( 

30 packages, 

31 suites=None, 

32 components=None, 

33 architectures=None, 

34 binary_types=None, 

35 source_and_binary=False, 

36 regex=False, 

37 format=None, 

38 highest=None, 

39): 

40 session = DBConn().session() 

41 try: 

42 t = DBConn().view_package_list 

43 

44 comparison_operator = "~" if regex else "=" 

45 

46 where = sql.false() 

47 for package in packages: 

48 where = where | t.c.package.op(comparison_operator)(package) 

49 if source_and_binary: 

50 where = where | t.c.source.op(comparison_operator)(package) 

51 

52 if suites is not None: 

53 where = where & (t.c.suite.in_(suites) | t.c.codename.in_(suites)) 

54 if components is not None: 54 ↛ 55line 54 didn't jump to line 55, because the condition on line 54 was never true

55 where = where & t.c.component.in_(components) 

56 if architectures is not None: 56 ↛ 57line 56 didn't jump to line 57, because the condition on line 56 was never true

57 where = where & t.c.architecture.in_(architectures) 

58 if binary_types is not None: 58 ↛ 59line 58 didn't jump to line 59, because the condition on line 58 was never true

59 where = where & t.c.type.in_(binary_types) 

60 

61 if format is None: 61 ↛ 96line 61 didn't jump to line 96, because the condition on line 61 was never false

62 c_architectures = sql.func.string_agg( 

63 t.c.architecture, 

64 pgsql.aggregate_order_by( 

65 ", ", t.c.architecture_is_source.desc(), t.c.architecture 

66 ), 

67 ) 

68 query = ( 

69 sql.select( 

70 [t.c.package, t.c.version, t.c.display_suite, c_architectures] 

71 ) 

72 .where(where) 

73 .group_by(t.c.package, t.c.version, t.c.display_suite) 

74 .order_by(t.c.package, t.c.version, t.c.display_suite) 

75 ) 

76 result = session.execute(query).fetchall() 

77 

78 if len(result) == 0: 

79 return 

80 

81 lengths = { 

82 "package": max(10, max(len(row[t.c.package]) for row in result)), 

83 "version": max(13, max(len(row[t.c.version]) for row in result)), 

84 "suite": max(10, max(len(row[t.c.display_suite]) for row in result)), 

85 } 

86 format = "{0:{lengths[package]}} | {1:{lengths[version]}} | {2:{lengths[suite]}} | {3}" 

87 

88 for row in result: 

89 yield format.format( 

90 row[t.c.package], 

91 row[t.c.version], 

92 row[t.c.display_suite], 

93 row[c_architectures], 

94 lengths=lengths, 

95 ) 

96 elif format in ("control-suite", "heidi"): 

97 query = sql.select([t.c.package, t.c.version, t.c.architecture]).where( 

98 where 

99 ) 

100 result = session.execute(query) 

101 for row in result: 

102 yield "{0} {1} {2}".format( 

103 row[t.c.package], row[t.c.version], row[t.c.architecture] 

104 ) 

105 elif format == "python": 

106 c_architectures = sql.func.string_agg( 

107 t.c.architecture, 

108 pgsql.aggregate_order_by( 

109 ",", t.c.architecture_is_source.desc(), t.c.architecture 

110 ), 

111 ) 

112 query = ( 

113 sql.select( 

114 [ 

115 t.c.package, 

116 t.c.version, 

117 t.c.display_suite, 

118 c_architectures, 

119 t.c.source, 

120 t.c.source_version, 

121 t.c.component, 

122 ] 

123 ) 

124 .where(where) 

125 .group_by( 

126 t.c.package, 

127 t.c.version, 

128 t.c.display_suite, 

129 t.c.source, 

130 t.c.component, 

131 t.c.source_version, 

132 ) 

133 ) 

134 result = session.execute(query).fetchall() 

135 

136 if len(result) == 0: 

137 return 

138 

139 def val(): 

140 return defaultdict(val) 

141 

142 ret = val() 

143 for row in result: 

144 ret[row[t.c.package]][row[t.c.display_suite]][row[t.c.version]] = { 

145 "component": row[t.c.component], 

146 "architectures": row[c_architectures].split(","), 

147 "source": row[t.c.source], 

148 "source_version": row[t.c.source_version], 

149 } 

150 

151 yield ret 

152 return 

153 else: 

154 raise ValueError("Unknown output format requested.") 

155 

156 if highest is not None: 156 ↛ 157line 156 didn't jump to line 157

157 query = ( 

158 sql.select([t.c.package, sql.func.max(t.c.version)]) 

159 .where(where) 

160 .group_by(t.c.package) 

161 .order_by(t.c.package) 

162 ) 

163 result = session.execute(query) 

164 yield "" 

165 for row in result: 

166 yield "{0} ({1} {2})".format(row[0], highest, row[1]) 

167 finally: 

168 session.close()