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

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

log = logging.getLogger(__name__)

installation_prefix = '/home/nhosoi/install'

ATTRIBUTE_UNIQUENESS_PLUGIN = 'cn=attribute uniqueness,cn=plugins,cn=config'
ENTRY_NAME = 'test_entry'
    

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'.
        At the beginning, It may exists a standalone instance.
        It may also exists a backup for the standalone instance.
    
        Principle:
            If standalone instance exists:
                restart it
            If backup of standalone exists:
                create/rebind to standalone

                restore standalone instance from backup
            else:
                Cleanup everything
                    remove instance
                    remove backup
                Create instance
                Create backup
    '''
    global installation_prefix

    if installation_prefix:
        args_instance[SER_DEPLOYED_DIR] = installation_prefix
    
    standalone = DirSrv(verbose=True)
    
    # 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 backups
    backup_standalone = standalone.checkBackupFS()
    
    # Get the status of the instance and restart it if it exists
    instance_standalone   = standalone.exists()
    if instance_standalone:
        # assuming the instance is already stopped, just wait 5 sec max
        standalone.stop(timeout=5)
        standalone.start(timeout=10)
    
    if backup_standalone:
        # The backup exist, assuming it is correct 
        # we just re-init the instance with it
        if not instance_standalone:
            standalone.create()
            # Used to retrieve configuration information (dbdir, confdir...)
            standalone.open()
        
        # restore standalone instance from backup
        standalone.stop(timeout=10)
        standalone.restoreFS(backup_standalone)
        standalone.start(timeout=10)
        
    else:
        # We should be here only in two conditions
        #      - This is the first time a test involve standalone instance
        #      - Something weird happened (instance/backup destroyed)
        #        so we discard everything and recreate all
        
        # Remove the backup. So even if we have a specific backup file
        # (e.g backup_standalone) we clear backup that an instance may have created
        if backup_standalone:
            standalone.clearBackupFS()
        
        # Remove the instance
        if instance_standalone:
            standalone.delete()
            
        # Create the instance
        standalone.create()
        
        # Used to retrieve configuration information (dbdir, confdir...)
        standalone.open()
                
        # Time to create the backups
        standalone.stop(timeout=10)
        standalone.backupfile = standalone.backupFS()
        standalone.start(timeout=10)
    
    # 
    # Here we have standalone instance up and running
    # Either coming from a backup recovery
    # or from a fresh (re)init
    # Time to return the topology
    return TopologyStandalone(standalone)


def test_ticket47808_run(topology):
    """
        It enables attribute uniqueness plugin with sn as a unique attribute
        Add an entry 1 with sn = ENTRY_NAME
        Add an entry 2 with sn = ENTRY_NAME
        If the second add does not crash the server and the following search found none,
        the bug is fixed.
    """
        
    # bind as directory manager
    topology.standalone.log.info("Bind as %s" % DN_DM)
    topology.standalone.simple_bind_s(DN_DM, PASSWORD)
    
    topology.standalone.log.info("\n\n######################### SETUP ATTR UNIQ PLUGIN ######################\n")

    # enable attribute uniqueness plugin
    mod = [(ldap.MOD_REPLACE, 'nsslapd-pluginEnabled', 'on'), (ldap.MOD_REPLACE, 'nsslapd-pluginarg0', 'sn'), (ldap.MOD_REPLACE, 'nsslapd-pluginarg1', SUFFIX)]
    topology.standalone.modify_s(ATTRIBUTE_UNIQUENESS_PLUGIN, mod)
    
    topology.standalone.log.info("\n\n######################### ADD USER 1 ######################\n")
    
    # Prepare entry 1
    entry_name = '%s 1' % (ENTRY_NAME)
    entry_dn_1 = 'cn=%s, %s' % (entry_name, SUFFIX)
    entry_1 = Entry(entry_dn_1)
    entry_1.setValues('objectclass', 'top', 'person')
    entry_1.setValues('sn', ENTRY_NAME)
    entry_1.setValues('cn', entry_name)
    topology.standalone.log.info("Try to add Add %s: %r" % (entry_1, entry_1))
    topology.standalone.add_s(entry_1)

    topology.standalone.log.info("\n\n######################### Restart Server ######################\n")
    topology.standalone.stop(timeout=10)
    topology.standalone.start(timeout=10)
    
    topology.standalone.log.info("\n\n######################### ADD USER 2 ######################\n")
    
    # Prepare entry 2 having the same sn, which crashes the server 
    entry_name = '%s 2' % (ENTRY_NAME)
    entry_dn_2 = 'cn=%s, %s' % (entry_name, SUFFIX)
    entry_2 = Entry(entry_dn_2)
    entry_2.setValues('objectclass', 'top', 'person')
    entry_2.setValues('sn', ENTRY_NAME)
    entry_2.setValues('cn', entry_name)
    topology.standalone.log.info("Try to add Add %s: %r" % (entry_2, entry_2))
    try:
      topology.standalone.add_s(entry_2)
    except:
      topology.standalone.log.warn("Adding %s failed" % entry_dn_2)
      pass

    topology.standalone.log.info("\n\n######################### IS SERVER UP? ######################\n")
    ents = topology.standalone.search_s(entry_dn_1, ldap.SCOPE_BASE, '(objectclass=*)')
    assert len(ents) == 1
    topology.standalone.log.info("Yes, it's up.")

    topology.standalone.log.info("\n\n######################### CHECK USER 2 NOT ADDED ######################\n")
    topology.standalone.log.info("Try to search %s" % entry_dn_2)
    try:
        ents = topology.standalone.search_s(entry_dn_2, ldap.SCOPE_BASE, '(objectclass=*)')
    except ldap.NO_SUCH_OBJECT:
        topology.standalone.log.info("Found none")
    
    topology.standalone.log.info("\n\n######################### DELETE USER 1 ######################\n")
    
    topology.standalone.log.info("Try to delete  %s " % entry_dn_1)
    topology.standalone.delete_s(entry_dn_1)
    
def test_ticket47808_final(topology):
    topology.standalone.stop(timeout=10)
    
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 = '/home/nhosoi/install'
        
    topo = topology(True)
    test_ticket47808_run(topo)
    
    test_ticket47808_final(topo)


if __name__ == '__main__':
    run_isolated()