Add ipalib.messages

The messages module contains message classes that can be added
to a RPC response to provide additional information or warnings.

This patch adds only the module with a single public message,
VersionMissing, and unit tests.
Since message classes are very similar to public errors, some
functionality and unit tests were shared.

Design page: http://freeipa.org/page/V3/Messages
Ticket: https://fedorahosted.org/freeipa/ticket/2732
This commit is contained in:
Petr Viktorin
2012-12-04 09:27:05 -05:00
committed by Martin Kosek
parent 7336a176b4
commit 8af5369cba
4 changed files with 263 additions and 98 deletions

View File

@@ -102,10 +102,9 @@ current block assignments:
- **5100 - 5999** *Reserved for future use*
"""
from inspect import isclass
from text import _ as ugettext, ngettext as ungettext
from text import Gettext, NGettext
from constants import TYPE_ERROR
from ipalib.text import ngettext as ungettext
import messages
class PrivateError(StandardError):
@@ -233,10 +232,10 @@ class PluginsPackageError(PrivateError):
##############################################################################
# Public errors:
__messages = []
_texts = []
def _(message):
__messages.append(message)
_texts.append(message)
return message
@@ -244,58 +243,14 @@ class PublicError(StandardError):
"""
**900** Base class for exceptions that can be forwarded in an RPC response.
"""
def __init__(self, format=None, message=None, **kw):
messages.process_message_arguments(self, format, message, **kw)
super(PublicError, self).__init__(self.msg)
errno = 900
rval = 1
format = None
def __init__(self, format=None, message=None, **kw):
self.kw = kw
name = self.__class__.__name__
if self.format is not None and format is not None:
raise ValueError(
'non-generic %r needs format=None; got format=%r' % (
name, format)
)
if message is None:
if self.format is None:
if format is None:
raise ValueError(
'%s.format is None yet format=None, message=None' % name
)
self.format = format
self.forwarded = False
self.msg = self.format % kw
if isinstance(self.format, basestring):
self.strerror = ugettext(self.format) % kw
else:
self.strerror = self.format % kw
if 'instructions' in kw:
def convert_instructions(value):
if isinstance(value, list):
result=u'\n'.join(map(lambda line: unicode(line), value))
return result
return value
instructions = u'\n'.join((unicode(_('Additional instructions:')),
convert_instructions(kw['instructions'])))
self.strerror = u'\n'.join((self.strerror, instructions))
else:
if isinstance(message, (Gettext, NGettext)):
message = unicode(message)
elif type(message) is not unicode:
raise TypeError(
TYPE_ERROR % ('message', unicode, message, type(message))
)
self.forwarded = True
self.msg = message
self.strerror = message
for (key, value) in kw.iteritems():
assert not hasattr(self, key), 'conflicting kwarg %s.%s = %r' % (
name, key, value,
)
setattr(self, key, value)
StandardError.__init__(self, self.msg)
class VersionError(PublicError):
"""
@@ -1711,21 +1666,8 @@ class GenericError(PublicError):
def __errors_iter():
"""
Iterate through all the `PublicError` subclasses.
"""
for (key, value) in globals().items():
if key.startswith('_') or not isclass(value):
continue
if issubclass(value, PublicError):
yield value
public_errors = tuple(
sorted(__errors_iter(), key=lambda E: E.errno)
)
public_errors = tuple(sorted(
messages.iter_messages(globals(), PublicError), key=lambda E: E.errno))
if __name__ == '__main__':
for klass in public_errors:
print '%d\t%s' % (klass.errno, klass.__name__)
print '(%d public errors)' % len(public_errors)
messages.print_report('public errors', public_errors)

151
ipalib/messages.py Normal file
View File

@@ -0,0 +1,151 @@
# Authors:
# Petr Viktorin <pviktori@redhat.com>
#
# Copyright (C) 2012 Red Hat
# see file 'COPYING' for use and warranty inmsgion
#
# 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 3 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, see <http://www.gnu.org/licenses/>.
"""
Custom message (debug, info, wraning) classes passed through RPC.
These are added to the "messages" entry in a RPC response, and printed to the
user as log messages.
Each message class has a unique numeric "errno" attribute from the 10000-10999
range, so that it does not clash with PublicError numbers.
Messages also have the 'type' argument, set to one of 'debug', 'info',
'warning', 'error'. This determines the severity of themessage.
"""
from inspect import isclass
from ipalib.constants import TYPE_ERROR
from ipalib.text import _ as ugettext
from ipalib.text import Gettext, NGettext
def process_message_arguments(obj, format=None, message=None, **kw):
obj.kw = kw
name = obj.__class__.__name__
if obj.format is not None and format is not None:
raise ValueError(
'non-generic %r needs format=None; got format=%r' % (
name, format)
)
if message is None:
if obj.format is None:
if format is None:
raise ValueError(
'%s.format is None yet format=None, message=None' % name
)
obj.format = format
obj.forwarded = False
obj.msg = obj.format % kw
if isinstance(obj.format, basestring):
obj.strerror = ugettext(obj.format) % kw
else:
obj.strerror = obj.format % kw
if 'instructions' in kw:
def convert_instructions(value):
if isinstance(value, list):
result = u'\n'.join(map(lambda line: unicode(line), value))
return result
return value
instructions = u'\n'.join((unicode(_('Additional instructions:')),
convert_instructions(kw['instructions'])))
obj.strerror = u'\n'.join((obj.strerror, instructions))
else:
if isinstance(message, (Gettext, NGettext)):
message = unicode(message)
elif type(message) is not unicode:
raise TypeError(
TYPE_ERROR % ('message', unicode, message, type(message))
)
obj.forwarded = True
obj.msg = message
obj.strerror = message
for (key, value) in kw.iteritems():
assert not hasattr(obj, key), 'conflicting kwarg %s.%s = %r' % (
name, key, value,
)
setattr(obj, key, value)
_texts = []
def _(message):
_texts.append(message)
return message
class PublicMessage(UserWarning):
"""
**10000** Base class for messages that can be forwarded in an RPC response.
"""
def __init__(self, format=None, message=None, **kw):
process_message_arguments(self, format, message, **kw)
super(PublicMessage, self).__init__(self.msg)
errno = 10000
format = None
def to_dict(self):
"""Export this message to a dict that can be sent through RPC"""
return dict(
type=unicode(self.type),
name=unicode(type(self).__name__),
message=self.strerror,
code=self.errno,
)
class VersionMissing(PublicMessage):
"""
**13001** Used when client did not send the API version.
For example:
>>> VersionMissing(server_version='2.123').strerror
u"API Version number was not sent, forward compatibility not guaranteed. Assuming server's API version, 2.123"
"""
errno = 13001
type = 'warning'
format = _("API Version number was not sent, forward compatibility not "
"guaranteed. Assuming server's API version, %(server_version)s")
def iter_messages(variables, base):
"""Return a tuple with all subclasses
"""
for (key, value) in variables.items():
if key.startswith('_') or not isclass(value):
continue
if issubclass(value, base):
yield value
public_messages = tuple(sorted(
iter_messages(globals(), PublicMessage), key=lambda E: E.errno))
def print_report(label, classes):
for cls in classes:
print '%d\t%s' % (cls.errno, cls.__name__)
print '(%d %s)' % (len(classes), label)
if __name__ == '__main__':
print_report('public messages', public_messages)

View File

@@ -23,6 +23,7 @@ Test the `ipalib.errors` module.
import re
import inspect
from tests.util import assert_equal, raises
from ipalib import errors, text
from ipalib.constants import TYPE_ERROR
@@ -210,8 +211,8 @@ class PublicExceptionTester(object):
for (key, value) in kw.iteritems():
assert not hasattr(self.klass, key), key
inst = self.klass(format=format, message=message, **kw)
assert isinstance(inst, StandardError)
assert isinstance(inst, errors.PublicError)
for required_class in self.required_classes:
assert isinstance(inst, required_class)
assert isinstance(inst, self.klass)
assert not isinstance(inst, errors.PrivateError)
for (key, value) in kw.iteritems():
@@ -224,11 +225,9 @@ class test_PublicError(PublicExceptionTester):
Test the `ipalib.errors.PublicError` exception.
"""
_klass = errors.PublicError
required_classes = StandardError, errors.PublicError
def test_init(self):
"""
Test the `ipalib.errors.PublicError.__init__` method.
"""
message = u'The translated, interpolated message'
format = 'key=%(key1)r and key2=%(key2)r'
uformat = u'Translated key=%(key1)r and key2=%(key2)r'
@@ -259,8 +258,8 @@ class test_PublicError(PublicExceptionTester):
# Test with format=None, message=None
e = raises(ValueError, self.klass, **kw)
assert str(e) == \
'PublicError.format is None yet format=None, message=None'
assert (str(e) == '%s.format is None yet format=None, message=None' %
self.klass.__name__)
######################################
@@ -336,27 +335,40 @@ class test_PublicError(PublicExceptionTester):
assert_equal(list(inst_match),list(instructions))
def test_public_errors():
class BaseMessagesTest(object):
"""Generic test for all of a module's errors or messages
"""
Test the `ipalib.errors.public_errors` module variable.
"""
i = 0
for klass in errors.public_errors:
assert issubclass(klass, StandardError)
assert issubclass(klass, errors.PublicError)
assert not issubclass(klass, errors.PrivateError)
assert type(klass.errno) is int
assert 900 <= klass.errno <= 5999
doc = inspect.getdoc(klass)
assert doc is not None, 'need class docstring for %s' % klass.__name__
m = re.match(r'^\*{2}(\d+)\*{2} ', doc)
assert m is not None, "need '**ERRNO**' in %s docstring" % klass.__name__
errno = int(m.group(1))
assert errno == klass.errno, (
'docstring=%r but errno=%r in %s' % (errno, klass.errno, klass.__name__)
)
def test_public_messages(self):
i = 0
for klass in self.message_list:
for required_class in self.required_classes:
assert issubclass(klass, required_class)
assert type(klass.errno) is int
assert klass.errno in self.errno_range
doc = inspect.getdoc(klass)
assert doc is not None, 'need class docstring for %s' % klass.__name__
m = re.match(r'^\*{2}(\d+)\*{2} ', doc)
assert m is not None, "need '**ERRNO**' in %s docstring" % klass.__name__
errno = int(m.group(1))
assert errno == klass.errno, (
'docstring=%r but errno=%r in %s' % (errno, klass.errno, klass.__name__)
)
self.extratest(klass)
# Test format
if klass.format is not None:
assert klass.format is errors.__messages[i]
i += 1
# Test format
if klass.format is not None:
assert klass.format is self.texts[i]
i += 1
def extratest(self, cls):
pass
class test_PublicErrors(object):
message_list = errors.public_errors
errno_range = xrange(900, 5999)
required_classes = (StandardError, errors.PublicError)
texts = errors._texts
def extratest(self, cls):
assert not issubclass(cls, errors.PrivateError)

View File

@@ -0,0 +1,60 @@
# Authors:
# Petr Viktorin <pviktori@redhat.com>
#
# Copyright (C) 1012 Red Hat
# see file 'COPYING' for use and warranty information
#
# 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 3 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, see <http://www.gnu.org/licenses/>.
"""
Test the `ipalib.messages` module.
"""
from ipalib import messages
from tests.test_ipalib import test_errors
class HelloMessage(messages.PublicMessage):
type = 'info'
format = '%(greeting)s, %(object)s!'
errno = 1234
class test_PublicMessage(test_errors.test_PublicError):
"""Test public messages"""
# The messages are a lot like public errors; defer testing to that.
klass = messages.PublicMessage
required_classes = (UserWarning, messages.PublicMessage)
class test_PublicMessages(test_errors.BaseMessagesTest):
message_list = messages.public_messages
errno_range = xrange(10000, 19999)
required_classes = (UserWarning, messages.PublicMessage)
texts = messages._texts
def extratest(self, cls):
if cls is not messages.PublicMessage:
assert cls.type in ('debug', 'info', 'warning', 'error')
def test_to_dict():
expected = dict(
name='HelloMessage',
type='info',
message='Hello, world!',
code=1234,
)
assert HelloMessage(greeting='Hello', object='world').to_dict() == expected