websms/freeipa
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Imported Debian patch 4.0.5-6~numeezy

Browse Source
This commit is contained in:
Alexandre Ellert
2016-02-17 15:07:45 +01:00
committed by Mario Fetka
parent c44de33144
commit 10dfc9587b
1203 changed files with 53869 additions and 241462 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
ipatests/test_ipalib/test_backend.py
View File

@@ -20,11 +20,6 @@
"""
Test the `ipalib.backend` module.
"""
from __future__ import print_function
# FIXME: Pylint errors
# pylint: disable=no-member
# pylint: disable=maybe-no-member
import threading
from ipatests.util import ClassChecker, raises, create_test_api
@@ -34,9 +29,7 @@ from ipalib.frontend import Command
from ipalib import backend, plugable, errors, base
from ipapython.version import API_VERSION
import pytest
pytestmark = pytest.mark.tier0
class test_Backend(ClassChecker):
"""
@@ -74,13 +67,12 @@ class test_Connectible(ClassChecker):
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)
o = example()
args = ('Arg1', 'Arg2', 'Arg3')
kw = dict(key1='Val1', key2='Val2', key3='Val3')
assert not hasattr(context, 'example')
@@ -92,11 +84,10 @@ class test_Connectible(ClassChecker):
assert conn.conn == 'The connection.'
assert conn.disconnect == o.disconnect
# Test that Exception is raised if already connected:
m = "{0} is already connected ({1} in {2})"
e = raises(Exception, o.connect, *args, **kw)
assert str(e) == m.format(
'example', o.id, threading.currentThread().getName())
# 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
@@ -106,11 +97,10 @@ class test_Connectible(ClassChecker):
"""
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)
o = klass()
e = raises(NotImplementedError, o.create_connection)
assert str(e) == '%s.create_connection()' % klass.__name__
@@ -118,15 +108,13 @@ class test_Connectible(ClassChecker):
"""
Test the `ipalib.backend.Connectible.disconnect` method.
"""
api = 'the api instance'
class example(self.cls):
destroy_connection = Disconnect()
o = example(api, shared_instance=True)
o = example()
m = "{0} is not connected ({1} in {2})"
e = raises(Exception, o.disconnect)
assert str(e) == m.format(
'example', o.id, threading.currentThread().getName())
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
@@ -136,11 +124,10 @@ class test_Connectible(ClassChecker):
"""
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)
o = klass()
e = raises(NotImplementedError, o.destroy_connection)
assert str(e) == '%s.destroy_connection()' % klass.__name__
@@ -148,11 +135,10 @@ class test_Connectible(ClassChecker):
"""
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)
o = klass()
assert o.isconnected() is False
conn = 'whatever'
setattr(context, klass.__name__, conn)
@@ -163,15 +149,14 @@ class test_Connectible(ClassChecker):
"""
Test the `ipalib.backend.Connectible.conn` property.
"""
api = 'the api instance'
msg = '{0} is not connected ({1} in {2})'
msg = 'no context.%s in thread %r'
class example(self.cls):
pass
for klass in (self.cls, example):
o = klass(api, shared_instance=True)
o = klass()
e = raises(AttributeError, getattr, o, 'conn')
assert str(e) == msg.format(
klass.__name__, o.id, threading.currentThread().getName()
assert str(e) == msg % (
klass.__name__, threading.currentThread().getName()
)
conn = Connection('The connection.', Disconnect())
setattr(context, klass.__name__, conn)
@@ -197,7 +182,7 @@ class test_Executioner(ClassChecker):
def execute(self, *args, **options):
assert type(args[1]) is tuple
return dict(result=args + (options,))
api.add_plugin(echo)
api.register(echo)
class good(Command):
def execute(self, **options):
@@ -205,12 +190,12 @@ class test_Executioner(ClassChecker):
name='nurse',
error=u'Not naughty!',
)
api.add_plugin(good)
api.register(good)
class bad(Command):
def execute(self, **options):
raise ValueError('This is private.')
api.add_plugin(bad)
api.register(bad)
class with_name(Command):
"""
@@ -219,21 +204,22 @@ class test_Executioner(ClassChecker):
takes_options = 'name'
def execute(self, **options):
return dict(result=options['name'].upper())
api.add_plugin(with_name)
api.register(with_name)
api.finalize()
o = self.cls(api)
o = self.cls()
o.set_api(api)
o.finalize()
# Test that CommandError is raised:
conn = Connection('The connection.', Disconnect('someconn'))
context.someconn = conn
print(str(list(context.__dict__)))
print str(context.__dict__.keys())
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__) == []
print str(context.__dict__.keys())
assert context.__dict__.keys() == []
# Test with echo command:
arg1 = unicode_str
@@ -244,15 +230,15 @@ class test_Executioner(ClassChecker):
conn = Connection('The connection.', Disconnect('someconn'))
context.someconn = conn
print(o.execute('echo', arg1, arg2, **options))
print(dict(
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__) == []
assert context.__dict__.keys() == []
conn = Connection('The connection.', Disconnect('someconn'))
context.someconn = conn
@@ -260,7 +246,7 @@ class test_Executioner(ClassChecker):
result=(arg1, arg2, options)
)
assert conn.disconnect.called is True # Make sure destroy_context() was called
assert list(context.__dict__) == []
assert context.__dict__.keys() == []
# Test with good command:
conn = Connection('The connection.', Disconnect('someconn'))
@@ -269,14 +255,14 @@ class test_Executioner(ClassChecker):
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__) == []
assert context.__dict__.keys() == []
# 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__) == []
assert context.__dict__.keys() == []
# Test with option 'name':
conn = Connection('The connection.', Disconnect('someconn'))