Python testlib.save_output函数代码示例

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› Python›Python编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Python中testlib.save_output函数的典型用法代码示例。如果您正苦于以下问题：Python save_output函数的具体用法？Python save_output怎么用？Python save_output使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。

在下文中一共展示了save_output函数的19个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: run

    def run(self):

        # try reading various ranges (these are (start, end) absolute ranges, not offset/length)
        ranges = [
            (0, 16384),
            (1, 200),
            (0, 4096),
            (0, 8192),
            (0, 1000),
            (0, 6000),
            (100, 4000),
            (5000, 10000),
            (4096, 10000),
            (5000, 8192),
            (4096, 16834),
            (5000, 16384),
            (4096, 16000),
            (5000, 16000)
        ]

        self.exitcode = 0

        for (start, end) in ranges:

            if self.errors:
                break

            # only clear reader's cache
            testlib.clear_cache( self.config_dir, volume_id=self.volume_id, gateway_id=self.read_gateway_id )

            for i in xrange(0, 2):

                if self.errors:
                    break

                exitcode, out = testlib.run( READ_PATH, '-d2', '-f', '-c', os.path.join(self.config_dir, 'syndicate.conf'),
                                            '-u', testconf.SYNDICATE_ADMIN, '-v', self.volume_name, '-g', self.read_gateway_name,
                                            self.output_path, start, end - start, valgrind=True )

                out_name = "uncached"
                if i > 0:
                    out_name = "cached"

                testlib.save_output( self.output_dir, 'syndicate-read-thread-%s-%s-%s-%s' % (self.get_ident(), start, end, out_name), out )
                if exitcode != 0:
                    self.exitcode = exitcode
                    self.errormsg = "syndicate-read exit code %s on %s-%s-%s" % (exitcode, start, end, out_name)
                    break

                # correctness 
                if expected_data[start:end] not in out:
                    self.exitcode = -1
                    self.errormsg = "Thread %s missing data for %s-%s-%s" % (self.get_ident(), start, end, out_name)
                    break

            
            if self.exitcode != 0:
                break

        return

开发者ID:iychoi，项目名称:syndicate-core，代码行数:60，代码来源:syndicate-ag-read-parallel.py

示例2: hex

    path = '/setxattr-%s' % random_part
    attr_name_base = 'foo-%s' % random_part
    attr_value_base = 'bar-%s' % random_part

    config_dir, output_dir = testlib.test_setup()
    volume_name = testlib.add_test_volume( config_dir )

    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    random_part = hex(random.randint(0, 2**32-1))[2:]
   
    exitcode, out = testlib.run( TOUCH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,
                                path )

    testlib.save_output( output_dir, "syndicate-touch", out )

    if exitcode != 0:
        raise Exception("Failed to touch %s" % path)

    # do setxattr a few times
    # 1 attr
    exitcode, out_1attr = testlib.run( SETXATTR_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                      '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,  
                                      path, attr_name_base + "-1attr", attr_value_base + "-1attr", valgrind=True )

    testlib.save_output( output_dir, "syndicate-setxattr-1attr", out_1attr )
    if exitcode != 0:
        raise Exception("%s exited %s" % (SETXATTR_PATH, exitcode))

    # 5 attrs at once

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-setxattr-removexattr.py

示例3: xrange

    for (start, end) in ranges:

        # only clear reader's cache
        testlib.clear_cache( config_dir, volume_id=volume_id, gateway_id=read_gateway_id )

        # do each read twice--once uncached, and one cached 
        for i in xrange(0, 2):
            exitcode, out = testlib.run( READ_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                        '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', read_gateway_name,
                                        output_path, start, end - start, valgrind=True )

            out_name = "uncached"
            if i > 0:
                out_name = "cached"

            testlib.save_output( output_dir, 'syndicate-read-%s-%s-%s' % (start, end, out_name), out )
            if exitcode != 0:
                stop_and_save( output_dir, ag_proc, ag_out_path, "syndicate-ag")
                raise Exception("%s exited %s" % (READ_PATH, exitcode))

            # correctness 
            if expected_data[start:end] not in out:
                stop_and_save( output_dir, ag_proc, ag_out_path, "syndicate-ag")
                raise Exception("Missing data for %s-%s" % (start, end))

    ag_exitcode, ag_out = testlib.stop_gateway( ag_proc, ag_out_path )

    testlib.save_output( output_dir, "syndicate-ag", ag_out )

    if ag_exitcode != 0:
        raise Exception("%s exited %s" % (AG_PATH, ag_exitcode))

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-ag-read.py

示例4: open

        local_fd = open(local_path, "r")
        expected_data = local_fd.read()
        local_fd.close()

        local_paths.append(local_path)
        expected_datas.append(expected_data)

    # start AG
    AG_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "AG", caps="ALL", email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, AG_gateway_name, "port=31112", "driver=%s" % AG_DRIVER )
    ag_proc, ag_out_path = testlib.start_gateway( config_dir, AG_PATH, testconf.SYNDICATE_ADMIN, volume_name, AG_gateway_name, valgrind=True )
    time.sleep(20)

    if ag_proc.poll() is not None:
        ag_exitcode, ag_out = testlib.stop_gateway( ag_proc, ag_out_path )
        testlib.save_output(output_dir, "syndicate-ag", ag_out)
        raise Exception("%s exited %s" % (AG_PATH, ag_proc.poll()))

    # should cause the AG to get updated that there's a new gateway 
    cat_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    expected_data_error = False
    for i in xrange(0,len(local_paths)):
        # do each file twice 
        for j in xrange(0, 2):
            local_path = local_paths[i]
            expected_data = expected_datas[i]

            output_path = "/" + os.path.basename(local_path)
            exitcode, out = testlib.run( CAT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', cat_gateway_name, output_path, valgrind=True )

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:30，代码来源:syndicate-ag-cat.py

示例5: xrange

    # put a file to publish
    local_paths = []
    for i in xrange(0, 1):
        local_path = testlib.make_random_file(16384, dir=inner_dir)
        local_paths.append(local_path)

    # start AG
    AG_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "AG", caps="ALL", email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, AG_gateway_name, "port=31112", "driver=%s" % AG_DRIVER )
    ag_proc, ag_out_path = testlib.start_gateway( config_dir, AG_PATH, testconf.SYNDICATE_ADMIN, volume_name, AG_gateway_name, valgrind=True )
    time.sleep(20)

    if ag_proc.poll() is not None:
        ag_exitcode, ag_out = testlib.stop_gateway( ag_proc, ag_out_path )
        testlib.save_output(output_dir, "syndicate-ag", ag_out)
        raise Exception("%s exited %s" % (AG_PATH, ag_proc.poll()))

    # should cause the AG to get updated that there's a new gateway 
    ug_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    # remove inner dir, and have the AG refresh
    shutil.rmtree(rmdir)
    exitcode, out = testlib.run( REFRESH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', ug_gateway_name, '/to_remove_1', valgrind=True )
    testlib.save_output( output_dir, "syndicate-refresh", out )

    if exitcode != 0:
        ag_exitcode, ag_out = testlib.stop_gateway( ag_proc, ag_out_path )
        testlib.save_output(output_dir, "syndicate-ag", ag_out)
        raise Exception("syndicate-refresh exit code %s" % exitcode)

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:29，代码来源:syndicate-ag-rmtree-cat.py

示例6: Exception

    if not testlib.gateway_ping( 31112, 15 ):
        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # should cause the RG to get updated that there's a new gateway 
    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )
    cat_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email="ANONYMOUS" )

    random_part = hex(random.randint(0, 2**32-1))[2:]
    output_paths = []

    for i in xrange(0, NUM_FILES):
        output_path = "/put-%s-%s" % (random_part, i)
        output_paths.append(output_path)

        exitcode, out = testlib.run( PUT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, local_path, output_path, valgrind=True )
        testlib.save_output( output_dir, "syndicate-put-%s" % i, out )

        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    for i in xrange(0, NUM_FILES):
        path = output_paths[i]
        exitcode, out = testlib.run( CAT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', 'ANONYMOUS', '-v', volume_name, '-g', cat_gateway_name, path, valgrind=True )
        testlib.save_output( output_dir, 'syndicate-cat-%s' % i, out )
        
        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (CAT_PATH, exitcode)) 

        # check for correctnes

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-put-cat-anonymous.py

示例7: xrange

    for (start, end) in ranges:

        # only clear reader's cache
        testlib.clear_cache( config_dir, volume_id=volume_id, gateway_id=read_gateway_id )

        # do each read twice--once uncached, and one cached 
        for i in xrange(0, 2):
            exitcode, out = testlib.run( READ_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                        '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', read_gateway_name,
                                        output_path, start, end - start, valgrind=True )

            out_name = "uncached"
            if i > 0:
                out_name = "cached"

            testlib.save_output( output_dir, 'syndicate-read-%s-%s-%s' % (start, end, out_name), out )
            if exitcode != 0:
                stop_and_save( output_dir, ag_proc, ag_out_path, "syndicate-ag")
                raise Exception("%s exited %s" % (READ_PATH, exitcode))

            # correctness 
            if expected_data[start:end] not in out:
                stop_and_save( output_dir, ag_proc, ag_out_path, "syndicate-ag")
                raise Exception("Missing data for %s-%s" % (start, end))

    # finally, list it
    for p in ['/', '/dir1', '/dir1/dir2', output_path]:
        exitcode, out = testlib.run( LS_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                     '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', read_gateway_name,
                                     p, valgrind=True )

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:30，代码来源:syndicate-ag-read-tree.py

示例8: stop_and_save

    rg_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "RG", caps="NONE", email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, rg_gateway_name, "port=31112", "driver=%s" % RG_DRIVER )

    # start the RG 
    rg_proc, rg_out_path = testlib.start_gateway( config_dir, RG_PATH, testconf.SYNDICATE_ADMIN, volume_name, rg_gateway_name )
    if not testlib.gateway_ping( 31112, 15 ):
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # put the file...
    exitcode, out = testlib.run( PUT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,
                                random_file_path, path )

    testlib.save_output( output_dir, "syndicate-put", out )

    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("Failed to touch %s" % path)

    # finish populating expected built-in xattrs 
    exitcode, out = testlib.run( STAT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, "syndicate.conf"),
                                "-u", testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', getxattr_gateway_name,
                                path )

    testlib.save_output( output_dir, "syndicate-stat", out )
    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("Failed to stat %s" % path)

开发者ID:iychoi，项目名称:syndicate-core，代码行数:29，代码来源:syndicate-getxattr-builtin.py

示例9: hex

    attr_name_base = 'foo-%s' % random_part
    attr_value_base = 'bar-%s' % random_part
    expected_listxattr_output = ""

    config_dir, output_dir = testlib.test_setup()
    volume_name = testlib.add_test_volume( config_dir )

    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    random_part = hex(random.randint(0, 2**32-1))[2:]
   
    exitcode, out = testlib.run( TOUCH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,
                                path )

    testlib.save_output( output_dir, "syndicate-touch", out )

    if exitcode != 0:
        raise Exception("Failed to touch %s" % path)

    # do setxattr a few times
    # 1 attr
    exitcode, out_1attr = testlib.run( SETXATTR_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                      '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,  
                                      path, attr_name_base + "-1attr", attr_value_base + "-1attr", valgrind=True )

    testlib.save_output( output_dir, "syndicate-setxattr-1attr", out_1attr )
    expected_listxattr_output += attr_name_base + "-1attr" + '\n'

    if exitcode != 0:
        raise Exception("%s exited %s" % (SETXATTR_PATH, exitcode))

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-setxattr-listxattr.py

示例10: Exception

    RG_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "RG", caps="NONE", email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, RG_gateway_name, "port=31112", "driver=%s" % RG_DRIVER )

    rg_proc, rg_out_path = testlib.start_gateway( config_dir, RG_PATH, testconf.SYNDICATE_ADMIN, volume_name, RG_gateway_name )
    time.sleep(1)
    if rg_proc.poll() is not None:
        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # should cause the RG to get updated that there's a new gateway 
    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    random_part = hex(random.randint(0, 2**32-1))[2:]
    output_path = "/put-%s" % random_part
    exitcode, out = testlib.run( PUT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, local_path, output_path )

    testlib.save_output( output_dir, "syndicate-put", out )

    if exitcode != 0:
        raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    # read the file, to populate the cache 
    exitcode, out = testlib.run( CAT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                 '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,
                                 output_path, valgrind=True )

    if exitcode != 0:
        rg_exitcode, rg_out = testlib.stop_gateway( rg_proc, rg_out_path )
        testlib.save_output( output_dir, "syndicate-rg", rg_out )
        raise Exception("Failed to read %s" % path)

    # check cache

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:31，代码来源:syndicate-put-getxattr-builtin.py

示例11: Exception

    expected_data = local_fd.read()
    local_fd.close()

    # start up RG
    RG_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "RG", port=32110, email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, RG_gateway_name, "driver=%s" % RG_DRIVER )

    rg_proc, rg_out_path = testlib.start_gateway( config_dir, RG_PATH, testconf.SYNDICATE_ADMIN, volume_name, RG_gateway_name, valgrind=True )
    time.sleep(5)
    if rg_proc.poll() is not None:
        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # put the file
    put_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", port=31110, email=testconf.SYNDICATE_ADMIN )
    exitcode, out = testlib.run( PUT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', put_gateway_name, local_path, output_path, valgrind=True )
    testlib.save_output( output_dir, "syndicate-put", out )

    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("syndicate-put exit code %s" % exitcode)

    # create separate threads for UGs 
    for i in xrange(0, num_threads):
        t = UGThread( config_dir, output_dir, output_path, volume_name, caps="ALL", port=(31111 + i + 1), email=testconf.SYNDICATE_ADMIN )
        threads.append(t)

    # run all threads 
    for i in xrange(0, num_threads):
        threads[i].start()

    # wait for them to finish

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:31，代码来源:syndicate-put-read-parallel.py

示例12: Exception

    if not testlib.gateway_ping( 31112, 15 ):
        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # should cause the RG to get updated that there's a new gateway 
    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    # touch $NUM_FILES files 
    random_part = hex(random.randint(0, 2**32-1))[2:]
    expected_paths = []

    for i in xrange(0, NUM_FILES):
        output_path = "/touch-%s-%s" % (random_part, i)
        expected_paths.append(output_path)

        exitcode, out = testlib.run( TOUCH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, output_path )
        testlib.save_output( output_dir, "syndicate-touch-%s" % i, out )
        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (TOUCH_PATH, exitcode))

    # list them 
    exitcode, out = testlib.run( LS_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, '/' )
    testlib.save_output( output_dir, "syndicate-ls", out )
    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("%s exited %s" % (LS_PATH, exitcode))

    # verify that they're all there 
    for path in expected_paths:
        if path.strip("/") not in out:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-create-delete.py

示例13: hex

    # look up RG 
    rg_gateway_info = testlib.read_gateway( config_dir, RG_gateway_name )
    rg_gateway_id = rg_gateway_info['g_id']

    volume_info = testlib.read_volume( config_dir, volume_name )
    volume_id = volume_info['volume_id']

    random_part = hex(random.randint(0, 2**32-1))[2:]
    output_paths = []

    for i in xrange(0, NUM_FILES):
        output_path = "/put-%s-%s" % (random_part, i)
        output_paths.append(output_path)

        exitcode, out = testlib.run( PUT_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, local_path, output_path, valgrind=True )
        testlib.save_output( output_dir, "syndicate-put-%s" % i, out )

        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    # make target directory
    exitcode, out = testlib.run( MKDIR_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, "/newdir", valgrind=True )
    testlib.save_output( output_dir, "syndicate-mkdir", out )

    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    # do several renames
    for j in xrange(0, NUM_RENAMES):

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-put-rename-cat.py

示例14: hex

    attr_value_base = 'bar-%s' % random_part

    config_dir, output_dir = testlib.test_setup()
    volume_name = testlib.add_test_volume( config_dir )

    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )
    gateway_client_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL",  email=testconf.SYNDICATE_ADMIN )
    testlib.update_gateway( config_dir, gateway_client_name, "port=31112" )

    random_part = hex(random.randint(0, 2**32-1))[2:]
   
    exitcode, out = testlib.run( TOUCH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,
                                path )

    testlib.save_output( output_dir, "syndicate-touch", out )

    if exitcode != 0:
        raise Exception("Failed to touch %s" % path)

    # do setxattr a few times
    # 1 attr
    exitcode, out_1attr = testlib.run( SETXATTR_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                      '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name,  
                                      path, attr_name_base + "-1attr", attr_value_base + "-1attr", valgrind=True )

    testlib.save_output( output_dir, "syndicate-setxattr-1attr", out_1attr )
    if exitcode != 0:
        raise Exception("%s exited %s" % (SETXATTR_PATH, exitcode))

    # 5 attrs at once

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-setxattr-getxattr-remote.py

示例15: Exception

        raise Exception("%s exited %s" % (RG_PATH, rg_proc.poll()))

    # should cause the RG to get updated that there's a new gateway 
    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    for rep in xrange(0, NUM_REPS):
        # touch $NUM_FILES files 
        random_part = hex(random.randint(0, 2**32-1))[2:]
        expected_paths = []

        for i in xrange(0, NUM_FILES):
            output_path = "/touch-%s-%s" % (random_part, i)
            expected_paths.append(output_path)

        exitcode, out = testlib.run( TOUCH_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, *expected_paths )
        testlib.save_output( output_dir, "syndicate-touch-%s" % rep, out )
        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (TOUCH_PATH, exitcode))

        # list them 
        exitcode, out = testlib.run( LS_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, '/' )
        testlib.save_output( output_dir, "syndicate-ls-full-%s" % rep, out )
        if exitcode != 0:
            stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
            raise Exception("%s exited %s" % (LS_PATH, exitcode))

        # verify that they're all there 
        for path in expected_paths:
            if path.strip("/") not in out:
                stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:31，代码来源:syndicate-create-delete-multi.py

示例16: xrange

        if abort:
            for i in xrange(0, num_threads):
                threads[i].set_errors(abort)

            for i in xrange(0, num_threads):
                threads[i].join()

        dead = True
        for i in xrange(0, num_threads):
            if threads[i].isAlive():
                dead = False

    if abort:
        stop_and_save( output_dir, ag_proc, ag_out_path, "syndicate-ag")
        msg = []
        for i in xrange(0, num_threads):
            if threads[i].get_exitcode() != 0:
                msg.append("Thread %s (%s) exited %s, error = '%s'" % (threads[i].get_ident(), i, threads[i].get_exitcode(), threads[i].get_errormsg()))

        raise Exception("\n" + "\n".join(msg))

    ag_exitcode, ag_out = testlib.stop_gateway( ag_proc, ag_out_path )

    testlib.save_output( output_dir, "syndicate-ag", ag_out )

    if ag_exitcode != 0:
        raise Exception("%s exited %s" % (AG_PATH, ag_exitcode))
  
    sys.exit(0)

开发者ID:iychoi，项目名称:syndicate-core，代码行数:29，代码来源:syndicate-ag-read-parallel.py

示例17: xrange

    os.makedirs(testdir)
    
    local_path = testlib.make_random_file(16384, dir=testdir)
    output_path = "/dir1/dir2/%s" % os.path.basename(local_path)

    # list files that do not yet exist
    # verify that we can't list them
    descendent_files = ['/dir1', '/dir1/dir2', output_path]
    for i in xrange(0, len(descendent_files)):
        p = descendent_files[i]
        exitcode, out = testlib.run( LS_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'),
                                     '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', read_gateway_name,
                                     p, valgrind=True )

        testlib.save_output(output_dir, "syndicate-ls-before-%s" % p.replace("/", "\\x2f" ), out )
        if ("\nFailed to stat '%s': No such file or directory\n" % p) not in out:
            raise Exception("Successfully listed '%s'" % p)

        p_base = p
        if p != '/':
            p_base = os.path.basename(p)

        if ("\nname:     %s\n" % p_base) in out:
            raise Exception("Found '%s' in listing" % p)

        if i + 1 < len(descendent_files):
            if ("name:     %s" % os.path.basename(descendent_files[i+1])) in out:
                raise Exception("Found '%s' in listing" % descendent_files[i+1])

    # start up AG

开发者ID:iychoi，项目名称:syndicate-deploy，代码行数:30，代码来源:syndicate-ag-ls-tree-timeout.py

示例18: hex

    # should cause the RG to get updated that there's a new gateway 
    gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )
    read_gateway_name = testlib.add_test_gateway( config_dir, volume_name, "UG", caps="ALL", email=testconf.SYNDICATE_ADMIN )

    read_gateway_info = testlib.read_gateway( config_dir, read_gateway_name )
    read_gateway_id = read_gateway_info['g_id']

    volume_info = testlib.read_volume( config_dir, volume_name )
    volume_id = volume_info['volume_id']

    random_part = hex(random.randint(0, 2**32-1))[2:]
    output_path = "/put-%s" % random_part
    exitcode, out = testlib.run( PUT_PATH, '-d3', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, local_path, output_path )

    testlib.save_output( output_dir, "syndicate-put", out )

    if exitcode != 0:
        raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    # try reading and writing various ranges (these are (start, end) absolute ranges, not offset/length)
    ranges = [
        (5000, 16000),
        (0, 1),     # 1 block, tail unaligned
        (1, 200),   # 1 block, unaligned head and tail
        (0, 4096),  # 4 block, aligned
        (0, 8192),  # 8 blocks, aligned
        (0, 1000),  # 1 block, tail unaligned
        (0, 6000),  
        (100, 4000),
        (5000, 10000),

开发者ID:iychoi，项目名称:syndicate-core，代码行数:30，代码来源:syndicate-create-write-read.py

示例19: hex

    cat_gateway_info = testlib.read_gateway( config_dir, cat_gateway_name )
    cat_gateway_id = cat_gateway_info['g_id']

    # look up RG 
    rg_gateway_info = testlib.read_gateway( config_dir, RG_gateway_name )
    rg_gateway_id = rg_gateway_info['g_id']

    volume_info = testlib.read_volume( config_dir, volume_name )
    volume_id = volume_info['volume_id']

    random_part = hex(random.randint(0, 2**32-1))[2:]
    output_paths = []

    # make target directory
    exitcode, out = testlib.run( MKDIR_PATH, '-d2', '-f', '-c', os.path.join(config_dir, 'syndicate.conf'), '-u', testconf.SYNDICATE_ADMIN, '-v', volume_name, '-g', gateway_name, "/newdir", valgrind=True )
    testlib.save_output( output_dir, "syndicate-mkdir", out )

    if exitcode != 0:
        stop_and_save( output_dir, rg_proc, rg_out_path, "syndicate-rg")
        raise Exception("%s exited %s" % (PUT_PATH, exitcode))

    # generate the commands to feed into the REPL
    repl_cmd = ""
    for i in xrange(0, NUM_FILES):
        output_path = "/put-%s-%s" % (random_part, i)
        output_paths.append(output_path)

        repl_cmd += "create %s 0644\n" % output_path
        repl_cmd += "write 0 0 %s %s\n" % (len(write_data), write_data)

        # truncate while open

开发者ID:iychoi，项目名称:syndicate-core，代码行数:31，代码来源:syndicate-write-trunc-samehandle-read-repl.py

注：本文中的testlib.save_output函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

鲜花

握手

雷人

路过

鸡蛋

该文章已有0人参与评论

请发表评论

全部评论

专题导读

More+

10-27 六六分期app的软件客服如何联系？(六六分期

11-06 可心卡盟:win10系统火狐flash插件崩溃怎么

11-06 亲亲特价:怎么删除回收站图标

11-06 济南大学虚拟社区:鲁大师节能降温的具体办

11-06 xlueops.exe:无线网络安装向导

11-06 女斗合众国:win7系统cf与主机连接不稳定怎

11-06 0xc000022-[cf烟雾头]cf怎么调烟雾头

11-06 qizideyouhuo:应用程序无法正常启动0xc0000

11-06 ipz-185:win7系统vcf文件怎么打开

11-06 傻哥蹦迪:win10系统s4怎么打开usb调试

11-06 八神浩树gtaste:回收站清空了怎么恢复

11-06 妖尾之黑色守护:win10系统电脑没有1440x900

11-06 校园至尊魔王小说:win7系统浏览网页时字体

11-06 女斗合众国:win10系统访问共享文件夹提示请

11-06 tokyo hot n0654:恢复win7系统默认字体一招

11-06 雨酷仙境:设置win7系统转移临时文件夹腾出

11-06 阿穆纳伊之杖:win7系统开始菜单在右边还原

11-06 tunespotting:win10系统火狐flash插件总是

11-06 甘尔葛分析师：计谋网站seo关键词暴涨有什

11-06 蔡贵霖: 计谋网站seo关键词暴涨有什么秘密

11-06 博益网首页:ao3网页版进入不了解决方法

11-06 漏斗子专栏: 网站数据分析小白易懂精华篇

11-06 见证双虹怎么做:win7系统开启telnet命令的

11-06 颾狐蝶蜋:系统资源不足无法完成请求的服务

11-06 国光中学校歌:提交网站到alexa查询详细步骤

11-06 西安有情天:静态网页和动态网页的区别

11-06 红木雅尚斋:外部链接构造对网站的好处

11-06 前官礼遇：防止域名劫持–增强域安全性的10

11-06 密传二转答案: 中文分词算法有哪些

11-06 金泉家园邮编:百度快照劫持的表现及应对方

Python testlib.temp_dir函数代码示例发布时间：2022-05-27

Python testlib.random_string函数代码示例发布时间：2022-05-27

Python util.grid_equal函数代码示例

1 Python 入门教程

Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P

阅读：13772|2022-01-22

2 Python wikiutil.getFrontPage函数代码示例

Python wikiutil.getFrontPage函数代码示例

阅读：9577|2022-05-24

3 Python 简介

Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本

阅读：4066|2022-01-22

4 Python tests.group函数代码示例

Python tests.group函数代码示例

阅读：4039|2022-05-27

5 Python util.check_if_user_has_permission

Python util.check_if_user_has_permission函数代码示例

阅读：3819|2022-05-27

6 Python 操练实例98

Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大

阅读：3499|2022-01-22

7 Python 环境搭建

Python 环境搭建本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py

阅读：3022|2022-01-22

8 Python output.darkgreen函数代码示例

Python output.darkgreen函数代码示例

阅读：2638|2022-05-25

9 Python 基础语法

Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也

阅读：2624|2022-01-22

10 Python 中文编码

Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没

阅读：2290|2022-01-22

客服电话

电子邮件

Python testlib.save_output函数代码示例

示例1: run

示例2: hex

示例3: xrange

示例4: open

示例5: xrange

示例6: Exception

示例7: xrange

示例8: stop_and_save

示例9: hex

示例10: Exception

示例11: Exception

示例12: Exception

示例13: hex

示例14: hex

示例15: Exception

示例16: xrange

示例17: xrange

示例18: hex

示例19: hex

请发表评论

全部评论

上一篇：

下一篇：

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.grid_equal函数代码示例

Python util.get_worker_name函数代码示例

Python util.get_webmention_target函数代

Python util.get_uuid函数代码示例

Python util.get_type_by_name函数代码示例

Python util.get_stdout函数代码示例

关于我们

产品与服务

解决方案

139-2527-9053