def test_filespace(self):
        """
        pg_basebackup should work with user filespace.

        @tags sanity
        """

        # Add standby entry in catalog before regisering filespace.
        fsbase = os.path.join(self.fsprefix, 'fs')
        shutil.rmtree(fsbase, True)
        os.makedirs(fsbase)
        shutil.rmtree(self.standby_datadir, True)
        dburl = dbconn.DbURL()
        gparray = GpArray.initFromCatalog(dburl, utility=True)
        if gparray.standbyMaster:
            self.standby.remove_catalog_standby(dburl)
        self.standby.add_catalog_standby(dburl, gparray)

        #self.preprocess_file(local_path('filespace.sql.in'))
        sql_file = local_path('filespace.sql')
        from mpp.lib.gpfilespace import Gpfilespace
        gpfile = Gpfilespace()
        gpfile.create_filespace('fs_walrepl_a')
        result = PSQL.run_sql_file(sql_file, dbname=self.db_name)
        self.assertTrue(result)
        subprocess.check_call(['pg_basebackup', '-D', self.standby_datadir])

        #fsdir = os.path.join(self.fsprefix, 'fs', 'gpdb1')
        fsdir = os.path.join(os.path.split(self.standby_datadir)[0], 'fs_walrepl_a','mirror','pg_system')
        self.assertTrue(os.path.isdir(fsdir),
                        '{0} does not dir'.format(fsdir))

 def test_gpinitstandby_prompt_for_filespace(self):
     from mpp.lib.gpfilespace import Gpfilespace
     gpfile = Gpfilespace()
     gpfile.create_filespace('fs_walrepl_a')
     PSQL.run_sql_file(local_path('filespace.sql'), dbname = self.db_name)
     filespace_loc = self.gp.get_filespace_location()
     self.create_directory(filespace_loc)
     self.assertTrue(self.gp.init_with_prompt(filespace_loc))

 def test_gpinitstandby_new_host_with_filespace(self):
     from mpp.lib.gpfilespace import Gpfilespace
     gpfile = Gpfilespace()
     gpfile.create_filespace('fs_walrepl_a')
     PSQL.run_sql_file(local_path('filespace.sql'), dbname = self.db_name)
     filespace_loc = self.gp.get_filespace_location()
     self.create_directory(filespace_loc)
     filespaces = "pg_system:%s,fs_walrepl_a:%s" % (self.mdd, filespace_loc)
     self.assertTrue(self.gp.run(option = '-F %s -s %s -P %s' % (filespaces, self.standby, self.standby_port)))
     self.assertTrue(self.gp.verify_gpinitstandby(self.primary_pid))

 def test_gpinitstandby_to_same_with_filespaces(self):
     from mpp.lib.gpfilespace import Gpfilespace
     gpfile = Gpfilespace()
     gpfile.create_filespace('fs_walrepl_a')
     PSQL.run_sql_file(local_path('filespace.sql'), dbname = self.db_name)
     filespace_loc = self.gp.get_filespace_location() 
     filespace_loc = os.path.join(os.path.split(filespace_loc)[0], 'newstandby')
     filespaces = "pg_system:%s,fs_walrepl_a:%s" %(self.standby_loc , filespace_loc)
     self.assertTrue(self.gp.run(option = '-F %s -s %s -P %s' % (filespaces, self.host, self.standby_port)))
     self.assertTrue(self.gp.verify_gpinitstandby(self.primary_pid))

    def test_standby_is_configured(self):
        """
        Create standby then move transfilespace.
        """

        gpfilespace = Gpfilespace()
        gpfilespace.create_filespace(self.fsname)
        cmd = self.create_local_standby(self.fsdict)
        self.assertIn(cmd.get_results().rc, (0, 1), 'could not create standby')

        gpfilespace.movetransfiles_localfilespace(self.fsname)

        self.transfiles_are_moved = True

    def test_tempfiles_are_moved(self):
        """
        Move tempfilespace then create standby.
        """

        gpfilespace = Gpfilespace()
        gpfilespace.create_filespace(self.fsname)
        gpfilespace.movetempfiles_localfilespace(self.fsname)
        self.tempfiles_are_moved = True

        cmd = self.create_local_standby(self.fsdict)
        self.assertIn(cmd.get_results().rc, (0, 1),
                      'could not create standby with movetempfilespace')
        cmd = self.remove_standby()
        self.assertIn(cmd.get_results().rc, (0, 1),
                      'could not remove standby with movetempfilespace')

 def setUpClass(cls):
     super(GPFilespaceTablespaceTest, cls).setUpClass()
     tinctest.logger.info("*** Running the pre-requisite sql files drop.sql and setup.sql")
     PSQL.run_sql_file(local_path('sqls/setup/drop.sql'))
     #separating dropping of filsepaces
     PSQL.run_sql_file(local_path('sqls/setup/drop_filespaces.sql'))
     PSQL.run_sql_file(local_path('sqls/setup/create.sql'))
     tinctest.logger.info("Starting the Filespace Tablespace test.. ")
     config = GPDBConfig()
     filespace = Gpfilespace()       
     filespace_name = 'cdbfast_fs_'
     if config.is_not_insync_segments():
         tinctest.logger.info("***** Creating filespaces...")
         filespace.create_filespace(filespace_name+'sch1')
         filespace.create_filespace(filespace_name+'sch2')
         filespace.create_filespace(filespace_name+'sch3')

 def test_gpactivatestandby_new_host_with_filespace(self):
     from mpp.lib.gpfilespace import Gpfilespace
     gpfile = Gpfilespace()
     gpfile.create_filespace('fs_walrepl_a')
     PSQL.run_sql_file(local_path('filespace.sql'), dbname= self.db_name)
     gputil.install_standby()
     initstdby = GpinitStandby()
     gpact_stdby = GpactivateStandby()
     self.mdd = gpact_stdby.get_standby_dd()
     self.host = initstdby.get_standbyhost()
     self.port = gpact_stdby.get_standby_port()
     self.standby_pid = gpact_stdby.get_standby_pid(self.host, self.port, self.mdd)
     PSQL.run_sql_file(local_path('create_tables.sql'), dbname = self.db_name)
     self.assertTrue(gpact_stdby.activate())
     self.assertTrue(gpact_stdby.verify_gpactivatestandby(self.standby_pid, self.host, self.port, self.mdd)) 
     gputil.failback_to_original_master(self.origin_mdd,self.host,self.mdd,self.port)

    def setUp(self):
        super(GPExpandTestCase, self).setUp()
        # Doing this in setUp to not impact test construction.
        self.hosts = get_gpexpand_hosts()

        mdd = os.path.join(self.testcase_master_dir, 'gpseg-1')
        os.environ["MASTER_DATA_DIRECTORY"] = mdd

        #initial config has master on HOST1 and segments on HOST2 and HOST3.
        #If we choose to add only segments ie self.number_of_expansion_hosts == 0, in the interview process we say use HOST1 and HOST2
        #if we choose to add expansion hosts ,ie self.number_of_expansion_hosts == 2, in the interview process we say use HOST3 and HOST4
        if self.number_of_expansion_hosts == 0:
            self.expansion_host_list = "%s , %s" %(self.hosts[1], self.hosts[2])
        elif self.number_of_expansion_hosts == 2:
            self.expansion_host_list = "%s , %s" %(self.hosts[3], self.hosts[4])
        if self.gpinitsystem:
            self._do_gpinitsystem()
        if self.standby_enabled:
            self._do_gpinitstandby()
        if self.use_filespaces:
            tinctest.logger.info("Setting filespaces")
            gpfs=Gpfilespace()
            gpfs.create_filespace('expand_filespace')

            res = {'rc': 0, 'stdout' : '', 'stderr': ''}
 
            cmdStr="export MASTER_DATA_DIRECTORY=%s; gpfilespace --movetransfilespace expand_filespace" % (mdd)
            run_shell_command(cmdStr, 'create segment dirs', res)
            if res['rc'] > 0:
                raise GpExpandTestCaseException("Failed to movetransfilespace")

            cmdStr="export MASTER_DATA_DIRECTORY=%s; gpfilespace --movetempfilespace expand_filespace" % (mdd)
            run_shell_command(cmdStr, 'create segment dirs', res)
            if res['rc'] > 0:
                raise GpExpandTestCaseException("Failed to movetempfilespace")

        tinctest.logger.info("Performing setup tasks")
        self._setup_gpexpand()

    def tearDown(self):

        # Remove standby first.
        self.remove_standby(validateAfter=False)

        # If we moved trans/tempfilespace, reset it before dropping
        # user filespace.
        gpfilespace = Gpfilespace()
        if gpfilespace.exists(self.fsname):
            if self.transfiles_are_moved:
                gpfilespace.move_transdefault()
            if self.tempfiles_are_moved:
                gpfilespace.move_tempdefault()
            PSQL.run_sql_command('drop filespace temp_fs')

    def method_setup(self):
        tinctest.logger.info("Performing setup tasks")
        gpfs=Gpfilespace()
        gpfs.create_filespace('filerep_fs_a')
        gpfs.create_filespace('filerep_fs_b')
        gpfs.create_filespace('filerep_fs_c')
        gpfs.create_filespace('filerep_fs_z')
        gpfs.create_filespace('sync1_fs_1') 
 
        # Set max_resource_queues to 100 
        cmd = 'gpconfig -c max_resource_queues -v 100 '
        ok = run_shell_command(cmd)
        if not ok:
            raise Exception('Failure during setting the max_resource_queues value to 100 using gpconfig tool')
        #Restart the cluster
        self.gpstop.run_gpstop_cmd(immediate = 'i')
        ok = self.gpstart.run_gpstart_cmd()
        if not ok:
            raise Exception('Failure during restarting the cluster')
        return True

 def __init__(self, methodName):
     self.gpfile = Gpfilespace()
     self.filereputil = Filerepe2e_Util()
     super(SuspendcheckpointCrashrecoveryTestCase, self).__init__(methodName)

class SuspendcheckpointCrashrecoveryTestCase(ScenarioTestCase):
    """ 
    Testing state of prepared transactions upon crash-recovery
    @gucs gp_create_table_random_default_distribution=off
    """

    def __init__(self, methodName):
        self.gpfile = Gpfilespace()
        self.filereputil = Filerepe2e_Util()
        super(SuspendcheckpointCrashrecoveryTestCase, self).__init__(methodName)

    def setUp(self):
        super(SuspendcheckpointCrashrecoveryTestCase, self).setUp()
        """Create filespace """
        self.gpfile.create_filespace("filespace_test_a")

    def tearDown(self):
        """ Cleanup up the filespace created , reset skip chekpoint fault"""
        self.gpfile.drop_filespace("filespace_test_a")
        port = os.getenv("PGPORT")
        self.filereputil.inject_fault(f="checkpoint", y="reset", r="primary", o="0", p=port)
        super(SuspendcheckpointCrashrecoveryTestCase, self).tearDown()

    def test_crash_recovery_04_to_10(self):
        """ 
        @note : Steps are same as Cdbfast and Previous tinc schedule
        @param skip_state : skip checkpoint
        @param cluster_state : sync/change_tracking/resync
        @param ddl_type : create/drop
        @fault_type : commit/abort . 
        @crash_type : gpstop_i/gpstop_a/failover_to_primary
        @description: Test the state of prepared transactions upon crash-recovery.
                      Faults are used to suspend the transactions before segments flush commit/abort to xlog.
                      Crash followed by recovery are performed to evaluate the transaction state
            Steps:
                0. Check the state of the cluster before proceeding the test execution
                1. Run any fault 'skip checkpoint' before pre_sqls 
                2. Run pre_sqls if any
                3. Run any faults required before the trigger_sqls based on the fault_type as well as cluster_state
                4. Run trigger_sqls - these are the transactions which will be suspended
                5. Crash and recover. 
                6. Run post_sqls to validate whether the transactions at step 4 are commited/ aborted as expected
                7. Recover and Validate using gpcheckcat and gpcheckmirrorseg

        @data_provider data_types_provider
        """
        test_num = self.test_data[0][0] + self.test_data[0][1]
        tinctest.logger.info("\n ===============================================")
        tinctest.logger.info("\n Starting New Test: %s " % test_num)
        tinctest.logger.info("\n ===============================================")
        pass_num = self.test_data[1][0]
        cluster_state = self.test_data[1][1]
        ddl_type = self.test_data[1][2]
        test_type = self.test_data[1][3]
        aborting_create_needed = self.test_data[1][4]

        if test_type == "abort":
            test_dir = "%s_%s_tests" % ("abort", ddl_type)
        elif aborting_create_needed == "True":
            test_dir = "%s_%s_%s_tests" % ("abort", ddl_type, "needed")
        else:
            test_dir = "%s_%s_tests" % (test_type, ddl_type)
        if aborting_create_needed == True and test_type == "commit":
            test_dir = "abort_create_needed_tests"
        elif aborting_create_needed == True and test_type == "abort":
            test_dir = "abort_abort_create_needed_tests"

        tinctest.logger.info("TestDir == %s " % test_dir)
        test_case_list0 = []
        test_case_list0.append("mpp.gpdb.tests.storage.lib.dbstate.DbStateClass.check_system")
        self.test_case_scenario.append(test_case_list0)
        test_case_list1 = []
        test_case_list1.append(
            (
                "mpp.gpdb.tests.storage.crashrecovery.SuspendCheckpointCrashRecovery.set_faults_before_executing_pre_sqls",
                [cluster_state],
            )
        )
        self.test_case_scenario.append(test_case_list1)

        test_case_list2 = []
        test_case_list2.append(
            "mpp.gpdb.tests.storage.crashrecovery.%s.pre_sql.test_pre_sqls.TestPreSQLClass" % test_dir
        )
        self.test_case_scenario.append(test_case_list2)

        test_case_list3 = []
        test_case_list3.append(
            (
                "mpp.gpdb.tests.storage.crashrecovery.SuspendCheckpointCrashRecovery.set_faults_before_executing_trigger_sqls",
                [pass_num, cluster_state, test_type, ddl_type, aborting_create_needed],
            )
        )
        self.test_case_scenario.append(test_case_list3)

        test_case_list4 = []
        test_case_list4.append(
            "mpp.gpdb.tests.storage.crashrecovery.%s.trigger_sql.test_triggersqls.TestTriggerSQLClass" % test_dir
        )
        test_case_list4.append(
#.........这里部分代码省略.........

 def __init__(self, methodName):
     tinctest.logger.info("\n =====================In __init__ method ==========================")
     self.gpfile = Gpfilespace()
     self.filereputil = Filerepe2e_Util()
     super(SuspendcheckpointCrashrecoveryTestCase,self).__init__(methodName)

 def method_setup(self):
     tinctest.logger.info("Performing setup tasks")
     gpfs=Gpfilespace()
     gpfs.create_filespace('subt_filespace_a')

 def __init__(self, methodName):
     self.gpfile = Gpfilespace()
     super(SwitchCheckpointTestCase,self).__init__(methodName)

class PgtwoPhaseTestCase(ScenarioTestCase, MPPTestCase):
    ''' Testing state of prepared transactions upon crash-recovery'''

    def __init__(self, methodName):
        self.gpfile = Gpfilespace()
        self.filereputil = Filerepe2e_Util()
        super(PgtwoPhaseTestCase,self).__init__(methodName)
    
    def setUp(self):
        '''Create filespace '''
        self.gpfile.create_filespace('filespace_test_a')
        super(PgtwoPhaseTestCase,self).setUp()
        
    def tearDown(self):
        ''' Cleanup up the filespace created , reset skip chekpoint fault'''
        self.gpfile.drop_filespace('filespace_test_a')
        port = os.getenv('PGPORT')
        self.filereputil.inject_fault(f='checkpoint', y='reset', r='primary', o='0', p=port)
        super(PgtwoPhaseTestCase,self).tearDown()

    def execute_split_sqls(self, skip_state, cluster_state, ddl_type, fault_type, crash_type):
        ''' 
        @param skip_state : skip/noskip checkpoint
        @param cluster_state : sync/change_tracking/resync
        @param ddl_type : create/drop
        @fault_type : commit/abort . Uses the same parameter to pass in 'end_prepare_two_phase_sleep'
        @crash_type : gpstop_i/gpstop_a/failover_to_primary/failover_to_mirror
        @description: Test the state of prepared transactions upon crash-recovery. 
                      Faults are used to suspend the transactions before segments flush commit/abort to xlog. 
                      Different types of crash followed by recovery are performed to evaluate the transaction state
            Steps:
                0. Check the state of the cluster before proceeding the test execution
                1. Run any faults before pre_sqls depending on the clsuter_state 
                2. Run pre_sqls if any
                3. Run any faults required before the trigger_sqls based on the fault_type as well as cluster_state
                4. Run trigger_sqls - these are the transactions which will be suspended
                5. Crash and recover. Resume suspended faults if needed
                6. Run post_sqls to validate whether the transactions at step 4 are commited/ aborted as expected
                7. Recover the cluster in case if neeed
                8. Validate using gpcheckcat and gpcheckmirrorseg

        '''
        if fault_type == 'end_prepare_two_phase_sleep':
            test_dir = '%s_%s_tests' % ('abort', ddl_type)
        else:
            test_dir = '%s_%s_tests' % (fault_type, ddl_type)
        
        tinctest.logger.info('fault_type %s test_dir %s ddl_type %s' % (fault_type, test_dir, ddl_type))
        test_case_list0 = []
        test_case_list0.append('mpp.gpdb.tests.storage.lib.dbstate.DbStateClass.check_system')
        self.test_case_scenario.append(test_case_list0)

        test_case_list1 = []
        test_case_list1.append(('mpp.gpdb.tests.storage.pg_twophase.PgtwoPhaseClass.run_faults_before_pre', [cluster_state]))
        self.test_case_scenario.append(test_case_list1)

        test_case_list2 = []
        test_case_list2.append('mpp.gpdb.tests.storage.pg_twophase.%s.pre_sql.test_presqls.TestPreSQLClass' % test_dir)
        self.test_case_scenario.append(test_case_list2)

        test_case_list3 = []
        test_case_list3.append(('mpp.gpdb.tests.storage.pg_twophase.PgtwoPhaseClass.run_faults_before_trigger', [skip_state, cluster_state, fault_type]))
        self.test_case_scenario.append(test_case_list3)
        
        test_case_list4 = []
        test_case_list4.append('mpp.gpdb.tests.storage.pg_twophase.%s.trigger_sql.test_triggersqls.TestTriggerSQLClass' % test_dir)
        test_case_list4.append(('mpp.gpdb.tests.storage.pg_twophase.PgtwoPhaseClass.run_crash_and_recover', [crash_type, fault_type, test_dir, cluster_state, skip_state]))
        self.test_case_scenario.append(test_case_list4)

        test_case_list5 = []
        test_case_list5.append('mpp.gpdb.tests.storage.pg_twophase.%s.post_sql.test_postsqls.TestPostSQLClass' % test_dir)
        self.test_case_scenario.append(test_case_list5)

        test_case_list6 = []
        test_case_list6.append(('mpp.gpdb.tests.storage.pg_twophase.PgtwoPhaseClass.run_gprecover', [crash_type, cluster_state]))
        self.test_case_scenario.append(test_case_list6)

        test_case_list7 = []
        test_case_list7.append('mpp.gpdb.tests.storage.lib.dbstate.DbStateClass.run_validation')
        self.test_case_scenario.append(test_case_list7)
        
        test_case_list8 = []
        test_case_list8.append('mpp.gpdb.tests.storage.pg_twophase.PgtwoPhaseClass.cleanup_dangling_processes')
        self.test_case_scenario.append(test_case_list8)

 def _create_filespace(self, fsname=None):
     if fsname is None:
         tinctest.logger.warning("Please specify a filespace name to create") 
     else:
         gpfs=Gpfilespace()
         gpfs.create_filespace(fsname)    

class GpFilespaceRegressionTests(unittest.TestCase):

    def __init__(self, methodName):
        self.gpfs = Gpfilespace()
        self.gpfs_h = HAWQGpfilespace()
        super(GpFilespaceRegressionTests, self).__init__(methodName)

    def tearDown(self):
        PSQL.run_sql_command('Drop filespace test_fs_a;')

    def test_create_filespace(self):
        self.gpfs.create_filespace('test_fs_a')
        fs_list = PSQL.run_sql_command("select fsname from pg_filespace where fsname<>'pg_system';", flags = '-q -t')
        self.assertTrue('test_fs_a' in fs_list)

    def test_drop_fiespace(self):
        self.gpfs.create_filespace('test_fs_b')
        self.assertTrue(self.gpfs.drop_filespace('test_fs_b'))

    def test_fs_exists(self):
        self.gpfs.create_filespace('test_fs_a')
        self.assertTrue(self.gpfs.exists('test_fs_a'))

    def test_showtempfiles(self):
        result = self.gpfs.showtempfiles()
        show = False
        for line in result.stdout.splitlines():
            if 'Current Filespace for TEMPORARY_FILES' in line:
                show = True
        self.assertTrue(show)
    
    def test_get_filespace_location(self):
        result = self.gpfs.get_filespace_location()
        self.assertTrue(len(result) >0)
    
    def test_get_filespace_directory(self):
        result = self.gpfs.get_filespace_directory()
        self.assertTrue(len(result) >0)

    def test_get_hosts_for_filespace(self):
        self.gpfs.create_filespace('test_fs_a')
        fs_location = PSQL.run_sql_command("select fselocation  from pg_filespace_entry where fselocation like '%test_fs_a%' and fsedbid=2;", flags = '-q -t')
        result = self.gpfs.get_hosts_for_filespace(fs_location.strip())
        self.assertEquals(result[0]['location'],fs_location.strip())
        
    def test_create_filespace_hawq(self):
        self.gpfs_h.create_filespace('test_fs_hq')
        fs_list = PSQL.run_sql_command("select fsname from pg_filespace where fsname<>'pg_system';", flags = '-q -t')
        self.assertTrue('test_fs_hq' in fs_list)

 def __init__(self, methodName):
     self.gpfs = Gpfilespace()
     self.gpfs_h = HAWQGpfilespace()
     super(GpFilespaceRegressionTests, self).__init__(methodName)