twaugh / journal-brief / 298

Committed 30 Jul 2020 - 12:07 coverage: 95.979%. Remained the same

Build # 298

Build Type

Pull #60

travis-ci

Committed by

web-flow

Commit Message

Merge 76499e229 into 154e39e06

Pull Request Pull Request #60: Use MIME-formatted email for both delivery modes

Run Details

47 of 50 new or added lines in 2 files covered. (94.0%)

907 of 945 relevant lines covered (95.98%)

1.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.58

/journal_brief/config.py

"""
Copyright (c) 2015, 2020 Tim Waugh <tim@cyberelk.net>

## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.

## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.

## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""

import errno
from journal_brief import list_formatters
from journal_brief.constants import CONFIG_DIR, PACKAGE, PRIORITY_MAP
from logging import getLogger
import os
import re
import sre_constants
import yaml


log = getLogger(__name__)


class ConfigError(Exception):
    pass


class SyntaxError(ConfigError):
    def __init__(self, config_file, scanner_error):
        super(SyntaxError, self).__init__(scanner_error.problem)
        self.scanner_error = scanner_error
        with open(config_file, 'rt') as cfp:
            self.config_lines = cfp.readlines()

    def __str__(self):
        mark = self.scanner_error.problem_mark
        ret = ('{sev}: {problem}\n'
               '  in "{file}", line {line}, column {column}:\n'.format(
                   sev='error',
                   problem=self.scanner_error.problem,
                   file=mark.name,
                   line=mark.line,
                   column=mark.column))
        assert mark.line > 0
        index = mark.line - 1
        assert index < len(self.config_lines)
        ret += self.config_lines[index]
        assert mark.column > 0
        ret += ' ' * (mark.column - 1) + '^'
        return ret


class SemanticError(ConfigError):
    def __init__(self, message, item, conf, index=None):
        super(SemanticError, self).__init__(message)
        self.message = message
        self.item = item
        self.conf = conf
        self.index = index

    def __str__(self):
        conf = yaml.dump(self.conf,
                         indent=2,
                         default_flow_style=False)
        if self.index is None:
            at = ''
        else:
            at = 'at item {index}, '.format(index=self.index)
            if self.index > 0:
                conflines = conf.split('\n')
                conflines.insert(1, '(...)')
                conf = '\n'.join(conflines)

        return "{sev}: {item}: {message}\n  {at}in:\n{conf}".format(
            sev='error',
            item=self.item,
            message=self.message,
            at=at,
            conf=conf
        )


def load_config(config_file):
    try:
        with open(config_file) as config_fp:
            try:
                config = yaml.safe_load(config_fp)
            except yaml.scanner.ScannerError as scanner_error:
                err = SyntaxError(config_file,
                                  scanner_error)
                log.error(err)
                raise err from scanner_error

        if not config:
            config = {}

        return config
    except IOError as ex:
        if ex.errno != errno.ENOENT:
            raise

        return {}


class Config(dict):
    ALLOWED_KEYWORDS = {
        'cursor-file',
        'debug',
        'exclusions',
        'inclusions',
        'output',
        'priority',
        'email',
    }

    def __init__(self, config_file=None):
        if not config_file:
            conf_filename = '{0}.conf'.format(PACKAGE)
            config_file = os.path.join(CONFIG_DIR, conf_filename)

        default_config = {'cursor-file': 'cursor'}
        config = load_config(config_file)

        if not isinstance(config, dict):
            error = SemanticError('must be a map', 'top level', config)
            log.error(error)
            raise error

        default_config.update(config)
        super(Config, self).__init__(default_config)
        exceptions = list(self.validate())
        for exception in exceptions:
            log.error("%s", exception)
        if exceptions:
            raise exceptions[0]

    def validate(self):
        valid_prios = [prio for prio in PRIORITY_MAP.keys()
                       if isinstance(prio, str)]
        valid_prios.sort()
        for errors in [self.validate_allowed_keywords(),
                       self.validate_cursor_file(),
                       self.validate_debug(),
                       self.validate_inclusions_or_exclusions(valid_prios,
                                                              'exclusions'),
                       self.validate_inclusions_or_exclusions(valid_prios,
                                                              'inclusions'),
                       self.validate_output(),
                       self.validate_priority(valid_prios),
                       self.validate_email()]:
            for error in errors:
                yield error

    def validate_allowed_keywords(self):
        for unexpected_key in set(self) - self.ALLOWED_KEYWORDS:
            yield SemanticError('unexpected keyword', unexpected_key,
                                {unexpected_key: self[unexpected_key]})

    def validate_cursor_file(self):
        if 'cursor-file' not in self:
            return

        if not (isinstance(self['cursor-file'], str) or
                isinstance(self['cursor-file'], int)):
            yield SemanticError('expected string', 'cursor-file',
                                {'cursor-file': self['cursor-file']})

    def validate_debug(self):
        if 'debug' not in self:
            return

        if not (isinstance(self['debug'], bool) or
                isinstance(self['debug'], int)):
            yield SemanticError('expected bool', 'debug',
                                {'debug': self['debug']})

    def validate_email(self):
        ALLOWED_EMAIL_KEYWORDS = {
            'bcc',
            'cc',
            'command',
            'from',
            'headers',
            'smtp',
            'subject',
            'suppress_empty',
            'to',
        }

        ALLOWED_SMTP_KEYWORDS = {
            'host',
            'password',
            'port',
            'starttls',
            'user',
        }

        DISALLOWED_HEADERS = {
            'From'.casefold(),
            'To'.casefold(),
            'Cc'.casefold(),
            'Bcc'.casefold(),
        }

        if 'email' not in self:
            return

        email = self.get('email')

        if not isinstance(email, dict):
            yield SemanticError('must be a map', 'email',
                                {'email': email})
            return

        for unexpected_key in set(email) - ALLOWED_EMAIL_KEYWORDS:
            yield SemanticError('unexpected \'email\' keyword', unexpected_key,
                                {unexpected_key: email[unexpected_key]})

        if 'suppress_empty' in email:
            if not (isinstance(email['suppress_empty'], bool)
                    or isinstance(email['suppress_empty'], int)):
                yield SemanticError('expected bool', 'suppress_empty',
                                    {'email': {'suppress_empty': email['suppress_empty']}})
        else:
            email['suppress_empty'] = True

        if ('smtp' in email and 'command' in email):
            yield SemanticError('cannot specify both smtp and command', 'email',
                                {'email':
                                 {'command': email['command'],
                                  'smtp': email['smtp']}})

        if not ('smtp' in email or 'command' in email):
            yield SemanticError('either smtp or command must be specified', 'email',
                                {'email': email})

        if ('command' in email and
                not isinstance(email['command'], str)):
            yield SemanticError('expected string', 'command',
                                {'email': {'command': email['command']}})

        if 'smtp' in email:
            smtp = email['smtp']

            if not isinstance(smtp, dict):
                yield SemanticError('must be a map', 'smtp',
                                    {'email': {'smtp': smtp}})
                return

            for unexpected_key in set(smtp) - ALLOWED_SMTP_KEYWORDS:
                yield SemanticError('unexpected \'smtp\' keyword', unexpected_key,
                                    {unexpected_key: smtp[unexpected_key]})

            if ('host' in smtp and
                    not isinstance(smtp['host'], str)):
                yield SemanticError('expected string', 'host',
                                    {'smtp': {'host': smtp['host']}})

            if ('port' in smtp and
                    not isinstance(smtp['port'], int)):
                yield SemanticError('expected int', 'port',
                                    {'smtp': {'port': smtp['port']}})

            if ('starttls' in smtp and
                not (isinstance(smtp['starttls'], bool) or
                     isinstance(smtp['starttls'], int))):
                yield SemanticError('expected bool', 'starttls',
                                    {'smtp': {'starttls': smtp['starttls']}})

            if ('user' in smtp and
                    not isinstance(smtp['user'], str)):
                yield SemanticError('expected string', 'user',
                                    {'smtp': {'user': smtp['user']}})

            if ('password' in smtp and
                    not isinstance(smtp['password'], str)):
                yield SemanticError('expected string', 'password',
                                    {'smtp': {'password': smtp['password']}})

        if 'from' not in email:
            yield SemanticError('\'email\' map must include \'from\'', 'email',
                                {'email': email})
        else:
            if not isinstance(email['from'], str):
                yield SemanticError('expected string', 'from',
                                    {'email': {'from': email['from']}})

        if 'to' not in email:
            yield SemanticError('\'email\' map must include \'to\'', 'email',
                                {'email': email})
        else:
            if isinstance(email['to'], list):
                pass
            elif isinstance(email['to'], str):
                email['to'] = [email['to']]
            else:
                yield SemanticError('expected list or string', 'to',
                                    {'email': {'to': email['to']}})

        if 'cc' in email:
            if isinstance(email['cc'], list):
                pass
            elif isinstance(email['cc'], str):
                email['cc'] = [email['cc']]
            else:
                yield SemanticError('expected list or string', 'cc',
                                    {'email': {'cc': email['cc']}})

        if 'bcc' in email:
            if isinstance(email['bcc'], list):
                pass
            elif isinstance(email['bcc'], str):
                email['bcc'] = [email['bcc']]
            else:
                yield SemanticError('expected list or string', 'bcc',
                                    {'email': {'bcc': email['bcc']}})

        if ('subject' in email and
                not isinstance(email['subject'], str)):
            yield SemanticError('expected string', 'subject',
                                {'email': {'subject': email['subject']}})

        if 'headers' in email:
            if not isinstance(email['headers'], dict):
                yield SemanticError('expected dict', 'headers',
                                    {'email': {'headers': email['headers']}})
            else:
                for key in email['headers'].keys():
                    if key.casefold() in DISALLOWED_HEADERS:
                        yield SemanticError("Header " + key + " cannot not be specified here", 'headers',
                                            {'email': {'headers': email['headers']}})

    def validate_output(self):
        if 'output' not in self:
            return

        formatters = list_formatters()
        output = self['output']
        if isinstance(output, list):
            outputs = output
        else:
            outputs = [output]
            self['output'] = outputs

        for output in outputs:
            if output not in formatters:
                yield SemanticError('invalid output format, must be in %s' %
                                    formatters, output,
                                    {'output': self['output']})

    def validate_priority(self, valid_prios):
        if 'priority' not in self:
            return

        try:
            valid = self['priority'] in PRIORITY_MAP
        except TypeError:
            valid = False
        finally:
            if not valid:
                yield SemanticError('invalid priority, must be in %s' %
                                    valid_prios, 'priority',
                                    {'priority': self['priority']})

    def validate_inclusions_or_exclusions(self, valid_prios, key):
        if key not in self:
            return

        if not isinstance(self[key], list):
            yield SemanticError('must be a list', key,
                                {key: self[key]})
            return

        for error in self.find_bad_rules(valid_prios, key):
            yield error

    def priority_rule_is_valid(self, values):
        try:
            if isinstance(values, list):
                valid = all(value in PRIORITY_MAP
                            for value in values)
            else:
                valid = values in PRIORITY_MAP
        except TypeError:
            valid = False
        finally:
            return valid

    def find_bad_rule_values(self, key, index, field, values):
        for value in values:
            if isinstance(value, list) or isinstance(value, dict):
                yield SemanticError('must be a string', value,
                                    {key: [{field: values}]},
                                    index=index)
                continue

            if (key == 'exclusions' and
                    isinstance(value, str) and
                    value.startswith('/') and
                    value.endswith('/')):
                try:
                    # TODO: use this computed value
                    re.compile(value[1:-1])
                except sre_constants.error as ex:
                    yield SemanticError(ex.args[0], value,
                                        {key: [{field: values}]},
                                        index=index)

    def find_bad_rules(self, valid_prios, key):
        log.debug("%s:", key)
        for index, rule in enumerate(self[key]):
            if not isinstance(rule, dict):
                yield SemanticError('must be a map', key, {key: [rule]}, index)
                continue

            log.debug('-')
            for field, values in rule.items():
                log.debug("%s: %r", field, values)
                if field == 'PRIORITY':
                    if not self.priority_rule_is_valid(values):
                        message = ('must be list or priority (%s)' %
                                   valid_prios)
                        yield SemanticError(message, field,
                                            {key: [{field: values}]},
                                            index=index)

                    continue

                if not isinstance(values, list):
                    yield SemanticError('must be a list', field,
                                        {key: [{field: values}]},
                                        index=index)
                    continue

                for error in self.find_bad_rule_values(key, index,
                                                       field, values):
                    yield error

1	"""
2	Copyright (c) 2015, 2020 Tim Waugh <tim@cyberelk.net>
3
4	## This program is free software; you can redistribute it and/or modify
5	## it under the terms of the GNU General Public License as published by
6	## the Free Software Foundation; either version 2 of the License, or
7	## (at your option) any later version.
8
9	## This program is distributed in the hope that it will be useful,
10	## but WITHOUT ANY WARRANTY; without even the implied warranty of
11	## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	## GNU General Public License for more details.
13
14	## You should have received a copy of the GNU General Public License
15	## along with this program; if not, write to the Free Software
16	## Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
17	"""
18
19	import errno	2×
20	from journal_brief import list_formatters	2×
21	from journal_brief.constants import CONFIG_DIR, PACKAGE, PRIORITY_MAP	2×
22	from logging import getLogger	2×
23	import os	2×
24	import re	2×
25	import sre_constants	2×
26	import yaml	2×
27
28
29	log = getLogger(__name__)	2×
30
31
32	class ConfigError(Exception):	2×
33	pass	2×
34
35
36	class SyntaxError(ConfigError):	2×
37	def __init__(self, config_file, scanner_error):	2×
38	super(SyntaxError, self).__init__(scanner_error.problem)	2×
39	self.scanner_error = scanner_error	2×
40	with open(config_file, 'rt') as cfp:	2×
41	self.config_lines = cfp.readlines()	2×
42
43	def __str__(self):	2×
44	mark = self.scanner_error.problem_mark	2×
45	ret = ('{sev}: {problem}\n'	2×
46	' in "{file}", line {line}, column {column}:\n'.format(
47	sev='error',
48	problem=self.scanner_error.problem,
49	file=mark.name,
50	line=mark.line,
51	column=mark.column))
52	assert mark.line > 0	2×
53	index = mark.line - 1	2×
54	assert index < len(self.config_lines)	2×
55	ret += self.config_lines[index]	2×
56	assert mark.column > 0	2×
57	ret += ' ' * (mark.column - 1) + '^'	2×
58	return ret	2×
59
60
61	class SemanticError(ConfigError):	2×
62	def __init__(self, message, item, conf, index=None):	2×
63	super(SemanticError, self).__init__(message)	2×
64	self.message = message	2×
65	self.item = item	2×
66	self.conf = conf	2×
67	self.index = index	2×
68
69	def __str__(self):	2×
70	conf = yaml.dump(self.conf,	2×
71	indent=2,
72	default_flow_style=False)
73	if self.index is None:	2×
74	at = ''	2×
75	else:
76	at = 'at item {index}, '.format(index=self.index)	2×
77	if self.index > 0:	2×
78	conflines = conf.split('\n')	2×
79	conflines.insert(1, '(...)')	2×
80	conf = '\n'.join(conflines)	2×
81
82	return "{sev}: {item}: {message}\n {at}in:\n{conf}".format(	2×
83	sev='error',
84	item=self.item,
85	message=self.message,
86	at=at,
87	conf=conf
88	)
89
90
91	def load_config(config_file):	2×
92	try:	2×
93	with open(config_file) as config_fp:	2×
94	try:	2×
95	config = yaml.safe_load(config_fp)	2×
96	except yaml.scanner.ScannerError as scanner_error:	2×
97	err = SyntaxError(config_file,	2×
98	scanner_error)
99	log.error(err)	2×
100	raise err from scanner_error	2×
101
102	if not config:	2×
103	config = {}	!
104
105	return config	2×
106	except IOError as ex:	2×
107	if ex.errno != errno.ENOENT:	2×
108	raise	!
109
110	return {}	2×
111
112
113	class Config(dict):	2×
114	ALLOWED_KEYWORDS = {	2×
115	'cursor-file',
116	'debug',
117	'exclusions',
118	'inclusions',
119	'output',
120	'priority',
121	'email',
122	}
123
124	def __init__(self, config_file=None):	2×
125	if not config_file:	2×
126	conf_filename = '{0}.conf'.format(PACKAGE)	2×
127	config_file = os.path.join(CONFIG_DIR, conf_filename)	2×
128
129	default_config = {'cursor-file': 'cursor'}	2×
130	config = load_config(config_file)	2×
131
132	if not isinstance(config, dict):	2×
133	error = SemanticError('must be a map', 'top level', config)	2×
134	log.error(error)	2×
135	raise error	2×
136
137	default_config.update(config)	2×
138	super(Config, self).__init__(default_config)	2×
139	exceptions = list(self.validate())	2×
140	for exception in exceptions:	2×
141	log.error("%s", exception)	2×
142	if exceptions:	2×
143	raise exceptions[0]	2×
144
145	def validate(self):	2×
146	valid_prios = [prio for prio in PRIORITY_MAP.keys()	2×
147	if isinstance(prio, str)]
148	valid_prios.sort()	2×
149	for errors in [self.validate_allowed_keywords(),	2×
150	self.validate_cursor_file(),
151	self.validate_debug(),
152	self.validate_inclusions_or_exclusions(valid_prios,
153	'exclusions'),
154	self.validate_inclusions_or_exclusions(valid_prios,
155	'inclusions'),
156	self.validate_output(),
157	self.validate_priority(valid_prios),
158	self.validate_email()]:
159	for error in errors:	2×
160	yield error	2×
161
162	def validate_allowed_keywords(self):	2×
163	for unexpected_key in set(self) - self.ALLOWED_KEYWORDS:	2×
164	yield SemanticError('unexpected keyword', unexpected_key,	2×
165	{unexpected_key: self[unexpected_key]})
166
167	def validate_cursor_file(self):	2×
168	if 'cursor-file' not in self:	2×
169	return	!
170
171	if not (isinstance(self['cursor-file'], str) or	2×
172	isinstance(self['cursor-file'], int)):
173	yield SemanticError('expected string', 'cursor-file',	2×
174	{'cursor-file': self['cursor-file']})
175
176	def validate_debug(self):	2×
177	if 'debug' not in self:	2×
178	return	2×
179
180	if not (isinstance(self['debug'], bool) or	2×
181	isinstance(self['debug'], int)):
182	yield SemanticError('expected bool', 'debug',	2×
183	{'debug': self['debug']})
184
185	def validate_email(self):	2×
186	ALLOWED_EMAIL_KEYWORDS = {	2×
187	'bcc',
188	'cc',
189	'command',
190	'from',
191	'headers',
192	'smtp',
193	'subject',
194	'suppress_empty',
195	'to',
196	}
197
198	ALLOWED_SMTP_KEYWORDS = {	2×
199	'host',
200	'password',
201	'port',
202	'starttls',
203	'user',
204	}
205
206	DISALLOWED_HEADERS = {	2×
207	'From'.casefold(),
208	'To'.casefold(),
209	'Cc'.casefold(),
210	'Bcc'.casefold(),
211	}
212
213	if 'email' not in self:	2×
214	return	2×
215
216	email = self.get('email')	2×
217
218	if not isinstance(email, dict):	2×
219	yield SemanticError('must be a map', 'email',	!
220	{'email': email})
221	return	!
222
223	for unexpected_key in set(email) - ALLOWED_EMAIL_KEYWORDS:	2×
224	yield SemanticError('unexpected \'email\' keyword', unexpected_key,	2×
225	{unexpected_key: email[unexpected_key]})
226
227	if 'suppress_empty' in email:	2×
228	if not (isinstance(email['suppress_empty'], bool)	2×
229	or isinstance(email['suppress_empty'], int)):
230	yield SemanticError('expected bool', 'suppress_empty',	2×
231	{'email': {'suppress_empty': email['suppress_empty']}})
232	else:
233	email['suppress_empty'] = True	2×
234
235	if ('smtp' in email and 'command' in email):	2×
236	yield SemanticError('cannot specify both smtp and command', 'email',	2×
237	{'email':
238	{'command': email['command'],
239	'smtp': email['smtp']}})
240
241	if not ('smtp' in email or 'command' in email):	2×
242	yield SemanticError('either smtp or command must be specified', 'email',	2×
243	{'email': email})
244
245	if ('command' in email and	2×
246	not isinstance(email['command'], str)):
247	yield SemanticError('expected string', 'command',	2×
248	{'email': {'command': email['command']}})
249
250	if 'smtp' in email:	2×
251	smtp = email['smtp']	2×
252
253	if not isinstance(smtp, dict):	2×
254	yield SemanticError('must be a map', 'smtp',	2×
255	{'email': {'smtp': smtp}})
256	return	2×
257
258	for unexpected_key in set(smtp) - ALLOWED_SMTP_KEYWORDS:	2×
259	yield SemanticError('unexpected \'smtp\' keyword', unexpected_key,	2×
260	{unexpected_key: smtp[unexpected_key]})
261
262	if ('host' in smtp and	2×
263	not isinstance(smtp['host'], str)):
264	yield SemanticError('expected string', 'host',	2×
265	{'smtp': {'host': smtp['host']}})
266
267	if ('port' in smtp and	2×
268	not isinstance(smtp['port'], int)):
269	yield SemanticError('expected int', 'port',	2×
270	{'smtp': {'port': smtp['port']}})
271
272	if ('starttls' in smtp and	2×
273	not (isinstance(smtp['starttls'], bool) or
274	isinstance(smtp['starttls'], int))):
275	yield SemanticError('expected bool', 'starttls',	2×
276	{'smtp': {'starttls': smtp['starttls']}})
277
278	if ('user' in smtp and	2×
279	not isinstance(smtp['user'], str)):
280	yield SemanticError('expected string', 'user',	2×
281	{'smtp': {'user': smtp['user']}})
282
283	if ('password' in smtp and	2×
284	not isinstance(smtp['password'], str)):
285	yield SemanticError('expected string', 'password',	2×
286	{'smtp': {'password': smtp['password']}})
287
288	if 'from' not in email:	2×
289	yield SemanticError('\'email\' map must include \'from\'', 'email',	2×
290	{'email': email})
291	else:
292	if not isinstance(email['from'], str):	2×
NEW 293	yield SemanticError('expected string', 'from',	!
294	{'email': {'from': email['from']}})
295
296	if 'to' not in email:	2×
297	yield SemanticError('\'email\' map must include \'to\'', 'email',	2×
298	{'email': email})
299	else:
300	if isinstance(email['to'], list):	2×
301	pass	2×
302	elif isinstance(email['to'], str):	2×
303	email['to'] = [email['to']]	2×
304	else:
305	yield SemanticError('expected list or string', 'to',	2×
306	{'email': {'to': email['to']}})
307
308	if 'cc' in email:	2×
309	if isinstance(email['cc'], list):	2×
310	pass	2×
311	elif isinstance(email['cc'], str):	2×
NEW 312	email['cc'] = [email['cc']]	!
313	else:
314	yield SemanticError('expected list or string', 'cc',	2×
315	{'email': {'cc': email['cc']}})
316
317	if 'bcc' in email:	2×
318	if isinstance(email['bcc'], list):	2×
319	pass	2×
320	elif isinstance(email['bcc'], str):	2×
NEW 321	email['bcc'] = [email['bcc']]	!
322	else:
323	yield SemanticError('expected list or string', 'bcc',	2×
324	{'email': {'bcc': email['bcc']}})
325
326	if ('subject' in email and	2×
327	not isinstance(email['subject'], str)):
328	yield SemanticError('expected string', 'subject',	2×
329	{'email': {'subject': email['subject']}})
330
331	if 'headers' in email:	2×
332	if not isinstance(email['headers'], dict):	2×
333	yield SemanticError('expected dict', 'headers',	2×
334	{'email': {'headers': email['headers']}})
335	else:
336	for key in email['headers'].keys():	2×
337	if key.casefold() in DISALLOWED_HEADERS:	2×
338	yield SemanticError("Header " + key + " cannot not be specified here", 'headers',	2×
339	{'email': {'headers': email['headers']}})
340
341	def validate_output(self):	2×
342	if 'output' not in self:	2×
343	return	2×
344
345	formatters = list_formatters()	2×
346	output = self['output']	2×
347	if isinstance(output, list):	2×
348	outputs = output	2×
349	else:
350	outputs = [output]	2×
351	self['output'] = outputs	2×
352
353	for output in outputs:	2×
354	if output not in formatters:	2×
355	yield SemanticError('invalid output format, must be in %s' %	2×
356	formatters, output,
357	{'output': self['output']})
358
359	def validate_priority(self, valid_prios):	2×
360	if 'priority' not in self:	2×
361	return	2×
362
363	try:	2×
364	valid = self['priority'] in PRIORITY_MAP	2×
365	except TypeError:	2×
366	valid = False	2×
367	finally:
368	if not valid:	2×
369	yield SemanticError('invalid priority, must be in %s' %	2×
370	valid_prios, 'priority',
371	{'priority': self['priority']})
372
373	def validate_inclusions_or_exclusions(self, valid_prios, key):	2×
374	if key not in self:	2×
375	return	2×
376
377	if not isinstance(self[key], list):	2×
378	yield SemanticError('must be a list', key,	2×
379	{key: self[key]})
380	return	2×
381
382	for error in self.find_bad_rules(valid_prios, key):	2×
383	yield error	2×
384
385	def priority_rule_is_valid(self, values):	2×
386	try:	2×
387	if isinstance(values, list):	2×
388	valid = all(value in PRIORITY_MAP	2×
389	for value in values)
390	else:
391	valid = values in PRIORITY_MAP	2×
392	except TypeError:	2×
393	valid = False	2×
394	finally:
395	return valid	2×
396
397	def find_bad_rule_values(self, key, index, field, values):	2×
398	for value in values:	2×
399	if isinstance(value, list) or isinstance(value, dict):	2×
400	yield SemanticError('must be a string', value,	2×
401	{key: [{field: values}]},
402	index=index)
403	continue	2×
404
405	if (key == 'exclusions' and	2×
406	isinstance(value, str) and
407	value.startswith('/') and
408	value.endswith('/')):
409	try:	2×
410	# TODO: use this computed value
411	re.compile(value[1:-1])	2×
412	except sre_constants.error as ex:	2×
413	yield SemanticError(ex.args[0], value,	2×
414	{key: [{field: values}]},
415	index=index)
416
417	def find_bad_rules(self, valid_prios, key):	2×
418	log.debug("%s:", key)	2×
419	for index, rule in enumerate(self[key]):	2×
420	if not isinstance(rule, dict):	2×
421	yield SemanticError('must be a map', key, {key: [rule]}, index)	2×
422	continue	2×
423
424	log.debug('-')	2×
425	for field, values in rule.items():	2×
426	log.debug("%s: %r", field, values)	2×
427	if field == 'PRIORITY':	2×
428	if not self.priority_rule_is_valid(values):	2×
429	message = ('must be list or priority (%s)' %	2×
430	valid_prios)
431	yield SemanticError(message, field,	2×
432	{key: [{field: values}]},
433	index=index)
434
435	continue	2×
436
437	if not isinstance(values, list):	2×
438	yield SemanticError('must be a list', field,	2×
439	{key: [{field: values}]},
440	index=index)
441	continue	2×
442
443	for error in self.find_bad_rule_values(key, index,	2×
444	field, values):
445	yield error	2×

twaugh / journal-brief / 298

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous