mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2025-02-25 18:55:28 -06:00
3bf91eab25
Python 2 has keys()/values()/items(), which return lists, iterkeys()/itervalues()/iteritems(), which return iterators, and viewkeys()/viewvalues()/viewitems() which return views. Python 3 has only keys()/values()/items(), which return views. To get iterators, one can use iter() or a for loop/comprehension; for lists there's the list() constructor. When iterating through the entire dict, without modifying the dict, the difference between Python 2's items() and iteritems() is negligible, especially on small dicts (the main overhead is extra memory, not CPU time). In the interest of simpler code, this patch changes many instances of iteritems() to items(), iterkeys() to keys() etc. In other cases, helpers like six.itervalues are used. Reviewed-By: Christian Heimes <cheimes@redhat.com> Reviewed-By: Jan Cholasta <jcholast@redhat.com>
282 lines
9.5 KiB
Python
282 lines
9.5 KiB
Python
# Authors:
|
|
# Jason Gerard DeRose <jderose@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/>.
|
|
|
|
"""
|
|
Test the `ipalib.backend` module.
|
|
"""
|
|
|
|
# FIXME: Pylint errors
|
|
# pylint: disable=no-member
|
|
# pylint: disable=maybe-no-member
|
|
|
|
import threading
|
|
from ipatests.util import ClassChecker, raises, create_test_api
|
|
from ipatests.data import unicode_str
|
|
from ipalib.request import context, Connection
|
|
from ipalib.frontend import Command
|
|
from ipalib import backend, plugable, errors, base
|
|
from ipapython.version import API_VERSION
|
|
|
|
|
|
|
|
class test_Backend(ClassChecker):
|
|
"""
|
|
Test the `ipalib.backend.Backend` class.
|
|
"""
|
|
|
|
_cls = backend.Backend
|
|
|
|
def test_class(self):
|
|
assert self.cls.__bases__ == (plugable.Plugin,)
|
|
|
|
|
|
class Disconnect(object):
|
|
called = False
|
|
|
|
def __init__(self, id=None):
|
|
self.id = id
|
|
|
|
def __call__(self):
|
|
assert self.called is False
|
|
self.called = True
|
|
if self.id is not None:
|
|
delattr(context, self.id)
|
|
|
|
|
|
class test_Connectible(ClassChecker):
|
|
"""
|
|
Test the `ipalib.backend.Connectible` class.
|
|
"""
|
|
|
|
_cls = backend.Connectible
|
|
|
|
def test_connect(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.connect` method.
|
|
"""
|
|
# Test that connection is created:
|
|
api = 'the api instance'
|
|
class example(self.cls):
|
|
def create_connection(self, *args, **kw):
|
|
object.__setattr__(self, 'args', args)
|
|
object.__setattr__(self, 'kw', kw)
|
|
return 'The connection.'
|
|
o = example(api, shared_instance=True)
|
|
args = ('Arg1', 'Arg2', 'Arg3')
|
|
kw = dict(key1='Val1', key2='Val2', key3='Val3')
|
|
assert not hasattr(context, 'example')
|
|
assert o.connect(*args, **kw) is None
|
|
conn = context.example
|
|
assert type(conn) is Connection
|
|
assert o.args == args
|
|
assert o.kw == kw
|
|
assert conn.conn == 'The connection.'
|
|
assert conn.disconnect == o.disconnect
|
|
|
|
# Test that StandardError is raised if already connected:
|
|
m = "connect: 'context.%s' already exists in thread %r"
|
|
e = raises(StandardError, o.connect, *args, **kw)
|
|
assert str(e) == m % ('example', threading.currentThread().getName())
|
|
|
|
# Double check that it works after deleting context.example:
|
|
del context.example
|
|
assert o.connect(*args, **kw) is None
|
|
|
|
def test_create_connection(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.create_connection` method.
|
|
"""
|
|
api = 'the api instance'
|
|
class example(self.cls):
|
|
pass
|
|
for klass in (self.cls, example):
|
|
o = klass(api, shared_instance=True)
|
|
e = raises(NotImplementedError, o.create_connection)
|
|
assert str(e) == '%s.create_connection()' % klass.__name__
|
|
|
|
def test_disconnect(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.disconnect` method.
|
|
"""
|
|
api = 'the api instance'
|
|
class example(self.cls):
|
|
destroy_connection = Disconnect()
|
|
o = example(api, shared_instance=True)
|
|
|
|
m = "disconnect: 'context.%s' does not exist in thread %r"
|
|
e = raises(StandardError, o.disconnect)
|
|
assert str(e) == m % ('example', threading.currentThread().getName())
|
|
|
|
context.example = 'The connection.'
|
|
assert o.disconnect() is None
|
|
assert example.destroy_connection.called is True
|
|
|
|
def test_destroy_connection(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.destroy_connection` method.
|
|
"""
|
|
api = 'the api instance'
|
|
class example(self.cls):
|
|
pass
|
|
for klass in (self.cls, example):
|
|
o = klass(api, shared_instance=True)
|
|
e = raises(NotImplementedError, o.destroy_connection)
|
|
assert str(e) == '%s.destroy_connection()' % klass.__name__
|
|
|
|
def test_isconnected(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.isconnected` method.
|
|
"""
|
|
api = 'the api instance'
|
|
class example(self.cls):
|
|
pass
|
|
for klass in (self.cls, example):
|
|
o = klass(api, shared_instance=True)
|
|
assert o.isconnected() is False
|
|
conn = 'whatever'
|
|
setattr(context, klass.__name__, conn)
|
|
assert o.isconnected() is True
|
|
delattr(context, klass.__name__)
|
|
|
|
def test_conn(self):
|
|
"""
|
|
Test the `ipalib.backend.Connectible.conn` property.
|
|
"""
|
|
api = 'the api instance'
|
|
msg = 'no context.%s in thread %r'
|
|
class example(self.cls):
|
|
pass
|
|
for klass in (self.cls, example):
|
|
o = klass(api, shared_instance=True)
|
|
e = raises(AttributeError, getattr, o, 'conn')
|
|
assert str(e) == msg % (
|
|
klass.__name__, threading.currentThread().getName()
|
|
)
|
|
conn = Connection('The connection.', Disconnect())
|
|
setattr(context, klass.__name__, conn)
|
|
assert o.conn is conn.conn
|
|
delattr(context, klass.__name__)
|
|
|
|
|
|
class test_Executioner(ClassChecker):
|
|
"""
|
|
Test the `ipalib.backend.Executioner` class.
|
|
"""
|
|
_cls = backend.Executioner
|
|
|
|
def test_execute(self):
|
|
"""
|
|
Test the `ipalib.backend.Executioner.execute` method.
|
|
"""
|
|
(api, home) = create_test_api(in_server=True)
|
|
|
|
class echo(Command):
|
|
takes_args = ('arg1', 'arg2+')
|
|
takes_options = ('option1?', 'option2?')
|
|
def execute(self, *args, **options):
|
|
assert type(args[1]) is tuple
|
|
return dict(result=args + (options,))
|
|
api.add_plugin(echo)
|
|
|
|
class good(Command):
|
|
def execute(self, **options):
|
|
raise errors.ValidationError(
|
|
name='nurse',
|
|
error=u'Not naughty!',
|
|
)
|
|
api.add_plugin(good)
|
|
|
|
class bad(Command):
|
|
def execute(self, **options):
|
|
raise ValueError('This is private.')
|
|
api.add_plugin(bad)
|
|
|
|
class with_name(Command):
|
|
"""
|
|
Test that a kwarg named 'name' can be used.
|
|
"""
|
|
takes_options = 'name'
|
|
def execute(self, **options):
|
|
return dict(result=options['name'].upper())
|
|
api.add_plugin(with_name)
|
|
|
|
api.finalize()
|
|
o = self.cls(api)
|
|
o.finalize()
|
|
|
|
# Test that CommandError is raised:
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
print str(list(context.__dict__))
|
|
e = raises(errors.CommandError, o.execute, 'nope')
|
|
assert e.name == 'nope'
|
|
assert conn.disconnect.called is True # Make sure destroy_context() was called
|
|
print str(list(context.__dict__))
|
|
assert list(context.__dict__) == []
|
|
|
|
# Test with echo command:
|
|
arg1 = unicode_str
|
|
arg2 = (u'Hello', unicode_str, u'world!')
|
|
args = (arg1,) + arg2
|
|
options = dict(option1=u'How are you?', option2=unicode_str,
|
|
version=API_VERSION)
|
|
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
print o.execute('echo', arg1, arg2, **options)
|
|
print dict(
|
|
result=(arg1, arg2, options)
|
|
)
|
|
assert o.execute('echo', arg1, arg2, **options) == dict(
|
|
result=(arg1, arg2, options)
|
|
)
|
|
assert conn.disconnect.called is True # Make sure destroy_context() was called
|
|
assert list(context.__dict__) == []
|
|
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
assert o.execute('echo', *args, **options) == dict(
|
|
result=(arg1, arg2, options)
|
|
)
|
|
assert conn.disconnect.called is True # Make sure destroy_context() was called
|
|
assert list(context.__dict__) == []
|
|
|
|
# Test with good command:
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
e = raises(errors.ValidationError, o.execute, 'good')
|
|
assert e.name == 'nurse'
|
|
assert e.error == u'Not naughty!'
|
|
assert conn.disconnect.called is True # Make sure destroy_context() was called
|
|
assert list(context.__dict__) == []
|
|
|
|
# Test with bad command:
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
e = raises(errors.InternalError, o.execute, 'bad')
|
|
assert conn.disconnect.called is True # Make sure destroy_context() was called
|
|
assert list(context.__dict__) == []
|
|
|
|
# Test with option 'name':
|
|
conn = Connection('The connection.', Disconnect('someconn'))
|
|
context.someconn = conn
|
|
expected = dict(result=u'TEST')
|
|
assert expected == o.execute('with_name', name=u'test',
|
|
version=API_VERSION)
|