Petr Viktorin
24bca144a8
The API version the client sends can now be used to check what the client expects or is capable of. All version tests IPA does will be be named and listed in one module, ipalib.capabilities, which includes a function to test a specific capability against an API version. Similarly to Python's __future__ module, capabilities.py also serves as documentation of backwards-incompatible changes to the API. The first capability to be defined is "messages". Recent enough clients can accept a list of warnings or other info under the "messages" key in the result dict. If a JSON client does not send the API version, it is assumed this is a testing client (e.g. curl from the command line). Such a client "has" all capabilities, but it will always receive a warning mentioning that forward compatibility is not guaranteed. If a XML client does not send the API version, it is assumed it uses the API version before capabilities were introduced. (This is to keep backwards compatibility with clients containing bug https://fedorahosted.org/freeipa/ticket/3294) Whenever a capability is added, the API version must be incremented. To ensure that, capabilities are written to API.txt and checked by `makeapi --validate`. Design page: http://freeipa.org/page/V3/Messages Ticket: https://fedorahosted.org/freeipa/ticket/2732
323 lines
10 KiB
Python
323 lines
10 KiB
Python
# Authors:
|
|
# Rob Crittenden <rcritten@redhat.com>
|
|
#
|
|
# Copyright (C) 2008 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/>.
|
|
|
|
"""
|
|
Base class for all XML-RPC tests
|
|
"""
|
|
|
|
import sys
|
|
import socket
|
|
import nose
|
|
from tests.util import assert_deepequal, Fuzzy
|
|
from ipalib import api, request, errors
|
|
from ipalib.x509 import valid_issuer
|
|
from ipapython.version import API_VERSION
|
|
|
|
|
|
# Matches a gidnumber like '1391016742'
|
|
# FIXME: Does it make more sense to return gidnumber, uidnumber, etc. as `int`
|
|
# or `long`? If not, we still need to return them as `unicode` instead of `str`.
|
|
fuzzy_digits = Fuzzy('^\d+$', type=basestring)
|
|
|
|
uuid_re = '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}'
|
|
|
|
# Matches an ipauniqueid like u'784d85fd-eae7-11de-9d01-54520012478b'
|
|
fuzzy_uuid = Fuzzy('^%s$' % uuid_re)
|
|
|
|
# Matches trusted domain GUID, like u'463bf2be-3456-4a57-979e-120304f2a0eb'
|
|
fuzzy_guid = fuzzy_uuid
|
|
|
|
# Matches SID of a trusted domain
|
|
# SID syntax: http://msdn.microsoft.com/en-us/library/ff632068.aspx
|
|
_sid_identifier_authority = '(0x[0-9a-f]{1,12}|[0-9]{1,10})'
|
|
fuzzy_domain_sid = Fuzzy(
|
|
'^S-1-5-21-%(idauth)s-%(idauth)s-%(idauth)s$' % dict(idauth=_sid_identifier_authority)
|
|
)
|
|
|
|
# Matches netgroup dn. Note (?i) at the beginning of the regexp is the ingnore case flag
|
|
fuzzy_netgroupdn = Fuzzy(
|
|
'(?i)ipauniqueid=%s,cn=ng,cn=alt,%s' % (uuid_re, api.env.basedn)
|
|
)
|
|
|
|
# Matches sudocmd dn
|
|
fuzzy_sudocmddn = Fuzzy(
|
|
'(?i)ipauniqueid=%s,cn=sudocmds,cn=sudo,%s' % (uuid_re, api.env.basedn)
|
|
)
|
|
|
|
# Matches a hash signature, not enforcing length
|
|
fuzzy_hash = Fuzzy('^([a-f0-9][a-f0-9]:)+[a-f0-9][a-f0-9]$', type=basestring)
|
|
|
|
# Matches a date, like Tue Apr 26 17:45:35 2016 UTC
|
|
fuzzy_date = Fuzzy('^[a-zA-Z]{3} [a-zA-Z]{3} \d{2} \d{2}:\d{2}:\d{2} \d{4} UTC$')
|
|
|
|
fuzzy_issuer = Fuzzy(type=basestring, test=lambda issuer: valid_issuer(issuer))
|
|
|
|
fuzzy_hex = Fuzzy('^0x[0-9a-fA-F]+$', type=basestring)
|
|
|
|
# Matches password - password consists of all printable characters without whitespaces
|
|
# The only exception is space, but space cannot be at the beggingin or end of the pwd
|
|
fuzzy_password = Fuzzy('^\S([\S ]*\S)*$')
|
|
|
|
# Matches generalized time value. Time format is: %Y%m%d%H%M%SZ
|
|
fuzzy_dergeneralizedtime = Fuzzy('^[0-9]{14}Z$')
|
|
|
|
# match any string
|
|
fuzzy_string = Fuzzy(type=basestring)
|
|
|
|
try:
|
|
if not api.Backend.xmlclient.isconnected():
|
|
api.Backend.xmlclient.connect(fallback=False)
|
|
res = api.Command['user_show'](u'notfound')
|
|
except errors.NetworkError:
|
|
server_available = False
|
|
except IOError:
|
|
server_available = False
|
|
except errors.NotFound:
|
|
server_available = True
|
|
|
|
|
|
|
|
def assert_attr_equal(entry, key, value):
|
|
if type(entry) is not dict:
|
|
raise AssertionError(
|
|
'assert_attr_equal: entry must be a %r; got a %r: %r' % (
|
|
dict, type(entry), entry)
|
|
)
|
|
if key not in entry:
|
|
raise AssertionError(
|
|
'assert_attr_equal: entry has no key %r: %r' % (key, entry)
|
|
)
|
|
if value not in entry[key]:
|
|
raise AssertionError(
|
|
'assert_attr_equal: %r: %r not in %r' % (key, value, entry[key])
|
|
)
|
|
|
|
|
|
def assert_is_member(entry, value, key='member'):
|
|
if type(entry) is not dict:
|
|
raise AssertionError(
|
|
'assert_is_member: entry must be a %r; got a %r: %r' % (
|
|
dict, type(entry), entry)
|
|
)
|
|
if key not in entry:
|
|
raise AssertionError(
|
|
'assert_is_member: entry has no key %r: %r' % (key, entry)
|
|
)
|
|
for member in entry[key]:
|
|
if member.startswith(value):
|
|
return
|
|
raise AssertionError(
|
|
'assert_is_member: %r: %r not in %r' % (key, value, entry[key])
|
|
)
|
|
|
|
|
|
# Initialize the API. We do this here so that one can run the tests
|
|
# individually instead of at the top-level. If API.bootstrap()
|
|
# has already been called we continue gracefully. Other errors will be
|
|
# raised.
|
|
|
|
class XMLRPC_test(object):
|
|
"""
|
|
Base class for all XML-RPC plugin tests
|
|
"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if not server_available:
|
|
raise nose.SkipTest('%r: Server not available: %r' %
|
|
(cls.__module__, api.env.xmlrpc_uri))
|
|
|
|
def setUp(self):
|
|
if not api.Backend.xmlclient.isconnected():
|
|
api.Backend.xmlclient.connect(fallback=False)
|
|
|
|
def tearDown(self):
|
|
"""
|
|
nose tear-down fixture.
|
|
"""
|
|
request.destroy_context()
|
|
|
|
def failsafe_add(self, obj, pk, **options):
|
|
"""
|
|
Delete possible leftover entry first, then add.
|
|
|
|
This helps speed us up when a partial test failure has left LDAP in a
|
|
dirty state.
|
|
|
|
:param obj: An Object like api.Object.user
|
|
:param pk: The primary key of the entry to be created
|
|
:param options: Kwargs to be passed to obj.add()
|
|
"""
|
|
try:
|
|
obj.methods['del'](pk)
|
|
except errors.NotFound:
|
|
pass
|
|
return obj.methods['add'](pk, **options)
|
|
|
|
|
|
IGNORE = """Command %r is missing attribute %r in output entry.
|
|
args = %r
|
|
options = %r
|
|
entry = %r"""
|
|
|
|
|
|
EXPECTED = """Expected %r to raise %s.
|
|
args = %r
|
|
options = %r
|
|
output = %r"""
|
|
|
|
|
|
UNEXPECTED = """Expected %r to raise %s, but caught different.
|
|
args = %r
|
|
options = %r
|
|
%s: %s"""
|
|
|
|
|
|
KWARGS = """Command %r raised %s with wrong kwargs.
|
|
args = %r
|
|
options = %r
|
|
kw_expected = %r
|
|
kw_got = %r"""
|
|
|
|
|
|
class Declarative(XMLRPC_test):
|
|
"""A declarative-style test suite
|
|
|
|
A Declarative test suite is controlled by the ``tests`` and
|
|
``cleanup_commands`` class variables.
|
|
|
|
The ``tests`` is a list of dictionaries with the following keys:
|
|
|
|
``desc``
|
|
A name/description of the test
|
|
``command``
|
|
A (command, args, kwargs) triple specifying the command to run
|
|
``expected``
|
|
Can be either an ``errors.PublicError`` instance, in which case
|
|
the command must fail with the given error; or the
|
|
expected result.
|
|
The result is checked with ``tests.util.assert_deepequal``.
|
|
``extra_check`` (optional)
|
|
A checking function that is called with the response. It must
|
|
return true for the test to pass.
|
|
|
|
The ``cleanup_commands`` is a list of (command, args, kwargs)
|
|
triples. These are commands get run both before and after tests,
|
|
and must not fail.
|
|
"""
|
|
|
|
cleanup_commands = tuple()
|
|
tests = tuple()
|
|
|
|
def cleanup_generate(self, stage):
|
|
for (i, command) in enumerate(self.cleanup_commands):
|
|
func = lambda: self.cleanup(command)
|
|
func.description = '%s %s-cleanup[%d]: %r' % (
|
|
self.__class__.__name__, stage, i, command
|
|
)
|
|
yield (func,)
|
|
|
|
def cleanup(self, command):
|
|
(cmd, args, options) = command
|
|
if cmd not in api.Command:
|
|
raise nose.SkipTest(
|
|
'cleanup command %r not in api.Command' % cmd
|
|
)
|
|
try:
|
|
api.Command[cmd](*args, **options)
|
|
except (errors.NotFound, errors.EmptyModlist):
|
|
pass
|
|
|
|
def test_generator(self):
|
|
"""
|
|
Iterate through tests.
|
|
|
|
nose reports each one as a seperate test.
|
|
"""
|
|
|
|
# Iterate through pre-cleanup:
|
|
for tup in self.cleanup_generate('pre'):
|
|
yield tup
|
|
|
|
# Iterate through the tests:
|
|
name = self.__class__.__name__
|
|
for (i, test) in enumerate(self.tests):
|
|
nice = '%s[%d]: %s: %s' % (
|
|
name, i, test['command'][0], test.get('desc', '')
|
|
)
|
|
func = lambda: self.check(nice, **test)
|
|
func.description = nice
|
|
yield (func,)
|
|
|
|
# Iterate through post-cleanup:
|
|
for tup in self.cleanup_generate('post'):
|
|
yield tup
|
|
|
|
def check(self, nice, desc, command, expected, extra_check=None):
|
|
(cmd, args, options) = command
|
|
options.setdefault('version', API_VERSION)
|
|
if cmd not in api.Command:
|
|
raise nose.SkipTest('%r not in api.Command' % cmd)
|
|
if isinstance(expected, errors.PublicError):
|
|
self.check_exception(nice, cmd, args, options, expected)
|
|
elif hasattr(expected, '__call__'):
|
|
self.check_callable(nice, cmd, args, options, expected)
|
|
else:
|
|
self.check_output(nice, cmd, args, options, expected, extra_check)
|
|
|
|
def check_exception(self, nice, cmd, args, options, expected):
|
|
klass = expected.__class__
|
|
name = klass.__name__
|
|
try:
|
|
output = api.Command[cmd](*args, **options)
|
|
except StandardError, e:
|
|
pass
|
|
else:
|
|
raise AssertionError(
|
|
EXPECTED % (cmd, name, args, options, output)
|
|
)
|
|
if not isinstance(e, klass):
|
|
raise AssertionError(
|
|
UNEXPECTED % (cmd, name, args, options, e.__class__.__name__, e)
|
|
)
|
|
# FIXME: the XML-RPC transport doesn't allow us to return structured
|
|
# information through the exception, so we can't test the kw on the
|
|
# client side. However, if we switch to using JSON-RPC for the default
|
|
# transport, the exception is a free-form data structure (dict).
|
|
# For now just compare the strings
|
|
assert_deepequal(expected.strerror, e.strerror)
|
|
|
|
def check_callable(self, nice, cmd, args, options, expected):
|
|
output = dict()
|
|
e = None
|
|
try:
|
|
output = api.Command[cmd](*args, **options)
|
|
except StandardError, e:
|
|
pass
|
|
if not expected(e, output):
|
|
raise AssertionError(
|
|
UNEXPECTED % (cmd, args, options, e.__class__.__name__, e)
|
|
)
|
|
|
|
def check_output(self, nice, cmd, args, options, expected, extra_check):
|
|
got = api.Command[cmd](*args, **options)
|
|
assert_deepequal(expected, got, nice)
|
|
if extra_check and not extra_check(got):
|
|
raise AssertionError('Extra check %s failed' % extra_check)
|