ipatests: Tracker implementation for Sub CA feature

The patch implements Tracker subclass for CA plugin
and the basic CRUD tests for the plugin entries.

https://fedorahosted.org/freeipa/ticket/4559

Reviewed-By: Fraser Tweedale <ftweedal@redhat.com>

4 files changed, 309 insertions, 0 deletions

diff --git a/ipatests/test_xmlrpc/objectclasses.py b/ipatests/test_xmlrpc/objectclasses.py
index 134a08803..1ea020b18 100644
--- a/ipatests/test_xmlrpc/objectclasses.py
+++ b/ipatests/test_xmlrpc/objectclasses.py
@@ -222,3 +222,8 @@ caacl = [
     u'ipaassociation',
     u'ipacaacl'
 ]
+
+ca = [
+    u'top',
+    u'ipaca',
+]
diff --git a/ipatests/test_xmlrpc/test_ca_plugin.py b/ipatests/test_xmlrpc/test_ca_plugin.py
new file mode 100644
index 000000000..1e0e52ff7
--- /dev/null
+++ b/ipatests/test_xmlrpc/test_ca_plugin.py
@@ -0,0 +1,175 @@
+#
+# Copyright (C) 2016  FreeIPA Contributors see COPYING for license
+#
+
+"""
+Test the `ipalib.plugins.ca` module.
+"""
+
+import pytest
+
+from ipalib import errors
+from ipatests.test_xmlrpc.xmlrpc_test import XMLRPC_test, fuzzy_issuer
+
+from ipatests.test_xmlrpc.tracker.certprofile_plugin import CertprofileTracker
+from ipatests.test_xmlrpc.tracker.caacl_plugin import CAACLTracker
+from ipatests.test_xmlrpc.tracker.ca_plugin import CATracker
+
+
+@pytest.fixture(scope='module')
+def default_profile(request):
+    name = 'caIPAserviceCert'
+    desc = u'Standard profile for network services'
+    tracker = CertprofileTracker(name, store=True, desc=desc)
+    tracker.track_create()
+    return tracker
+
+
+@pytest.fixture(scope='module')
+def default_acl(request):
+    name = u'hosts_services_caIPAserviceCert'
+    tracker = CAACLTracker(name, service_category=u'all', host_category=u'all')
+    tracker.track_create()
+    tracker.attrs.update(
+        {u'ipamembercertprofile_certprofile': [u'caIPAserviceCert']})
+    return tracker
+
+
+@pytest.fixture(scope='module')
+def default_ca(request):
+    name = u'ipa'
+    desc = u'IPA CA'
+    tracker = CATracker(name, fuzzy_issuer, desc=desc)
+    tracker.track_create()
+    return tracker
+
+
+@pytest.fixture(scope='class')
+def crud_subca(request):
+    name = u'crud-subca'
+    subject = u'CN=crud subca test,O=crud testing inc'
+    tracker = CATracker(name, subject)
+
+    return tracker.make_fixture(request)
+
+
+@pytest.fixture(scope='class')
+def subject_conflict_subca(request):
+    name = u'crud-subca-2'
+    subject = u'CN=crud subca test,O=crud testing inc'
+    tracker = CATracker(name, subject)
+
+    # Should not get created, no need to delete
+    return tracker
+
+
+@pytest.mark.tier0
+class TestDefaultCA(XMLRPC_test):
+    def test_default_ca_present(self, default_ca):
+        default_ca.retrieve()
+
+    def test_default_ca_delete(self, default_ca):
+        with pytest.raises(errors.ProtectedEntryError):
+            default_ca.delete()
+
+
+@pytest.mark.tier1
+class TestCAbasicCRUD(XMLRPC_test):
+
+    ATTR_ERROR_MSG = u'attribute is not configurable'
+
+    def test_create(self, crud_subca):
+        crud_subca.create()
+
+    def test_retrieve(self, crud_subca):
+        crud_subca.retrieve()
+
+    def test_retrieve_all(self, crud_subca):
+        crud_subca.retrieve(all=True)
+
+    def test_delete(self, crud_subca):
+        crud_subca.delete()
+
+    def test_find(self, crud_subca):
+        crud_subca.ensure_exists()
+        crud_subca.find()
+
+    def test_modify_description(self, crud_subca):
+        new_desc = u'updated CA description'
+        crud_subca.update(
+            dict(
+                description=new_desc,
+            ),
+            expected_updates=dict(
+                description=[new_desc]
+            )
+        )
+
+    def test_modify_issuerdn(self, crud_subca):
+        bogus_issuer = u'ipacaissuerdn="cn=phony issuer,o=phony industries'
+        cmd = crud_subca.make_update_command(
+            updates=dict(setattr=bogus_issuer)
+        )
+
+        with pytest.raises(errors.ValidationError) as error:
+            cmd()
+
+        assert self.ATTR_ERROR_MSG in str(error.value)
+
+    def test_modify_subjectdn(self, crud_subca):
+        bogus_subject = u'ipacasubjectdn="cn=phony subject,o=phony industries'
+        cmd = crud_subca.make_update_command(
+            updates=dict(setattr=bogus_subject)
+        )
+
+        with pytest.raises(errors.ValidationError) as error:
+            cmd()
+
+        assert self.ATTR_ERROR_MSG in str(error.value)
+
+    def test_delete_subjectdn(self, crud_subca):
+        cmd = crud_subca.make_update_command(
+            updates=dict(delattr=u'ipacasubjectdn=%s'
+                         % crud_subca.ipasubjectdn)
+        )
+
+        with pytest.raises(errors.ValidationError) as error:
+            cmd()
+
+        assert self.ATTR_ERROR_MSG in str(error.value)
+
+    def test_add_bogus_subjectdn(self, crud_subca):
+        bogus_subject = u'ipacasubjectdn="cn=phony subject,o=phony industries'
+        cmd = crud_subca.make_update_command(
+            updates=dict(addattr=bogus_subject)
+        )
+
+        with pytest.raises(errors.ValidationError) as error:
+            cmd()
+
+        assert self.ATTR_ERROR_MSG in str(error.value)
+
+    def test_add_bogus_issuerdn(self, crud_subca):
+        bogus_issuer = u'ipacaissuerdn="cn=phony issuer,o=phony industries'
+        cmd = crud_subca.make_update_command(
+            updates=dict(addattr=bogus_issuer)
+        )
+
+        with pytest.raises(errors.ValidationError) as error:
+            cmd()
+
+        assert self.ATTR_ERROR_MSG in str(error.value)
+
+    def test_create_subca_with_conflicting_name(self, crud_subca):
+        crud_subca.ensure_exists()
+
+        cmd = crud_subca.make_create_command()
+        with pytest.raises(errors.DuplicateEntry):
+            cmd()
+
+    def test_create_subca_with_subject_conflict(
+            self, crud_subca, subject_conflict_subca):
+        crud_subca.ensure_exists()
+
+        with pytest.raises(errors.DuplicateEntry):
+            subject_conflict_subca.create()
diff --git a/ipatests/test_xmlrpc/tracker/ca_plugin.py b/ipatests/test_xmlrpc/tracker/ca_plugin.py
new file mode 100644
index 000000000..7586c771c
--- /dev/null
+++ b/ipatests/test_xmlrpc/tracker/ca_plugin.py
@@ -0,0 +1,126 @@
+#
+# Copyright (C) 2016  FreeIPA Contributors see COPYING for license
+#
+from __future__ import absolute_import
+
+import six
+
+from ipapython.dn import DN
+from ipatests.test_xmlrpc.tracker.base import Tracker
+from ipatests.util import assert_deepequal
+from ipatests.test_xmlrpc.xmlrpc_test import fuzzy_issuer, fuzzy_caid
+from ipatests.test_xmlrpc import objectclasses
+
+
+if six.PY3:
+    unicode = str
+
+
+class CATracker(Tracker):
+    """Implementation of a Tracker class for CA plugin."""
+
+    retrieve_keys = {
+        'dn', 'cn', 'ipacaid', 'ipacasubjectdn', 'ipacaissuerdn', 'description'
+    }
+    retrieve_all_keys = {'objectclass'} | retrieve_keys
+    create_keys = retrieve_all_keys
+    update_keys = retrieve_keys - {'dn'}
+
+    def __init__(self, name, subject, desc=u"Test generated CA",
+                 default_version=None):
+        super(CATracker, self).__init__(default_version=default_version)
+        self.attrs = {}
+        self.ipasubjectdn = subject
+        self.description = desc
+
+        self.dn = DN(('cn', name),
+                     self.api.env.container_ca,
+                     self.api.env.basedn)
+
+    def make_create_command(self, force=True):
+        """Make function that creates the plugin entry object."""
+        return self.make_command(
+            'ca_add', self.name, ipacasubjectdn=self.ipasubjectdn,
+            description=self.description
+        )
+
+    def check_create(self, result):
+        assert_deepequal(dict(
+            value=self.name,
+            summary=u'Created CA "{}"'.format(self.name),
+            result=dict(self.filter_attrs(self.create_keys))
+        ), result)
+
+    def track_create(self):
+        self.attrs = dict(
+            dn=unicode(self.dn),
+            cn=[self.name],
+            description=[self.description],
+            ipacasubjectdn=[self.ipasubjectdn],
+            ipacaissuerdn=[fuzzy_issuer],
+            ipacaid=[fuzzy_caid],
+            objectclass=objectclasses.ca
+        )
+        self.exists = True
+
+    def make_delete_command(self):
+        """Make function that deletes the plugin entry object."""
+        return self.make_command('ca_del', self.name)
+
+    def check_delete(self, result):
+        assert_deepequal(dict(
+            value=[self.name],
+            summary=u'Deleted CA "{}"'.format(self.name),
+            result=dict(failed=[])
+        ), result)
+
+    def make_retrieve_command(self, all=False, raw=False):
+        """Make function that retrieves the entry using ${CMD}_show"""
+        return self.make_command('ca_show', self.name, all=all, raw=raw)
+
+    def check_retrieve(self, result, all=False, raw=False):
+        """Check the plugin's `show` command result"""
+        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):
+        """Make function that finds the entry using ${CMD}_find
+
+        Note that the name (or other search terms) needs to be specified
+        in arguments.
+        """
+        return self.make_command('ca_find', *args, **kwargs)
+
+    def check_find(self, result, all=False, raw=False):
+        """Check the plugin's `find` command result"""
+        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 matched',
+            result=[expected]
+        ), result)
+
+    def make_update_command(self, updates):
+        """Make function that modifies the entry using ${CMD}_mod"""
+        return self.make_command('ca_mod', self.name, **updates)
+
+    def check_update(self, result, extra_keys=()):
+        """Check the plugin's `find` command result"""
+        assert_deepequal(dict(
+            value=self.name,
+            summary=u'Modified CA "{}"'.format(self.name),
+            result=self.filter_attrs(self.update_keys | set(extra_keys))
+        ), result)
diff --git a/ipatests/test_xmlrpc/xmlrpc_test.py b/ipatests/test_xmlrpc/xmlrpc_test.py
index c3bba9abf..ddfe9c17c 100644
--- a/ipatests/test_xmlrpc/xmlrpc_test.py
+++ b/ipatests/test_xmlrpc/xmlrpc_test.py
@@ -81,6 +81,9 @@ fuzzy_caacldn = Fuzzy(
     '(?i)ipauniqueid=%s,cn=caacls,cn=ca,%s' % (uuid_re, api.env.basedn)
 )
 
+# Matches internal CA ID
+fuzzy_caid = fuzzy_uuid
+
 # Matches fuzzy ipaUniqueID DN group (RDN)
 fuzzy_ipauniqueid = Fuzzy('(?i)ipauniqueid=%s' % uuid_re)