From fe8547bc0ba8e79ae5cc34154d1ea930875c6401 Mon Sep 17 00:00:00 2001
From: Anuj Borah <aborah@redhat.com>
Date: Apr 22 2020 07:57:16 +0000
Subject: Issue: 48055 - CI test - automember_plugin(part3)


Bug Description: CI test - automember_plugin(part3)

Relates: https://pagure.io/389-ds-base/issue/48055

Author: aborah

Reviewed by: Viktor Ashirov

---

diff --git a/dirsrvtests/tests/suites/automember_plugin/basic_test.py b/dirsrvtests/tests/suites/automember_plugin/basic_test.py
index 85d3224..dd1c8e5 100644
--- a/dirsrvtests/tests/suites/automember_plugin/basic_test.py
+++ b/dirsrvtests/tests/suites/automember_plugin/basic_test.py
@@ -12,18 +12,22 @@ Will test AutoMememer Plugin with AotoMember Task and Retro Changelog
 
 import os
 import pytest
+import time
+import re
 from lib389.topologies import topology_m1 as topo
 from lib389.idm.organizationalunit import OrganizationalUnits
 from lib389.idm.domain import Domain
 from lib389.idm.posixgroup import PosixGroups
 from lib389.plugins import AutoMembershipPlugin, AutoMembershipDefinitions, \
-    MemberOfPlugin, AutoMembershipRegexRules, AutoMembershipDefinition
+    MemberOfPlugin, AutoMembershipRegexRules, AutoMembershipDefinition, RetroChangelogPlugin
 from lib389.backend import Backends
 from lib389.config import Config
 from lib389._constants import DEFAULT_SUFFIX
 from lib389.idm.user import UserAccounts
 from lib389.idm.group import Groups, Group, UniqueGroup, nsAdminGroups, nsAdminGroup
+from lib389.tasks import Tasks, AutomemberRebuildMembershipTask, ExportTask
 from lib389.utils import ds_is_older
+from lib389.paths import Paths
 import ldap
 
 pytestmark = pytest.mark.tier1
@@ -32,6 +36,7 @@ BASE_SUFF = "dc=autoMembers,dc=com"
 TEST_BASE = "dc=testAutoMembers,dc=com"
 BASE_REPL = "dc=replAutoMembers,dc=com"
 SUBSUFFIX = f'dc=SubSuffix,{BASE_SUFF}'
+PLUGIN_AUTO = "cn=Auto Membership Plugin,cn=plugins,cn=config"
 REPMANDN = "cn=ReplManager"
 CACHE_SIZE = '-1'
 CACHEMEM_SIZE = '10485760'
@@ -849,6 +854,332 @@ def test_reject_invalid_config_and_we_donot_deadlock_the_server(topo, _create_al
     topo.ms["master1"] .restart()
 
 
+@pytest.fixture(scope="module")
+def _startuptask(topo):
+    """
+    Fixture module that will change required entries for test cases.
+    """
+    for Configs in ["cn=Managers,cn=subsuffGroups",
+                    "cn=Contractors,cn=subsuffGroups",
+                    "cn=testuserGroups",
+                    "cn=subsuffGroups",
+                    "cn=hostGroups"]:
+        AutoMembershipDefinition(topo.ms["master1"], f'{Configs},{PLUGIN_AUTO}').delete()
+    AutoMembershipDefinition(topo.ms["master1"], "cn=userGroups,{}".format(PLUGIN_AUTO)).replace(
+        'autoMemberScope', 'ou=TaskEmployees,dc=autoMembers,dc=com')
+    topo.ms['master1'].restart()
+
+
+@pytest.fixture(scope="function")
+def _fixture_for_build_task(request, topo):
+    def finof():
+        master = topo.ms['master1']
+        auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+        for user in nsAdminGroups(master, auto_mem_scope, rdn=None).list():
+            user.delete()
+
+    request.addfinalizer(finof)
+
+
+def bulk_check_groups(topo, GROUP_DN, MEMBATTR, TOTAL_MEM):
+    assert len(nsAdminGroup(topo, GROUP_DN).get_attr_vals_utf8(MEMBATTR)) == TOTAL_MEM
+
+
+def test_automemtask_re_build_task(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff973a8-e7ff-11e8-a89b-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users and enable autoMembers plug-in
+        2. Run automembers re-build task to create the member attributes
+        3. Search for any error logs
+    :expected results:
+        1. Success
+        2. Success
+        3. Success
+    """
+    master = topo.ms['master1']
+    testid = "autoMemTask_01"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    contract_grp = "cn=Contractors,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # make sure the retro changelog is disabled
+    RetroChangelogPlugin(master).disable()
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for i in range(10):
+        add_user(topo, "{}{}".format(user_rdn, str(i)), auto_mem_scope, str(1188), str(1189), "Manager")
+    for grp in (managers_grp, contract_grp):
+        with pytest.raises(AssertionError):
+            assert check_groups(topo, grp, f'uid=User_autoMemTask_010,{auto_mem_scope}', 'member')
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    error_string = "automember_rebuild_task_thread"
+    AutomemberRebuildMembershipTask(master).create(properties={
+        'basedn': auto_mem_scope,
+        'filter': "objectClass=posixAccount"
+    })
+    # Search for any error logs
+    assert not master.searchErrorsLog(error_string)
+    for grp in (managers_grp, contract_grp):
+        bulk_check_groups(master, grp, "member", 10)
+
+
+def ldif_check_groups(USERS_DN, MEMBATTR, TOTAL_MEM, LDIF_FILE):
+    study = open('{}'.format(LDIF_FILE), 'r')
+    study_ready = study.read()
+    assert len(re.findall("{}: {}".format(MEMBATTR, USERS_DN.lower()), study_ready)) == TOTAL_MEM
+
+
+def check_file_exists(export_ldif):
+    count = 0
+    while not os.path.exists(export_ldif) and count < 3:
+        time.sleep(1)
+        count += 1
+
+    count = 0
+    while (os.stat(export_ldif).st_size == 0) and count < 3:
+        time.sleep(1)
+        count += 1
+
+    if os.path.exists(export_ldif) and os.stat(export_ldif).st_size != 0:
+        return True
+    else:
+        return False
+
+
+def test_automemtask_export_task(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff98b18-e7ff-11e8-872a-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users and enable autoMembers plug-in
+        2. Run automembers export task to create an ldif file with member attributes
+    :expected results:
+        1. Success
+        2. Success
+    """
+    master = topo.ms['master1']
+    p = Paths('master1')
+    testid = "autoMemTask_02"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # Disabling plugin
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for i in range(10):
+        add_user(topo, "{}{}".format(user_rdn, str(i)), auto_mem_scope, str(2788), str(2789), "Manager")
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    export_ldif = p.backup_dir + "/Out_Export_02.ldif"
+    if os.path.exists(export_ldif):
+        os.remove(export_ldif)
+    exp_task = Tasks(master)
+    exp_task.automemberExport(suffix=auto_mem_scope, fstr='objectclass=posixAccount', ldif_out=export_ldif)
+    check_file_exists(export_ldif)
+    ldif_check_groups("cn={}".format(user_rdn), "member", 10, export_ldif)
+    os.remove(export_ldif)
+
+
+def test_automemtask_mapping(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff9a206-e7ff-11e8-bf59-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users and enable autoMembers plug-in
+        2. Run automembers Mapping task with input/output ldif files
+    :expected results:
+        1. Should success
+        2. Should success
+    """
+    master = topo.ms['master1']
+    p = Paths('master1')
+    testid = "autoMemTask_02"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    export_ldif = p.backup_dir+"/Out_Export_02.ldif"
+    output_ldif3 = p.backup_dir+"/Output_03.ldif"
+    for file in [export_ldif, output_ldif3]:
+        if os.path.exists(file):
+            os.remove(file)
+    for i in range(10):
+        add_user(topo, "{}{}".format(user_rdn, str(i)), auto_mem_scope, str(2788), str(2789), "Manager")
+    ExportTask(master).export_suffix_to_ldif(ldiffile=export_ldif, suffix=BASE_SUFF)
+    check_file_exists(export_ldif)
+    map_task = Tasks(master)
+    map_task.automemberMap(ldif_in=export_ldif, ldif_out=output_ldif3)
+    check_file_exists(output_ldif3)
+    ldif_check_groups("cn={}".format(user_rdn), "member", 10, output_ldif3)
+    for file in [export_ldif, output_ldif3]:
+        os.remove(file)
+
+
+def test_automemtask_re_build(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff9b944-e7ff-11e8-ad35-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users with inetOrgPerson object class
+        2. Run automembers re-build task to create the member attributes, exp to FAIL
+    :expected results:
+        1. Should success
+        2. Should not success
+    """
+    master = topo.ms['master1']
+    testid = "autoMemTask_04"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # Disabling plugin
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for number in range(10):
+        add_user(topo, f'{user_rdn}{number}', auto_mem_scope, str(number), str(number), "Manager")
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+    # Enabling plugin
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    AutomemberRebuildMembershipTask(master).create(properties={
+        'basedn': auto_mem_scope,
+        'filter': "objectClass=inetOrgPerson"
+    })
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+
+
+def test_automemtask_export(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff9cf74-e7ff-11e8-b712-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users with inetOrgPerson objectClass
+        2. Run automembers export task to create an ldif file with member attributes, exp to FAIL
+    :expected results:
+        1. Should success
+        2. Should not success
+    """
+    master = topo.ms['master1']
+    p = Paths('master1')
+    testid = "autoMemTask_05"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # Disabling plugin
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for number in range(10):
+        add_user(topo, f'{user_rdn}{number}', auto_mem_scope, str(number), str(number), "Manager")
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+    # Enabling plugin
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    export_ldif = p.backup_dir + "/Out_Export_02.ldif"
+    if os.path.exists(export_ldif):
+        os.remove(export_ldif)
+    exp_task = Tasks(master)
+    exp_task.automemberExport(suffix=auto_mem_scope, fstr='objectclass=inetOrgPerson', ldif_out=export_ldif)
+    check_file_exists(export_ldif)
+    with pytest.raises(AssertionError):
+        ldif_check_groups("uid={}".format(user_rdn), "member", 10, export_ldif)
+    os.remove(export_ldif)
+
+
+def test_automemtask_run_re_build(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff9e5c2-e7ff-11e8-943e-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users with inetOrgPerson obj class
+        2. Change plugin config
+        3. Enable plug-in and run re-build task to create the member attributes
+    :expected results:
+        1. Should success
+        2. Should success
+        3. Should success
+    """
+    master = topo.ms['master1']
+    p = Paths('master1')
+    testid = "autoMemTask_06"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # Disabling plugin
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for number in range(10):
+        add_user(topo, f'{user_rdn}{number}', auto_mem_scope, '111', '111', "Manager")
+    for user in nsAdminGroups(master, auto_mem_scope, rdn=None).list():
+        user.add('objectclass', 'inetOrgPerson')
+    AutoMembershipDefinition(master,
+                             f'cn=userGroups,{PLUGIN_AUTO}').replace('autoMemberFilter',
+                                                                     "objectclass=inetOrgPerson")
+    master.restart()
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    AutomemberRebuildMembershipTask(master).create(properties={
+        'basedn': auto_mem_scope,
+        'filter': "objectClass=inetOrgPerson"})
+    time.sleep(2)
+    bulk_check_groups(master, managers_grp, "member", 10)
+    AutoMembershipDefinition(master,
+                             f'cn=userGroups,{PLUGIN_AUTO}').replace('autoMemberFilter',
+                                                                     "objectclass=posixAccount")
+    master.restart()
+
+
+def test_automemtask_run_export(topo, _create_all_entries, _startuptask, _fixture_for_build_task):
+    """
+    :id: 4ff9fba2-e7ff-11e8-a5ec-8c16451d917b
+    :setup: 4 Instances with replication
+    :steps:
+        1. Add 10 users with inetOrgPerson objectClass
+        2. change plugin config
+        3. Run export task to create an ldif file with member attributes
+    :expected results:
+        1. Should success
+        2. Should success
+        3. Should success
+    """
+    master = topo.ms['master1']
+    p = Paths('master1')
+    testid = "autoMemTask_07"
+    auto_mem_scope = "ou=TaskEmployees,{}".format(BASE_SUFF)
+    managers_grp = "cn=Managers,ou=userGroups,{}".format(BASE_SUFF)
+    user_rdn = "User_{}".format(testid)
+    # Disabling plugin
+    AutoMembershipPlugin(master).disable()
+    master.restart()
+    for number in range(10):
+        add_user(topo, f'{user_rdn}{number}', auto_mem_scope, '222', '222', "Manager")
+    for user in nsAdminGroups(master, auto_mem_scope, rdn=None).list():
+        user.add('objectclass', 'inetOrgPerson')
+    AutoMembershipDefinition(master, f'cn=userGroups,{PLUGIN_AUTO}').replace('autoMemberFilter',
+                                                                             "objectclass=inetOrgPerson")
+    master.restart()
+    # Enabling plugin
+    AutoMembershipPlugin(master).enable()
+    master.restart()
+    with pytest.raises(AssertionError):
+        bulk_check_groups(master, managers_grp, "member", 10)
+    export_ldif = p.backup_dir + "/Out_Export_02.ldif"
+    if os.path.exists(export_ldif):
+        os.remove(export_ldif)
+    exp_task = Tasks(master)
+    exp_task.automemberExport(suffix=auto_mem_scope, fstr='objectclass=inetOrgPerson', ldif_out=export_ldif)
+    check_file_exists(export_ldif)
+    ldif_check_groups("cn={}".format(user_rdn), "member", 10, export_ldif)
+    AutoMembershipDefinition(master, f'cn=userGroups,{PLUGIN_AUTO}').\
+        replace('autoMemberFilter', "objectclass=posixAccount")
+
+
 if __name__ == "__main__":
     CURRENT_FILE = os.path.realpath(__file__)
-    pytest.main("-s -v %s" % CURRENT_FILE)
+    pytest.main("-s -v %s" % CURRENT_FILE)
\ No newline at end of file

