mirror of
https://salsa.debian.org/freeipa-team/freeipa.git
synced 2024-12-23 07:33:27 -06:00
18: Moved base2 stuff into base
This commit is contained in:
Jason Gerard DeRose
parent
66cd39f519
commit
9b3e2f5cec
181
ipalib/base.py
181
ipalib/base.py
@ -127,80 +127,96 @@ class NameSpace(object):
|
||||
return len(self.__keys)
|
||||
|
||||
|
||||
|
||||
class Named(object):
|
||||
def __get_name(self):
|
||||
return self.__class__.__name__
|
||||
name = property(__get_name)
|
||||
|
||||
def __get_cli(self):
|
||||
return self.name.replace('_', '-')
|
||||
cli = property(__get_cli)
|
||||
|
||||
def __get_doc(self):
|
||||
return self.__class__.__doc__.strip()
|
||||
doc = property(__get_doc)
|
||||
|
||||
|
||||
class ObjectMember(Named):
|
||||
def __init__(self, obj):
|
||||
self.__obj = obj
|
||||
class WithObj(Named):
|
||||
_obj = None
|
||||
__obj = None
|
||||
__obj_locked = False
|
||||
|
||||
def __get_obj(self):
|
||||
return self.__obj
|
||||
obj = property(__get_obj)
|
||||
def __set_obj(self, obj):
|
||||
if self.__obj_locked:
|
||||
raise exceptions.TwiceSetError(self.__class__.__name__, 'obj')
|
||||
self.__obj_locked = True
|
||||
if obj is None:
|
||||
assert self.__obj is None
|
||||
assert self.obj is None
|
||||
else:
|
||||
assert isinstance(obj, Named)
|
||||
assert isinstance(self._obj, str)
|
||||
assert obj.name == self._obj
|
||||
self.__obj = obj
|
||||
assert self.obj is obj
|
||||
obj = property(__get_obj, __set_obj)
|
||||
|
||||
|
||||
class Command(ObjectMember):
|
||||
def __get_full_name(self):
|
||||
return '%s_%s' % (self.name, self.obj.name)
|
||||
full_name = property(__get_full_name)
|
||||
|
||||
|
||||
class Attribute(ObjectMember):
|
||||
def __get_full_name(self):
|
||||
return '%s_%s' % (self.obj.name, self.name)
|
||||
full_name = property(__get_full_name)
|
||||
class Command(WithObj):
|
||||
pass
|
||||
|
||||
class Property(WithObj):
|
||||
pass
|
||||
|
||||
class Object(Named):
|
||||
def __init__(self):
|
||||
self.__commands = self.__build_ns(self.get_commands)
|
||||
self.__attributes = self.__build_ns(self.get_attributes, True)
|
||||
__commands = None
|
||||
|
||||
def __get_commands(self):
|
||||
return self.__commands
|
||||
commands = property(__get_commands)
|
||||
|
||||
def __get_attributes(self):
|
||||
return self.__attributes
|
||||
attributes = property(__get_attributes)
|
||||
|
||||
def __build_ns(self, callback, preserve=False):
|
||||
d = {}
|
||||
o = []
|
||||
for cls in callback():
|
||||
i = cls(self)
|
||||
assert i.name not in d
|
||||
d[i.name] = i
|
||||
o.append(i.name)
|
||||
if preserve:
|
||||
return NameSpace(d, order=o)
|
||||
return NameSpace(d)
|
||||
|
||||
def get_commands(self):
|
||||
return []
|
||||
|
||||
def get_attributes(self):
|
||||
return []
|
||||
def __set_commands(self, commands):
|
||||
if self.__commands is not None:
|
||||
raise exceptions.TwiceSetError(
|
||||
self.__class__.__name__, 'commands'
|
||||
)
|
||||
assert type(commands) is NameSpace
|
||||
self.__commands = commands
|
||||
assert self.commands is commands
|
||||
commands = property(__get_commands, __set_commands)
|
||||
|
||||
|
||||
class API(object):
|
||||
__objects = None
|
||||
class Collector(object):
|
||||
def __init__(self):
|
||||
self.__d = {}
|
||||
self.globals = []
|
||||
|
||||
def __getitem__(self, key):
|
||||
assert isinstance(key, str)
|
||||
if key not in self.__d:
|
||||
self.__d[key] = []
|
||||
return self.__d[key]
|
||||
|
||||
def __iter__(self):
|
||||
for key in self.__d:
|
||||
yield key
|
||||
|
||||
def add(self, i):
|
||||
assert isinstance(i, WithObj)
|
||||
if i._obj is None:
|
||||
self.globals.append(i)
|
||||
else:
|
||||
self[i._obj].append(i)
|
||||
|
||||
def namespaces(self):
|
||||
for key in self:
|
||||
d = dict((i.name, i) for i in self[key])
|
||||
yield (key, NameSpace(d))
|
||||
|
||||
|
||||
|
||||
class Registrar(object):
|
||||
__object = None
|
||||
__commands = None
|
||||
__max_cmd_len = None
|
||||
__properties = None
|
||||
|
||||
def __init__(self):
|
||||
self.__obj_d = {}
|
||||
self.__tmp_objects = {}
|
||||
self.__tmp_commands = {}
|
||||
self.__tmp_properties = {}
|
||||
|
||||
def __get_objects(self):
|
||||
return self.__objects
|
||||
@ -210,30 +226,43 @@ class API(object):
|
||||
return self.__commands
|
||||
commands = property(__get_commands)
|
||||
|
||||
def __get_max_cmd_len(self):
|
||||
if self.__max_cmd_len is None:
|
||||
if self.__commands is None:
|
||||
return 0
|
||||
self.__max_cmd_len = max(len(n) for n in self.__commands)
|
||||
return self.__max_cmd_len
|
||||
max_cmd_len = property(__get_max_cmd_len)
|
||||
def __get_target(self, i):
|
||||
if isinstance(i, Object):
|
||||
return (self.__tmp_objects, i.name)
|
||||
if isinstance(i, Command):
|
||||
return (self.__tmp_commands, i.name)
|
||||
assert isinstance(i, Property)
|
||||
|
||||
def register_object(self, cls, override=False):
|
||||
assert type(override) is bool
|
||||
if not (inspect.isclass(cls) and issubclass(cls, Object)):
|
||||
raise exceptions.RegistrationError(cls, 'Object')
|
||||
obj = cls()
|
||||
if obj.name in self.__obj_d and not override:
|
||||
raise exceptions.OverrideError(obj.name)
|
||||
self.__obj_d[obj.name] = obj
|
||||
|
||||
def register(self, cls):
|
||||
assert inspect.isclass(cls)
|
||||
assert issubclass(cls, Named)
|
||||
i = cls()
|
||||
(target, key) = self.__get_target(i)
|
||||
target[key] = i
|
||||
|
||||
def finalize(self):
|
||||
cmd_d = {}
|
||||
cmd_l = {}
|
||||
for obj in self.__obj_d.values():
|
||||
for cmd in obj.commands():
|
||||
assert cmd.full_name not in cmd_d
|
||||
cmd_d[cmd.full_name] = cmd
|
||||
self.__commands = NameSpace(cmd_d)
|
||||
self.__objects = NameSpace(self.__obj_d)
|
||||
self.__obj_d = None
|
||||
obj_cmd = Collector()
|
||||
for cmd in self.__tmp_commands.values():
|
||||
if cmd._obj is None:
|
||||
cmd.obj = None
|
||||
else:
|
||||
obj = self.__tmp_objects[cmd._obj]
|
||||
cmd.obj = obj
|
||||
obj_cmd.add(cmd)
|
||||
self.__objects = NameSpace(self.__tmp_objects)
|
||||
self.__commands = NameSpace(self.__tmp_commands)
|
||||
for (key, ns) in obj_cmd.namespaces():
|
||||
self.objects[key].commands = ns
|
||||
|
||||
|
||||
class API(Registrar):
|
||||
__max_cmd_len = None
|
||||
|
||||
def __get_max_cmd_len(self):
|
||||
if self.__max_cmd_len is None:
|
||||
if self.commands is None:
|
||||
return 0
|
||||
self.__max_cmd_len = max(len(n) for n in self.commands)
|
||||
return self.__max_cmd_len
|
||||
max_cmd_len = property(__get_max_cmd_len)
|
||||
|
||||
@ -23,26 +23,15 @@ Base classes for objects with CRUD functionality.
|
||||
|
||||
import base
|
||||
|
||||
class add(base.Command):
|
||||
pass
|
||||
|
||||
class find(base.Command):
|
||||
pass
|
||||
class Add(base.Command):
|
||||
pass
|
||||
|
||||
class edit(base.Command):
|
||||
pass
|
||||
class Del(base.Command):
|
||||
pass
|
||||
|
||||
class delete(base.Command):
|
||||
pass
|
||||
class Mod(base.Command):
|
||||
pass
|
||||
|
||||
|
||||
|
||||
|
||||
class CrudLike(base.Object):
|
||||
def get_commands(self):
|
||||
return [
|
||||
add,
|
||||
find,
|
||||
edit,
|
||||
delete,
|
||||
]
|
||||
class Find(base.Command):
|
||||
pass
|
||||
|
||||
@ -22,18 +22,25 @@ Some example plugins.
|
||||
"""
|
||||
|
||||
import crud
|
||||
import base
|
||||
from run import api
|
||||
|
||||
class user(crud.CrudLike):
|
||||
class user(base.Object):
|
||||
pass
|
||||
api.register_object(user)
|
||||
api.register(user)
|
||||
|
||||
class adduser(crud.Add):
|
||||
_obj = 'user'
|
||||
api.register(adduser)
|
||||
|
||||
class group(crud.CrudLike):
|
||||
pass
|
||||
api.register_object(group)
|
||||
class deluser(crud.Del):
|
||||
_obj = 'user'
|
||||
api.register(deluser)
|
||||
|
||||
class moduser(crud.Mod):
|
||||
_obj = 'user'
|
||||
api.register(moduser)
|
||||
|
||||
class service(crud.CrudLike):
|
||||
pass
|
||||
api.register_object(service)
|
||||
class finduser(crud.Find):
|
||||
_obj = 'user'
|
||||
api.register(finduser)
|
||||
|
||||
@ -185,189 +185,96 @@ class test_NameSpace:
|
||||
|
||||
def test_Named():
|
||||
class named_class(base.Named):
|
||||
"""
|
||||
This class is so introspective!
|
||||
"""
|
||||
pass
|
||||
|
||||
i = named_class()
|
||||
assert i.name == 'named_class'
|
||||
assert i.cli == 'named-class'
|
||||
assert i.doc == 'This class is so introspective!'
|
||||
|
||||
|
||||
def test_Command():
|
||||
class user(object):
|
||||
name = 'user'
|
||||
class add(base.Command):
|
||||
pass
|
||||
i = add(user())
|
||||
assert i.name == 'add'
|
||||
assert i.full_name == 'add_user'
|
||||
|
||||
|
||||
def test_Attribute():
|
||||
class user(object):
|
||||
name = 'user'
|
||||
class sn(base.Attribute):
|
||||
pass
|
||||
i = sn(user())
|
||||
assert i.name == 'sn'
|
||||
assert i.full_name == 'user_sn'
|
||||
|
||||
|
||||
def test_Object():
|
||||
class create(base.Command):
|
||||
pass
|
||||
|
||||
class retrieve(base.Command):
|
||||
pass
|
||||
|
||||
class update(base.Command):
|
||||
pass
|
||||
|
||||
class delete(base.Command):
|
||||
pass
|
||||
|
||||
class givenName(base.Attribute):
|
||||
def test_WithObj():
|
||||
class some_object(base.Named):
|
||||
pass
|
||||
|
||||
class sn(base.Attribute):
|
||||
class another_object(base.Named):
|
||||
pass
|
||||
|
||||
class login(base.Attribute):
|
||||
pass
|
||||
class some_command(base.WithObj):
|
||||
_obj = 'some_object'
|
||||
|
||||
obj = some_object()
|
||||
cmd = some_command()
|
||||
|
||||
# Test that it can be set:
|
||||
assert cmd.obj is None
|
||||
cmd.obj = obj
|
||||
assert cmd.obj is obj
|
||||
|
||||
# Test that it cannot be set twice:
|
||||
raised = False
|
||||
try:
|
||||
cmd.obj = obj
|
||||
except exceptions.TwiceSetError:
|
||||
raised = True
|
||||
assert raised
|
||||
|
||||
# Test that it can't be set with the wrong name:
|
||||
obj = another_object()
|
||||
cmd = some_command()
|
||||
raised = False
|
||||
try:
|
||||
cmd.obj = obj
|
||||
except AssertionError:
|
||||
raised = True
|
||||
assert raised
|
||||
|
||||
|
||||
def test_Registar():
|
||||
class adduser(base.Command):
|
||||
_obj = 'user'
|
||||
class moduser(base.Command):
|
||||
_obj = 'user'
|
||||
class deluser(base.Command):
|
||||
_obj = 'user'
|
||||
class finduser(base.Command):
|
||||
_obj = 'user'
|
||||
class kinit(base.Command):
|
||||
pass
|
||||
class user(base.Object):
|
||||
def get_commands(self):
|
||||
return [
|
||||
create,
|
||||
retrieve,
|
||||
update,
|
||||
delete,
|
||||
]
|
||||
pass
|
||||
class group(base.Object):
|
||||
pass
|
||||
|
||||
def get_attributes(self):
|
||||
return [
|
||||
givenName,
|
||||
sn,
|
||||
login,
|
||||
]
|
||||
r = base.Registrar()
|
||||
r.register(adduser)
|
||||
r.register(moduser)
|
||||
r.register(deluser)
|
||||
r.register(finduser)
|
||||
r.register(kinit)
|
||||
r.register(user)
|
||||
r.register(group)
|
||||
|
||||
i = user()
|
||||
assert i.name == 'user'
|
||||
r.finalize()
|
||||
assert len(r.commands) == 5
|
||||
assert len(r.objects) == 2
|
||||
|
||||
# Test commands:
|
||||
commands = i.commands
|
||||
assert isinstance(commands, base.NameSpace)
|
||||
assert list(commands) == ['create', 'delete', 'retrieve', 'update']
|
||||
assert len(commands) == 4
|
||||
for name in commands:
|
||||
cls = locals()[name]
|
||||
cmd = commands[name]
|
||||
assert type(cmd) is cls
|
||||
assert getattr(commands, name) is cmd
|
||||
assert cmd.name == name
|
||||
assert cmd.full_name == ('%s_user' % name)
|
||||
obj = r.objects.user
|
||||
assert type(obj) is user
|
||||
for name in ['adduser', 'moduser', 'deluser', 'finduser']:
|
||||
cmd = r.commands[name]
|
||||
assert type(cmd) is locals()[name]
|
||||
assert cmd.obj is obj
|
||||
|
||||
# Test attributes:
|
||||
attributes = i.attributes
|
||||
assert isinstance(attributes, base.NameSpace)
|
||||
assert list(attributes) == ['givenName', 'sn', 'login']
|
||||
assert len(attributes) == 3
|
||||
for name in attributes:
|
||||
cls = locals()[name]
|
||||
attr = attributes[name]
|
||||
assert type(attr) is cls
|
||||
assert getattr(attributes, name) is attr
|
||||
assert attr.name == name
|
||||
assert attr.full_name == ('user_%s' % name)
|
||||
assert r.commands.kinit.obj is None
|
||||
|
||||
|
||||
class test_API:
|
||||
"""
|
||||
Unit tests for `API` class.
|
||||
"""
|
||||
|
||||
def new(self):
|
||||
"""
|
||||
Returns a new API instance.
|
||||
"""
|
||||
return base.API()
|
||||
|
||||
def test_fresh(self):
|
||||
"""
|
||||
Test expectations of a fresh API instance.
|
||||
"""
|
||||
api = self.new()
|
||||
assert read_only(api, 'objects') is None
|
||||
assert read_only(api, 'commands') is None
|
||||
|
||||
def test_register_exception(self):
|
||||
"""
|
||||
Check that RegistrationError is raised when registering anything
|
||||
other than a subclass of Command.
|
||||
"""
|
||||
api = self.new()
|
||||
|
||||
class my_command(base.Command):
|
||||
pass
|
||||
|
||||
for obj in [object, my_command]:
|
||||
raised = False
|
||||
try:
|
||||
api.register_object(obj)
|
||||
except exceptions.RegistrationError:
|
||||
raised = True
|
||||
assert raised
|
||||
|
||||
def test_override_exception(self):
|
||||
class some_object(base.Object):
|
||||
def get_commands(self):
|
||||
return []
|
||||
def get_attributes(self):
|
||||
return []
|
||||
|
||||
api = self.new()
|
||||
api.register_object(some_object)
|
||||
for cmd in r.commands():
|
||||
raised = False
|
||||
try:
|
||||
api.register_object(some_object)
|
||||
except exceptions.OverrideError:
|
||||
cmd.obj = None
|
||||
except exceptions.TwiceSetError:
|
||||
raised = True
|
||||
assert raised
|
||||
api.register_object(some_object, override=True)
|
||||
|
||||
def test_finalize(self):
|
||||
class user(crud.CrudLike):
|
||||
pass
|
||||
class group(crud.CrudLike):
|
||||
pass
|
||||
class service(crud.CrudLike):
|
||||
pass
|
||||
|
||||
names = list(user().commands)
|
||||
assert len(names) == 4
|
||||
full_names = set()
|
||||
for o in ['user', 'group', 'service']:
|
||||
full_names.update('%s_%s' % (v, o) for v in names)
|
||||
assert len(full_names) == 12
|
||||
|
||||
|
||||
api = self.new()
|
||||
api.register_object(user)
|
||||
api.register_object(group)
|
||||
api.register_object(service)
|
||||
api.finalize()
|
||||
|
||||
# Test API.objects property:
|
||||
objects = read_only(api, 'objects')
|
||||
assert type(objects) is base.NameSpace
|
||||
assert objects is api.objects # Same instance must be returned
|
||||
assert len(objects) is 3
|
||||
assert list(objects) == ['group', 'service', 'user']
|
||||
|
||||
# Test API.commands property:
|
||||
commands = read_only(api, 'commands')
|
||||
assert type(commands) is base.NameSpace
|
||||
assert commands is api.commands # Same instance must be returned
|
||||
assert len(commands) is 12
|
||||
assert list(commands) == sorted(full_names)
|
||||
u = r.objects.user
|
||||
assert isinstance(u.commands, base.NameSpace)
|
||||
assert len(u.commands) == 4
|
||||
assert list(u.commands) == ['adduser', 'deluser', 'finduser', 'moduser']
|
||||
|
||||
Loading…
Reference in New Issue
Block a user