From 438a29fa8f694500d3693b0050c80d04c2da3d53 Mon Sep 17 00:00:00 2001 From: Milan Kubík Date: Oct 27 2015 08:59:16 +0000 Subject: ipatests: Add initial CAACLTracker implementation The patch implements the tracker for CA ACL feature. The basic CRUD checkers has been implemented. The methods for adding and removing the association of the resources with the ACL do not have the check methods. These will be provided as a separate test suite. https://fedorahosted.org/freeipa/ticket/57 Reviewed-By: Martin Basti --- diff --git a/ipatests/test_xmlrpc/objectclasses.py b/ipatests/test_xmlrpc/objectclasses.py index 1cd77c7..134a088 100644 --- a/ipatests/test_xmlrpc/objectclasses.py +++ b/ipatests/test_xmlrpc/objectclasses.py @@ -217,3 +217,8 @@ certprofile = [ u'top', u'ipacertprofile', ] + +caacl = [ + u'ipaassociation', + u'ipacaacl' +] diff --git a/ipatests/test_xmlrpc/test_caacl_plugin.py b/ipatests/test_xmlrpc/test_caacl_plugin.py new file mode 100644 index 0000000..6cf835b --- /dev/null +++ b/ipatests/test_xmlrpc/test_caacl_plugin.py @@ -0,0 +1,378 @@ +# +# Copyright (C) 2015 FreeIPA Contributors see COPYING for license +# + +""" +Test the `ipalib.plugins.caacl` module. +""" + +import os + +import pytest + +from ipapython import ipautil +from ipalib import errors, x509 +from ipapython.dn import DN +from ipatests.test_xmlrpc.ldaptracker import Tracker +from ipatests.test_xmlrpc.xmlrpc_test import (XMLRPC_test, fuzzy_caacldn, + fuzzy_uuid, fuzzy_ipauniqueid, + raises_exact) +from ipatests.test_xmlrpc import objectclasses +from ipatests.util import assert_deepequal + + +class CAACLTracker(Tracker): + """Tracker class for CA ACL LDAP object. + + The class implements methods required by the base class + to help with basic CRUD operations. + + Methods for adding and deleting actual member entries into an ACL + do not have check methods as these would make the class + unnecessarily complicated. The checks are implemented as + a standalone test suite. However, this makes the test crucial + in debugging more complicated test cases. Since the add/remove + operation won't be checked right away by the tracker, a problem + in this operation may propagate into more complicated test case. + + It is possible to pass a list of member uids to these methods. + + The test uses instances of Fuzzy class to compare results as they + are in the UUID format. The dn and rdn properties were modified + to reflect this as well. + """ + + member_keys = { + u'memberuser_user', u'memberuser_group', + u'memberhost_host', u'memberhost_hostgroup', + u'memberservice_service', + u'ipamembercertprofile_certprofile'} + category_keys = { + u'ipacacategory', u'ipacertprofilecategory', u'usercategory', + u'hostcategory', u'servicecategory'} + retrieve_keys = { + u'dn', u'cn', u'description', u'ipaenabledflag', + u'ipamemberca', u'ipamembercertprofile', u'memberuser', + u'memberhost', u'memberservice'} | member_keys | category_keys + retrieve_all_keys = retrieve_keys | {u'objectclass', u'ipauniqueid'} + create_keys = {u'dn', u'cn', u'description', u'ipacertprofilecategory', + u'usercategory', u'hostcategory', u'ipacacategory', + u'servicecategory', u'ipaenabledflag', u'objectclass', + u'ipauniqueid'} + update_keys = create_keys - {u'dn'} + + def __init__(self, name, ipacertprofile_category=None, user_category=None, + service_category=None, host_category=None, description=None, + default_version=None): + super(CAACLTracker, self).__init__(default_version=default_version) + + self._name = name + self.description = description + self._categories = dict( + ipacertprofilecategory=ipacertprofile_category, + usercategory=user_category, + servicecategory=service_category, + hostcategory=host_category) + + self.dn = fuzzy_caacldn + + @property + def name(self): + return self._name + + @property + def rdn(self): + return fuzzy_ipauniqueid + + @property + def categories(self): + """To be used in track_create""" + return {cat: v for cat, v in self._categories.items() if v} + + @property + def create_categories(self): + """ Return the categories set on create. + Unused categories are left out. + """ + return {cat: [v] for cat, v in self.categories.items() if v} + + def make_create_command(self, force=True): + return self.make_command(u'caacl_add', self.name, + description=self.description, + **self.categories) + + def check_create(self, result): + assert_deepequal(dict( + value=self.name, + summary=u'Added CA ACL "{}"'.format(self.name), + result=dict(self.filter_attrs(self.create_keys)) + ), result) + + def track_create(self): + self.attrs = dict( + dn=self.dn, + ipauniqueid=[fuzzy_uuid], + cn=[self.name], + objectclass=objectclasses.caacl, + ipaenabledflag=[u'TRUE']) + + self.attrs.update(self.create_categories) + if self.description: + self.attrs.update({u'description', [self.description]}) + + self.exists = True + + def make_delete_command(self): + return self.make_command('caacl_del', self.name) + + def check_delete(self, result): + assert_deepequal(dict( + value=[self.name], + summary=u'Deleted CA ACL "{}"'.format(self.name), + result=dict(failed=[]) + ), result) + + def make_retrieve_command(self, all=False, raw=False): + return self.make_command('caacl_show', self.name, all=all, raw=raw) + + def check_retrieve(self, result, all=False, raw=False): + if all: + expected = self.filter_attrs(self.retrieve_all_keys) + else: + expected = self.filter_attrs(self.retrieve_keys) + + assert_deepequal(dict( + value=self.name, + summary=None, + result=expected + ), result) + + def make_find_command(self, *args, **kwargs): + return self.make_command('caacl_find', *args, **kwargs) + + def check_find(self, result, all=False, raw=False): + if all: + expected = self.filter_attrs(self.retrieve_all_keys) + else: + expected = self.filter_attrs(self.retrieve_keys) + + assert_deepequal(dict( + count=1, + truncated=False, + summary=u'1 CA ACL matched', + result=[expected] + ), result) + + def make_update_command(self, updates): + return self.make_command('caacl_mod', self.name, **updates) + + def update(self, updates, expected_updates=None, silent=False): + """If removing a category, delete it from tracker as well""" + # filter out empty categories and track changes + + filtered_updates = dict() + for key, value in updates.items(): + if key in self.category_keys: + if not value: + try: + del self.attrs[key] + except IndexError: + if silent: + pass + else: + # if there is a value, prepare the pair for update + filtered_updates.update({key: value}) + else: + filtered_updates.update({key: value}) + + if expected_updates is None: + expected_updates = {} + + command = self.make_update_command(updates) + + try: + result = command() + except errors.EmptyModlist: + if silent: + self.attrs.update(filtered_updates) + self.attrs.update(expected_updates) + self.check_update(result, + extra_keys=set(self.update_keys) | + set(expected_updates.keys())) + + def check_update(self, result, extra_keys=()): + assert_deepequal(dict( + value=self.name, + summary=u'Modified CA ACL "{}"'.format(self.name), + result=self.filter_attrs(self.update_keys | set(extra_keys)) + ), result) + + # Helper methods for caacl subcommands. The check methods will be + # implemented in standalone test + # + # The methods implemented here will be: + # caacl_{add,remove}_{host, service, certprofile, user [, subca]} + + def _add_acl_component(self, command_name, keys, track): + """ Add a resource into ACL rule and track it. + + command_name - the name in the API + keys = { + 'tracker_attr': { + 'api_key': 'value' + } + } + + e.g. + + keys = { + 'memberhost_host': { + 'host': 'hostname' + }, + 'memberhost_hostgroup': { + 'hostgroup': 'hostgroup_name' + } + } + """ + + if not self.exists: + raise errors.NotFound(reason="The tracked entry doesn't exist.") + + command = self.make_command(command_name, self.name) + command_options = dict() + + # track + for tracker_attr in keys: + api_options = keys[tracker_attr] + if track: + for option in api_options: + try: + if type(option) in (list, tuple): + self.attrs[tracker_attr].extend(api_options[option]) + else: + self.attrs[tracker_attr].append(api_options[option]) + except KeyError: + if type(option) in (list, tuple): + self.attrs[tracker_attr] = api_options[option] + else: + self.attrs[tracker_attr] = [api_options[option]] + # prepare options for the command call + command_options.update(api_options) + + return command(**command_options) + + def _remove_acl_component(self, command_name, keys, track): + """ Remove a resource from ACL rule and track it. + + command_name - the name in the API + keys = { + 'tracker_attr': { + 'api_key': 'value' + } + } + + e.g. + + keys = { + 'memberhost_host': { + 'host': 'hostname' + }, + 'memberhost_hostgroup': { + 'hostgroup': 'hostgroup_name' + } + } + """ + command = self.make_command(command_name, self.name) + command_options = dict() + + for tracker_attr in keys: + api_options = keys[tracker_attr] + if track: + for option in api_options: + if type(option) in (list, tuple): + for item in option: + self.attrs[tracker_attr].remove(item) + else: + self.attrs[tracker_attr].remove(api_options[option]) + if len(self.attrs[tracker_attr]) == 0: + del self.attrs[tracker_attr] + command_options.update(api_options) + + return command(**command_options) + + def add_host(self, host=None, hostgroup=None, track=True): + """Associates an host or hostgroup entry with the ACL. + + The command takes an unicode string with the name + of the entry (RDN). + + It is the responsibility of a test writer to provide + the correct value, object type as the method does not + verify whether the entry exists. + + The method can add only one entry of each type + in one call. + """ + + options = { + u'memberhost_host': {u'host': host}, + u'memberhost_hostgroup': {u'hostgroup': hostgroup}} + + return self._add_acl_component(u'caacl_add_host', options, track) + + def remove_host(self, host=None, hostgroup=None, track=True): + options = { + u'memberhost_host': {u'host': host}, + u'memberhost_hostgroup': {u'hostgroup': hostgroup}} + + return self._remove_acl_component(u'caacl_remove_host', options, track) + + def add_user(self, user=None, group=None, track=True): + options = { + u'memberuser_user': {u'user': user}, + u'memberuser_group': {u'group': group}} + + return self._add_acl_component(u'caacl_add_user', options, track) + + def remove_user(self, user=None, group=None, track=True): + options = { + u'memberuser_user': {u'user': user}, + u'memberuser_group': {u'group': group}} + + return self._remove_acl_component(u'caacl_remove_user', options, track) + + def add_service(self, service=None, track=True): + options = { + u'memberservice_service': {u'service': service}} + + return self._add_acl_component(u'caacl_add_service', options, track) + + def remove_service(self, service=None, track=True): + options = { + u'memberservice_service': {u'service': service}} + + return self._remove_acl_component(u'caacl_remove_service', options, track) + + def add_profile(self, certprofile=None, track=True): + options = { + u'ipamembercertprofile_certprofile': + {u'certprofile': certprofile}} + + return self._add_acl_component(u'caacl_add_profile', options, track) + + def remove_profile(self, certprofile=None, track=True): + options = { + u'ipamembercertprofile_certprofile': + {u'certprofile': certprofile}} + + return self._remove_acl_component(u'caacl_remove_profile', options, track) + + def enable(self): + command = self.make_command(u'caacl_enable', self.name) + self.attrs.update({u'ipaenabledflag': [u'TRUE']}) + command() + + def disable(self): + command = self.make_command(u'caacl_disable', self.name) + self.attrs.update({u'ipaenabledflag': [u'FALSE']}) + command()