ngamsArchiveStressTest.py 文件源码

def archiveThread(testObj,
                  no,
                  inc,
                  dummy):
    """
    Archive a file X times.

    testObj:  Reference to instance of ngamsTestSuite object (ngamsTestSuite).

    no:       Number allocated to thread (integer).

    inc:      Increment ARCFILE keyword before submitting each Archive Request
              (0|1/integer).

    Returns:  Void
    """
    if (inc):
        filename = "tmp/ngamsArchiveStressTest_%d.fits" % no
        cpFile("src/TinyTestFile.fits", filename)
        incArcfile(filename, step=(100 * no))
    else:
        filename = "src/TinyTestFile.fits"
    testStr = TST_STR1 % os.path.basename(filename)

    for n in range(5):
        if (not RUN_TEST):
            THREAD_STAT[no] = "STOPPED"
            break
        if (inc): incArcfile(filename)
        statObj = sendPclCmd(auth=AUTH).archive(filename)
        if (statObj.getMessage() != testStr):
            THREAD_STAT[no] = "FAILURE: Archive Request failed"
    THREAD_STAT[no] = "SUCCESS"
    thread.exit()