testFileTools.py

Go to the documentation of this file.

#!/usr/local/bin/python
#
#                               Copyright 2005
#                                     by
#                        The Board of Trustees of the
#                     Leland Stanford Junior University.
#                            All rights reserved.
#

__facility__ = "Online"
__abstract__ = "File Upload test - VSC version"
__author__   = "S. Tuvi <Stuvi@SLAC.Stanford.edu> SLAC - GLAST LAT I&T/Online"
__date__     = "2005/07/15 00:08:27"
__updated__  = "$Date: 2006/03/08 00:40:22 $"
__version__  = "$Revision: 1.6 $"
__release__  = "$Name: HEAD $"
__credits__  = "SLAC"

import LICOS.copyright_SLAC

import logging as log
import            os
import            sys

from LICOS.tests.lib.testAppBase  import TestAppBase
#import LICOS.tests.lib.fileTools  as fileTools
import LICOS.lib.cmdTlmDb.FILE    as FILE
from LICOS.lib.LATconstants       import *

#  Example application implementation.
class testFileTools(TestAppBase):
  "Test file and memory tools"
  def __init__(self, common):
    TestAppBase.__init__(self, common)
    log.debug("%s.__init__()" % self.getName())

  def getName(self):
    return self.__class__.__name__

  def testStartRun(self):
    self.setCompletionStatus(self.COMPL_STATUS_PASSED)

    self.__diskFile = os.path.join(ONLINE_ROOT, 'LICOS/tests/data/12346_hdr')
    self.__noHeader = os.path.join(ONLINE_ROOT, 'LICOS/tests/data/12346')
    self.__tempFile = os.path.expandvars('/tmp/licos_download_delete')

    self.__testDirectory = 37
    self.__testDirectory = 2
    self.__testDevice    = FSW_DEVICE_RAM
    self.__testDevice    = FSW_DEVICE_EE1

    # compute a semi-arbitrary file ID.  Directory/fileNum up to user.
    self.__testID   = FILE.FswFileID( device     = self.__testDevice,
                                      directory  = self.__testDirectory,
                                      fileNumber = 12346)

    self.__testID2  = FILE.FswFileID( device     = self.__testDevice,
                                      directory  = self.__testDirectory,
                                      fileNumber = 33333)

    # directory file IDs have filenumper=0
    self.__testDir  = FILE.FswFileID( device     = self.__testDevice,
                                      directory  = self.__testDirectory,
                                      fileNumber = 0)

    # root ids have directory=127
    self.__rootID   = FILE.FswFileID( device     = self.__testDevice,
                                      directory  = FSW_DIR_ROOT,
                                      fileNumber = 0)

    # build compressed version of self.__diskFile in memory
    tmp = FILE.FswFileFromDisk(self.__testID2, self.__noHeader)
    self.__zipFile  = FILE.addFmxHeader(tmp, doCompress=True)

    self.lcat.setCmdConfirmCallback(self.lcat.quietConfirmation)

    pass

  def testRun(self):

    status  = 0

#    log.info("%s: Testing creation of a directory on the RAM disk" % \
#              ( self.getName() ) )
#    status |= self.testCreateDirectory()
#
    log.info("%s: Testing dump of RAM directory" % \
              ( self.getName() ) )
    status |= self.testDirectory()

#    log.info("%s: Testing upload of disk file %s to RAM" % \
#              ( self.getName(), self.__diskFile ) )
#    status |= self.testFileUpload()
#
#    log.info("%s: Testing download of file" % \
#              ( self.getName() ) )
#    status |= self.testFileDownload()
#
#    log.info("%s: Testing copy of file" % \
#              ( self.getName() ) )
#    status |= self.testFileCopy()
#
#    log.info("%s: Testing dump of RAM directory" % \
#              ( self.getName() ) )
#    status |= self.testDirectory()

    #log.info("%s: Testing deletion of RAM directory" % \
    #          ( self.getName() ) )
    #status |= self.testDeleteDirectory()

#    log.info("%s: Testing dump of RAM directory" % \
#              ( self.getName() ) )
#    status |= self.testDirectory()


    self.setCompletionStatus(self.COMPL_STATUS_PASSED)
    # End of run

  def testStopRun(self):
    pass


  def testFileUpload(self):
    """! Test the bare file upload from disk.

    \return Status
    """
    try:

      uplFile = self.lcat.FILE.uploadFileFromDisk(fileName = self.__diskFile,
                                                  unit     = FSW_UNIT_SIU,
                                                  fileID   = self.__testID)

      if uplFile is None:
        return self.COMPL_STATUS_FAILED

      uplFile = self.lcat.FILE.uploadFileFromData(fileData = self.__zipFile.binary(),
                                                  unit     = FSW_UNIT_SIU,
                                                  fileID   = self.__zipFile.fileID())

      if uplFile is None:
        return self.COMPL_STATUS_FAILED

      return self.COMPL_STATUS_PASSED
    except:
      log.exception("Exception in testFileUpload")
      return self.COMPL_STATUS_FAILED

  def testFileDownload(self):
    """!\brief test the bare file download system

    \return status
    """
    try:
      fswFile = self.lcat.FILE.downloadFile(unit     = FSW_UNIT_SIU,
                                            fileID   = self.__testID)

      f = open(self.__tempFile, 'wb')
      fswFile.write(f)
      f.close()

      return self.COMPL_STATUS_PASSED
    except:
      log.exception("Exception in testFileDownload")
      return self.COMPL_STATUS_FAILED

  def testDirectory(self):
    """!\brief test the directory structure

    \return status
    """
    try:

      # Two types of directory lists.
      # 1: list of the files in a directory
      directory = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                                directoryID = self.__testID)

      directory.dump(sys.stdout)

      # 2: list of the directories on a unit (on a device)
      directory2 = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                                 directoryID = self.__rootID,
                                                 root = True)

      directory2.dump(sys.stdout)

      return self.COMPL_STATUS_PASSED
    except:
      log.exception("Exception in testDirectory")
      return self.COMPL_STATUS_FAILED

  def testDeleteDirectory(self):
    """!\brief test deleting a directory structure

    \return status
    """

    try:
      d0 = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                         directoryID = self.__rootID,
                                         root = True)

      # deleting a directory deletes the files in it if force is set.
      status = self.lcat.FILE.deleteDirectory( unit = FSW_UNIT_SIU,
                                               directoryID = self.__testDir,
                                               force = False)

      status = self.lcat.FILE.deleteDirectory( unit = FSW_UNIT_SIU,
                                               directoryID = self.__testDir,
                                               force = True)

      d1 = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                         directoryID = self.__rootID,
                                         root = True)

      if len(d0.files()) <= len(d1.files()):
        log.error("directory deletion didn't work.  Check any previous warnings")
        return self.COMPL_STATUS_FAILED

      return self.COMPL_STATUS_PASSED
    except:
      log.exception("Exception in testDeleteDirectory")
      return self.COMPL_STATUS_FAILED

  def testCreateDirectory(self):
    """!\brief test deleting a directory structure

    \return status
    """
    try:
      oldRoot = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                              directoryID = self.__rootID,
                                              root = True)
      oldRoot.dump(sys.stdout)

      status = self.lcat.FILE.createDirectory(unit =FSW_UNIT_SIU,
                                              directoryID = self.__testDir)

      newRoot = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                              directoryID = self.__rootID,
                                              root = True)
      newRoot.dump(sys.stdout)

      if len(oldRoot.files()) < len(newRoot.files()):
        return self.COMPL_STATUS_PASSED
      else:
        log.error( "old root directory had %d directories, new has %d" % \
                   (len(oldRoot.files()), len(newRoot.files())) )

        return self.COMPL_STATUS_FAILED

    except:
      log.exception("Exception in testDeleteDirectory")
      return self.COMPL_STATUS_FAILED

  def testFileCopy(self):
    """!\brief test copying a file

    \return status
    """
    try:
      newID = FILE.FswFileID( device     = self.__testDevice,
                              directory  = self.__testDirectory,
                              fileNumber = 654321)

      d0 = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                         directoryID = self.__testID)

      status = self.lcat.FILE.copyFile( unit = FSW_UNIT_SIU,
                                        source = self.__testID,
                                        dest = newID)

      d1 = self.lcat.FILE.dumpDirectory( unit = FSW_UNIT_SIU,
                                         directoryID = newID)


      print "old directory had %d files, new has %d" % \
            (len(d0.files()), len(d1.files()))


      return self.COMPL_STATUS_PASSED
    except:
      log.exception("Exception in testDeleteDirectory")
      return self.COMPL_STATUS_FAILED

---