@pytest.fixture(scope="module")

      """Add users to the default suffix

      """

          })

      """Tests basic anonymous search operations

      :id: c7831e04-f458-4e50-83c7-b6f77109f639

  @pytest.mark.ds604

  @pytest.mark.bz915801

      """Test ldapsearch returning original attributes

          using nsslapd-search-return-original-type-switch

  @pytest.fixture(scope="module")

      """User for binding operation"""

      users = UserAccounts(topology_st.standalone, DEFAULT_SUFFIX)

  @pytest.mark.parametrize('add_attr', ['', '*', 'objectClass'])

  @pytest.mark.parametrize('search_suffix,regular_user,oper_attr_list',

                           TEST_PARAMS)

                        search_suffix, regular_user, oper_attr_list):

      """Verify that you can get all expected operational attributes

         by a Search Request [RFC2251] with '+' (ASCII 43) filter.

          assert all(attr in found_attrs

                     for attr in ['objectClass', expected_attrs[0]])

      else:

  if __name__ == '__main__':

  @pytest.fixture(scope="module")

      """User for binding operation"""

      log.info('Adding user simplepaged_test')

  @pytest.mark.parametrize("page_size,users_num", [(6, 5), (5, 5), (5, 25)])

      """Verify that search with a simple paged results control

      returns all entries it should without errors.

      search_flt = r'(uid=test*)'

      searchreq_attrlist = ['dn', 'sn']

      req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

      all_results = paged_search(conn, DEFAULT_SUFFIX, [req_ctrl], search_flt, searchreq_attrlist)

       ldap.SIZELIMIT_EXCEEDED),

      (5, 50, 'cn=config,%s' % DN_LDBM, 'nsslapd-lookthroughlimit', '20',

       ldap.ADMINLIMIT_EXCEEDED)])

                              suffix, attr_name, attr_value, expected_err):

      """Verify that search with a simple paged results control

      throws expected exceptoins when corresponding limits are

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          change_conf_attr(topology_st, suffix, attr_name, attr_value_bck)

      """Verify that search with a simple paged results control

      and a server side sort control returns all entries

      it should without errors.

      searchreq_attrlist = ['dn', 'sn']

      try:

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          sort_ctrl = SSSRequestControl(True, ['sn'])

          del_users(users_list)

      """Verify that search with simple paged results control

      can be abandon

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          del_users(users_list)

      """Verify that after performing multiple simple paged searches

      to completion, each with a timelimit, it wouldn't fail, if we sleep

      for a time more than the timelimit.

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

  @pytest.mark.parametrize('aci_subject',

                           ('dns = "{}"'.format(HOSTNAME),

                            'ip = "{}"'.format(IP_ADDRESS)))

      """Verify that after performing multiple simple paged searches

      to completion on the suffix with DNS or IP based ACI

          ACI_BODY = ensure_bytes(ACI_TARGET + ACI_ALLOW + ACI_SUBJECT)

          topology_st.standalone.modify_s(DEFAULT_SUFFIX, [(ldap.MOD_REPLACE, 'aci', ACI_BODY)])

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          del_users(users_list)

      """Verify that after performing multiple simple paged searches

      on a single connection without a complition, it wouldn't fail.

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

  @pytest.mark.parametrize("invalid_cookie", [1000, -1])

      """Verify that using invalid cookie while performing

      search with the simple paged results control throws

      a TypeError exception

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          del_users(users_list)

      """Verify that search with simple paged results control

      can be abandon using page_size = 0

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          del_users(users_list)

      """Verify that search with a simple paged results control

      returns all entries it should without errors while

      valid value set to nsslapd-pagedsizelimit.

      try:

          log.info('Set user bind')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          controls = [req_ctrl]

  @pytest.mark.parametrize('conf_attr,user_attr,expected_rs',

                           (('5', '15', 'PASS'), ('15', '5', ldap.SIZELIMIT_EXCEEDED)))

                                   conf_attr, user_attr, expected_rs):

      """Verify that nsPagedSizeLimit attribute overrides

      nsslapd-pagedsizelimit while performing search with

      search_flt = r'(uid=test*)'

      searchreq_attrlist = ['dn', 'sn']

      conf_attr_bck = change_conf_attr(topology_st, DN_CONFIG, 'nsslapd-pagedsizelimit', conf_attr)

      try:

          log.info('Set user bind')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          controls = [req_ctrl]

      finally:

          del_users(users_list)

          change_conf_attr(topology_st, DN_CONFIG, 'nsslapd-pagedsizelimit', conf_attr_bck)

  @pytest.mark.parametrize('conf_attr_values,expected_rs',

                           ((('5000', '100', '100'), ldap.ADMINLIMIT_EXCEEDED),

                            (('5000', '120', '122'), 'PASS')))

      """Verify that nsslapd-idlistscanlimit and

      nsslapd-lookthroughlimit can limit the administrator

      search abilities.

      try:

          log.info('Set user bind')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          controls = [req_ctrl]

  @pytest.mark.parametrize('conf_attr_values,expected_rs',

                           ((('1000', '100', '100'), ldap.ADMINLIMIT_EXCEEDED),

                            (('1000', '120', '122'), 'PASS')))

      """Verify that nsPagedIDListScanLimit and nsPagedLookthroughLimit

      override nsslapd-idlistscanlimit and nsslapd-lookthroughlimit

      while performing search with the simple paged results control.

      searchreq_attrlist = ['dn', 'sn']

      lookthrough_attr_bck = change_conf_attr(topology_st, 'cn=config,%s' % DN_LDBM, 'nsslapd-lookthroughlimit', conf_attr_values[0])

      idlistscan_attr_bck = change_conf_attr(topology_st, 'cn=config,%s' % DN_LDBM, 'nsslapd-idlistscanlimit', conf_attr_values[0])

      try:

          log.info('Set user bind')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

          controls = [req_ctrl]

          del_users(users_list)

          change_conf_attr(topology_st, 'cn=config,%s' % DN_LDBM, 'nsslapd-lookthroughlimit', lookthrough_attr_bck)

          change_conf_attr(topology_st, 'cn=config,%s' % DN_LDBM, 'nsslapd-idlistscanlimit', idlistscan_attr_bck)

      """Verify that search with a simple paged results control

      and get effective rights control returns all entries

      it should without errors.

          del_users(users_list)

      """Verify that page result search returns empty cookie

      if there is no returned entry.

  @pytest.mark.parametrize('conf_attr_value', (None, '-1', '1000'))

      """Verify that nsslapd-maxsimplepaged-per-conn acts according design

      :id: 192e2f25-04ee-4ff9-9340-d875dcbe8011

      try:

          log.info('Set user bind')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

  @pytest.mark.parametrize('conf_attr_value', ('0', '1'))

      """Verify that nsslapd-maxsimplepaged-per-conn acts according design

      :id: eb609e63-2829-4331-8439-a35f99694efa

      try:

          log.info('Set user bind')

          log.info('Create simple paged results control instance')

          req_ctrl = SimplePagedResultsControl(True, size=page_size, cookie='')

  @pytest.fixture(scope="module")

      """User for binding operation"""

      log.info('Adding user {}'.format(TEST_USER_DN))

  @pytest.fixture(scope="module")

      """Set up password policy for subtree and user"""

      log.info('Enable fine-grained policy')

                           [('on', 'off', ldap.UNWILLING_TO_PERFORM),

                            ('off', 'off', ldap.UNWILLING_TO_PERFORM),

                            ('off', 'on', None), ('on', 'on', None)])

                      subtree_pwchange, user_pwchange, exception):

      """Verify that 'passwordChange' attr works as expected

      User should have a priority over a subtree.

                                                          ensure_bytes(TEST_USER_PWD))])

      """If we set passwordMinAge to some value, for example to 10, then it

      should not allow the user to change the password within 10 seconds after

      his previous change.

  @pytest.fixture(scope="module")

      """User for binding operation"""

      log.info('Adding user {}'.format(BN))

  @pytest.fixture(scope="module")

      """Set global password policy.

      Then, set fine-grained subtree level password policy

      to ou=People with no password syntax.

  @pytest.mark.parametrize('inherit_value,checksyntax_value',

                           [('off', 'off'), ('on', 'off'), ('off', 'on')])

                                     inherit_value, checksyntax_value):

      """Make sure an entry added to ou=people has no password syntax restrictions

  @pytest.mark.parametrize('container', [DN_CONFIG, PWP_CONTAINER_PEOPLE])

      """Set 'nsslapd-pwpolicy-inherit-global: on' and 'passwordCheckSyntax: on'.

      Make sure that syntax rules work, if set them at both: cn=config and

      ou=people policy container.

  @pytest.fixture(scope="module")

      """Create the test user."""

      topology_st.standalone.add_s(Entry((

      setPolicy(inst, policy_attr, reset_value)

      """Ensure that on a password change, the policy syntax

      is enforced correctly.

  @pytest.fixture(scope="module")

      """Add test users using UserAccounts"""

      log.info('Adding user-uid={},ou=people,{}'.format(user_data['uid'], SUFFIX))

      return tuser

      """Test subtree policies use the same global default for passwordUnlock

      :id: 741a8417-5f65-4012-b9ed-87987ce3ca1b

      """

      log.info("Verify user can bind...")

      log.info('Test passwordUnlock default - user should be able to reset password after lockout')

      for i in range(0, 2):

          try:

          except ldap.INVALID_CREDENTIALS:

              # expected

              pass

      log.info('Verify account is locked')

      with pytest.raises(ldap.CONSTRAINT_VIOLATION):

      log.info('Wait for lockout duration...')

      time.sleep(4)

      log.info('Check if user can now bind with correct password')

  @pytest.mark.bz1465600

  @pytest.mark.parametrize("user_pasw", TEST_PASSWORDS)

      """PasswordCheckSyntax attribute fails to validate cn, sn, uid, givenname, ou and mail attributes

      :id: bf9fe1ef-56cb-46a3-a6f8-5530398a06dc

          4. Resetting userPassword to cn, sn, uid and mail should be rejected.

      """

      try:

          log.info('Replace userPassword attribute with {}'.format(user_pasw))

          with pytest.raises(ldap.CONSTRAINT_VIOLATION) as excinfo:

              log.fatal('Failed: Userpassword with {} is accepted'.format(user_pasw))

          assert 'password based off of user entry' in str(excinfo.value)

      finally:

          conn.unbind_s()

  @pytest.mark.parametrize("user_pasw", TEST_PASSWORDS)

      """Passwords rejected if its similar to uid, cn, sn, givenname, ou and mail attributes

      :id: dfd6cf5d-8bcd-4895-a691-a43ad9ec1be8

      log.info('Configure Pwpolicy with PasswordCheckSyntax and nsslapd-pwpolicy-local set to off')

      topo.standalone.config.set('nsslapd-pwpolicy-local', 'off')

      log.info('Replace userPassword attribute with {}'.format(user_pasw))

      try:

          try:

          except ldap.LDAPError as e:

              log.fatal('Failed to replace userPassword: error {}'.format(e.message['desc']))

              raise e

      finally:

          conn.unbind_s()

  if __name__ == '__main__':

  @pytest.fixture(scope="function")

      """Add test entry to master1"""

      log.info('Adding entry {}'.format(TEST_ENTRY_DN))

      request.addfinalizer(fin)

      """Check that entries are replicated after add operation

      :id: 024250f1-5f7e-4f3b-a9f5-27741e6fd405

      assert all(entries), "Entry {} wasn't replicated successfully".format(TEST_ENTRY_DN)

      """Check that entries are replicated after modify operation

      :id: 36764053-622c-43c2-a132-d7a3ab7d9aaa

          assert "{}@greenhat.com".format(TEST_ENTRY_NAME) not in u.get_attr_vals_utf8('mail')

      """Check that entry deletion is replicated after delete operation

      :id: 18437262-9d6a-4b98-a47a-6182501ab9bc

  @pytest.mark.parametrize("delold", [0, 1])

      """Check that entries are replicated after modrdn operation

      :id: 02558e6d-a745-45ae-8d88-34fe9b16adc9

      repl.remove_master(m1)

      repl.remove_master(m2)

      """Check a replication with many attributes (add and delete)

      :id: d540b358-f67a-43c6-8df5-7c74b3cb7523

              assert value not in delete_list

      """Check that double delete of the entry doesn't crash server

      :ID: 3496c82d-636a-48c9-973c-2455b12164cc

      assert not entries, "Entry deletion {} wasn't replicated successfully".format(TEST_ENTRY_DN)

      """Check that error about userpassword replication is properly logged

      :ID: 714130ff-e4f0-4633-9def-c1f4b24abfef

      M1 = topology.ms["master1"]

      conts = nsContainers(M1, SUFFIX)

      for inst in topology:

          inst.config.loglevel([ErrorLog.DEFAULT, ErrorLog.REPLICA], service='error')

          inst.config.enable_log('audit')

          inst.restart()

      """Delete test container with entries and entry conflicts"""

      for ent in sorted(ents, key=lambda e: len(e.dn), reverse=True):

          log.debug("Delete entry children {}".format(ent.dn))

  @pytest.fixture

      tb = _test_base(topology_m2)

      def fin():

  @pytest.fixture

      tb = _test_base(topology_m3)

      def fin():

  class TestTwoMasters:

          """Check that conflict properly resolved for create - modrdn operations

          :id: 77f09b18-03d1-45da-940b-1ad2c2908ebb

          M1 = topology_m2.ms["master1"]

          M2 = topology_m2.ms["master2"]

          repl = ReplicationManager(SUFFIX)

          for user_num in range(1000, 1005):

          user_dns_m2 = [user.dn for user in test_users_m2.list()]

          assert set(user_dns_m1) == set(user_dns_m2)

          """Check that conflict properly resolved for complex operations

          which involve add, modify, modrdn and delete

          M1 = topology_m2.ms["master1"]

          M2 = topology_m2.ms["master2"]

          repl = ReplicationManager(SUFFIX)

          for user_num in range(1100, 1110):

          user_dns_m2 = [user.dn for user in test_users_m2.list()]

          assert set(user_dns_m1) == set(user_dns_m2)

          """Check that conflict properly resolved for operations

          with memberOf and groups

          M1 = topology_m2.ms["master1"]

          M2 = topology_m2.ms["master2"]

          repl = ReplicationManager(SUFFIX)

          user_dns_m2 = [user.dn for user in test_users_m2.list()]

          assert set(user_dns_m1) == set(user_dns_m2)

          """Check that conflict properly resolved for operations

          with nested entries with children

          M1 = topology_m2.ms["master1"]

          M2 = topology_m2.ms["master2"]

          repl = ReplicationManager(SUFFIX)

          _create_user(test_users_m1, 4000)

          _create_user(test_users_m1, 4001)

          cont_list = []

          for num in range(15):

              cont_list.append(cont)

          repl.test_replication(M1, M2)

          topology_m2.pause_all_replicas()

          log.info("Create parent-child on master2 and master1")

          _create_container(M1, cont_p.dn, 'c0', sleep=True)

          _create_container(M2, cont_p.dn, 'c0', sleep=True)

          log.info("Create parent-child on master1 and master2")

          _create_container(M1, cont_p.dn, 'c1', sleep=True)

          _create_container(M2, cont_p.dn, 'c1', sleep=True)

          log.info("Create parent-child on master1 and master2 different child rdn")

          _create_container(M1, cont_p.dn, 'c2', sleep=True)

          _create_container(M2, cont_p.dn, 'c3', sleep=True)

          for num in range(1, 3):

              inst = topology_m2.ms["master{}".format(num)]

              conts_dns[inst.serverid] = []

              for cont in conts.list():

                  conts_p = nsContainers(inst, cont.dn)

                  for cont_p in conts_p.list():

  class TestThreeMasters:

          """Check that conflict properly resolved for operations

          with nested entries with children

          cont_list = []

          for num in range(11):

              cont_list.append(cont)

          repl.test_replication(M1, M2)

          for num in range(1, 4):

              inst = topology_m3.ms["master{}".format(num)]

              conts_dns[inst.serverid] = []

              for cont in conts.list():

                  conts_p = nsContainers(inst, cont.dn)

                  for cont_p in conts_p.list():

      # -s for DEBUG mode

      CURRENT_FILE = os.path.realpath(__file__)

      pytest.main("-s %s" % CURRENT_FILE)

      return count

  @pytest.fixture()

      """Add test entry using UserAccounts"""

      log.info('Adding a test entry user')