389-ds-base/dirsrvtests/tickets/ticket47819_test.py

import os
import sys
import time
import ldap
import logging
import pytest
from lib389 import DirSrv, Entry, tools, tasks
from lib389.tools import DirSrvTools
from lib389._constants import *
from lib389.properties import *
from lib389.tasks import *

log = logging.getLogger(__name__)

installation_prefix = None


class TopologyStandalone(object):
    def __init__(self, standalone):
        standalone.open()
        self.standalone = standalone


@pytest.fixture(scope="module")
def topology(request):
    '''
        This fixture is used to standalone topology for the 'module'.
    '''
    global installation_prefix

    if installation_prefix:
        args_instance[SER_DEPLOYED_DIR] = installation_prefix

    standalone = DirSrv(verbose=False)

    # Args for the standalone instance
    args_instance[SER_HOST] = HOST_STANDALONE
    args_instance[SER_PORT] = PORT_STANDALONE
    args_instance[SER_SERVERID_PROP] = SERVERID_STANDALONE
    args_standalone = args_instance.copy()
    standalone.allocate(args_standalone)

    # Get the status of the instance and restart it if it exists
    instance_standalone = standalone.exists()

    # Remove the instance
    if instance_standalone:
        standalone.delete()

    # Create the instance
    standalone.create()

    # Used to retrieve configuration information (dbdir, confdir...)
    standalone.open()

    # clear the tmp directory
    standalone.clearTmpDir(__file__)

    # Here we have standalone instance up and running
    return TopologyStandalone(standalone)


def test_ticket47819(topology):
    """
        Testing precise tombstone purging:
            [1]  Make sure "nsTombstoneCSN" is added to new tombstones
            [2]  Make sure an import of a replication ldif adds "nsTombstoneCSN"
                 to old tombstones
            [4]  Test fixup task
            [3]  Make sure tombstone purging works
    """

    log.info('Testing Ticket 47819 - Test precise tombstone purging')

    #
    # Setup Replication
    #
    log.info('Setting up replication...')
    topology.standalone.replica.enableReplication(suffix=DEFAULT_SUFFIX, role=REPLICAROLE_MASTER,
                                                  replicaId=REPLICAID_MASTER_1)

    #
    # Part 1 create a tombstone entry and make sure nsTombstoneCSN is added
    #
    log.info('Part 1:  Add and then delete an entry to create a tombstone...')

    try:
        topology.standalone.add_s(Entry(('cn=entry1,dc=example,dc=com', {
                                  'objectclass': 'top person'.split(),
                                  'sn': 'user',
                                  'cn': 'entry1'})))
    except ldap.LDAPError, e:
        log.error('Failed to add entry: ' + e.message['desc'])
        assert False

    try:
        topology.standalone.delete_s('cn=entry1,dc=example,dc=com')
    except ldap.LDAPError, e:
        log.error('Failed to delete entry: ' + e.message['desc'])
        assert False

    log.info('Search for tombstone entries...')
    try:
        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
        if not entries:
            log.fatal('Search failed to the new tombstone(nsTombstoneCSN is probably missing).')
            assert False
    except ldap.LDAPError, e:
        log.fatal('Search failed: ' + e.message['desc'])
        assert False

    log.info('Part 1 - passed')

    #
    # Part 2 - import ldif with tombstones missing 'nsTombstoneCSN'
    #
    # First, export the replication ldif, edit the file(remove nstombstonecsn),
    # and reimport it.
    #
    log.info('Part 2:  Exporting replication ldif...')

    # Get the the full path and name for our LDIF we will be exporting
    ldif_file = topology.standalone.getDir(__file__, TMP_DIR) + "export.ldif"

    args = {EXPORT_REPL_INFO: True,
            TASK_WAIT: True}
    exportTask = Tasks(topology.standalone)
    try:
        exportTask.exportLDIF(DEFAULT_SUFFIX, None, ldif_file, args)
    except ValueError:
        assert False

    # open the ldif file, get the lines, then rewrite the file
    ldif = open(ldif_file, "r")
    lines = ldif.readlines()
    ldif.close()

    ldif = open(ldif_file, "w")
    for line in lines:
        if not line.lower().startswith('nstombstonecsn'):
            ldif.write(line)
    ldif.close()

    # import the new ldif file
    log.info('Import replication LDIF file...')
    importTask = Tasks(topology.standalone)
    args = {TASK_WAIT: True}
    try:
        importTask.importLDIF(DEFAULT_SUFFIX, None, ldif_file, args)
        os.remove(ldif_file)
    except ValueError:
        os.remove(ldif_file)
        assert False

    # Search for the tombstone again
    log.info('Search for tombstone entries...')
    try:
        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
        if not entries:
            log.fatal('Search failed to fine the new tombstone(nsTombstoneCSN is probably missing).')
            assert False
    except ldap.LDAPError, e:
        log.fatal('Search failed: ' + e.message['desc'])
        assert False

    log.info('Part 2 - passed')

    #
    # Part 3 - test fixup task
    #
    log.info('Part 4:  test the fixup task')

    # Run fixup task using the strip option.  This removes nsTombstoneCSN
    # so we can test if the fixup task works.
    args = {TASK_WAIT: True,
            TASK_TOMB_STRIP: True}
    fixupTombTask = Tasks(topology.standalone)
    try:
        fixupTombTask.fixupTombstones(DEFAULT_BENAME, args)
    except:
        assert False

    # Search for tombstones with nsTombstoneCSN - better not find any
    log.info('Search for tombstone entries...')
    try:
        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
        if entries:
            log.fatal('Search found tombstones with nsTombstoneCSN')
            assert False
    except ldap.LDAPError, e:
        log.fatal('Search failed: ' + e.message['desc'])
        assert False

    # Now run the fixup task
    args = {TASK_WAIT: True}
    fixupTombTask = Tasks(topology.standalone)
    try:
        fixupTombTask.fixupTombstones(DEFAULT_BENAME, args)
    except:
        assert False

    # Search for tombstones with nsTombstoneCSN - better find some
    log.info('Search for tombstone entries...')
    try:
        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
        if not entries:
            log.fatal('Search did not find any fixed-up tombstones')
            assert False
    except ldap.LDAPError, e:
        log.fatal('Search failed: ' + e.message['desc'])
        assert False

    log.info('Part 3 - passed')

    #
    # Part 4 - Test tombstone purging
    #
    log.info('Part 4:  test tombstone purging...')

    args = {REPLICA_PRECISE_PURGING: 'on',
            REPLICA_PURGE_DELAY: '5',
            REPLICA_PURGE_INTERVAL: '5'}
    try:
        topology.standalone.replica.setProperties(DEFAULT_SUFFIX, None, None, args)
    except:
        log.fatal('Failed to configure replica')
        assert False

    # Wait for the interval to pass
    log.info('Wait for tombstone purge interval to pass...')
    time.sleep(6)

    # Add an entry to trigger replication
    log.info('Perform an update to help trigger tombstone purging...')
    try:
        topology.standalone.add_s(Entry(('cn=test_entry,dc=example,dc=com', {
                                  'objectclass': 'top person'.split(),
                                  'sn': 'user',
                                  'cn': 'entry1'})))
    except ldap.LDAPError, e:
        log.error('Failed to add entry: ' + e.message['desc'])
        assert False

    # Wait for the interval to pass again
    log.info('Wait for tombstone purge interval to pass again...')
    time.sleep(10)

    # search for tombstones, there should be none
    log.info('Search for tombstone entries...')
    try:
        entries = topology.standalone.search_s(DEFAULT_SUFFIX, ldap.SCOPE_SUBTREE,
                                               '(&(nsTombstoneCSN=*)(objectclass=nsTombstone))')
        if entries:
            log.fatal('Search unexpectedly found tombstones')
            assert False
    except ldap.LDAPError, e:
        log.fatal('Search failed: ' + e.message['desc'])
        assert False

    log.info('Part 4 - passed')


def test_ticket47819_final(topology):
    topology.standalone.delete()
    log.info('Testcase PASSED')


def run_isolated():
    '''
        run_isolated is used to run these test cases independently of a test scheduler (xunit, py.test..)
        To run isolated without py.test, you need to
            - edit this file and comment '@pytest.fixture' line before 'topology' function.
            - set the installation prefix
            - run this program
    '''
    global installation_prefix
    installation_prefix = None

    topo = topology(True)
    test_ticket47819(topo)
    test_ticket47819_final(topo)

if __name__ == '__main__':
    run_isolated()