1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

# Copyright 2013 The Distro Tracker Developers 

# See the COPYRIGHT file at the top-level directory of this distribution and 

# at https://deb.li/DTAuthors 

# 

# This file is part of Distro Tracker. It is subject to the license terms 

# in the LICENSE file found in the top-level directory of this 

# distribution and at https://deb.li/DTLicense. No part of Distro Tracker, 

# including this file, may be copied, modified, propagated, or distributed 

# except according to the terms contained in the LICENSE file. 

 

""" 

Defines and implements all Distro Tracker control commands. 

""" 

 

import inspect 

import sys 

 

from django.conf import settings 

 

from distro_tracker.core.utils import distro_tracker_render_to_string 

from distro_tracker.mail.control.commands.base import Command 

from distro_tracker.mail.control.commands.keywords import ( # noqa 

ViewDefaultKeywordsCommand, 

ViewPackageKeywordsCommand, 

SetDefaultKeywordsCommand, 

SetPackageKeywordsCommand, 

) 

from distro_tracker.mail.control.commands.teams import ( # noqa 

JoinTeam, 

LeaveTeam, 

ListTeamPackages, 

WhichTeams, 

) 

from distro_tracker.mail.control.commands.misc import ( # noqa 

SubscribeCommand, 

UnsubscribeCommand, 

WhichCommand, 

WhoCommand, 

QuitCommand, 

UnsubscribeallCommand, 

) 

from distro_tracker.mail.control.commands.confirmation import ( # noqa 

ConfirmCommand 

) 

 

MAX_ALLOWED_ERRORS = settings.DISTRO_TRACKER_MAX_ALLOWED_ERRORS_CONTROL_COMMANDS 

 

 

class HelpCommand(Command): 

""" 

Displays help for all the other commands -- their description. 

""" 

META = { 

'description': '''help 

Shows all available commands''', 

'name': 'help', 

'position': 5, 

} 

 

REGEX_LIST = ( 

r'$', 

) 

 

def handle(self): 

self.reply(distro_tracker_render_to_string('control/help.txt', { 

'descriptions': [ 

command.META.get('description', '') 

for command in UNIQUE_COMMANDS 

], 

})) 

 

 

UNIQUE_COMMANDS = sorted( 

(klass 

for _, klass in inspect.getmembers(sys.modules[__name__], inspect.isclass) 

if klass != Command and issubclass(klass, Command)), 

key=lambda cmd: cmd.META.get('position', float('inf')) 

) 

""" 

A list of all :py:class:`Command` that are defined. 

""" 

 

 

class CommandFactory(object): 

""" 

Creates instances of 

:py:class:`Command <distro_tracker.mail.control.commands.base.Command>` 

classes based on the given context. 

 

Context is used to fill in parameters when the command has not found 

it in the given command line. 

""" 

def __init__(self, context): 

#: A dict which is used to fill in parameters' values when they are not 

#: found in the command line. 

self.context = context 

 

def get_command_function(self, line): 

""" 

Returns a function which executes the functionality of the command 

which corresponds to the given arguments. 

 

:param line: The line for which a command function should be returned. 

:type line: string 

 

:returns: A callable which when called executes the functionality of a 

command matching the given line. 

:rtype: :py:class:`Command 

<distro_tracker.mail.control.commands.base.Command>` subclass 

""" 

for cmd in UNIQUE_COMMANDS: 

# Command exists 

match = cmd.match_line(line) 

if not match: 

continue 

kwargs = match.groupdict() 

if not kwargs: 

# No named patterns found, pass them in the order they were 

# matched. 

args = match.groups() 

return cmd(*args) 

else: 

# Update the arguments which weren't matched from the given 

# context, if available. 

kwargs.update({ 

key: value 

for key, value in self.context.items() 

if key in kwargs and not kwargs[key] and value 

}) 

command = cmd(**kwargs) 

command.context = dict(self.context.items()) 

return command 

 

 

class CommandProcessor(object): 

""" 

A class which performs command processing. 

""" 

def __init__(self, factory, confirmed=False): 

""" 

:param factory: Used to obtain 

:py:class:`Command 

<distro_tracker.mail.control.commands.base.Command>` instances 

from command text which is processed. 

:type factory: :py:class`CommandFactory` instance 

:param confirmed: Indicates whether the commands being executed have 

already been confirmed or if those which require confirmation will 

be added to the set of commands requiring confirmation. 

:type confirmed: Boolean 

""" 

self.factory = factory 

self.confirmed = confirmed 

self.confirmation_set = None 

 

self.out = [] 

self.errors = 0 

self.processed = set() 

 

def echo_command(self, line): 

""" 

Echoes the line to the command processing output. The line is quoted in 

the output. 

 

:param line: The line to be echoed back to the output. 

""" 

self.out.append('> ' + line) 

 

def output(self, text): 

""" 

Include the given line in the command processing output. 

 

:param line: The line of text to be included in the output. 

""" 

self.out.append(text) 

 

def run_command(self, command): 

""" 

Runs the given command. 

 

:param command: The command to be ran. 

:type command: :py:class:`Command 

<distro_tracker.mail.control.commands.base.Command>` 

""" 

if command.get_command_text() not in self.processed: 

# Only process the command if it was not previously processed. 

if getattr(command, 'needs_confirmation', False): 

command.is_confirmed = self.confirmed 

command.confirmation_set = self.confirmation_set 

# Now run the command 

command_output = command() 

191 ↛ 192line 191 didn't jump to line 192, because the condition on line 191 was never true if not command_output: 

command_output = '' 

self.output(command_output) 

self.processed.add(command.get_command_text()) 

 

def process(self, lines): 

""" 

Processes all the given lines of text which are interpreted as 

commands. 

 

:param lines: A list of strings each representing a single line which 

is to be regarded as a command. 

""" 

204 ↛ 205line 204 didn't jump to line 205, because the condition on line 204 was never true if self.errors == MAX_ALLOWED_ERRORS: 

return 

 

for line in lines: 

line = line.strip() 

self.echo_command(line) 

 

if not line or line.startswith('#'): 

continue 

command = self.factory.get_command_function(line) 

 

if not command: 

self.errors += 1 

if self.errors == MAX_ALLOWED_ERRORS: 

self.output( 

'{MAX_ALLOWED_ERRORS} lines ' 

'without commands: stopping.'.format( 

MAX_ALLOWED_ERRORS=MAX_ALLOWED_ERRORS)) 

return 

else: 

self.run_command(command) 

 

if isinstance(command, QuitCommand): 

return 

 

def is_success(self): 

""" 

Checks whether any command was successfully processed. 

 

:returns True: when at least one command is successfully executed 

:returns False: when no commands were successfully executed 

:rtype: Boolean 

""" 

# Send a response only if there were some commands processed 

if self.processed: 

return True 

else: 

return False 

 

def get_output(self): 

""" 

Returns the resulting output of processing all given commands. 

 

:rtype: string 

""" 

return '\n'.join(self.out)