Attachment 318031 Details for Bug 462349 – opticalDiskTest.py common functions for cd and dvd testing

opticalDiskTest.py common functions for cd and dvd testing

opticalDiskTest.py (text/x-python), 20.92 KB, created by Greg Nichols on 2008-09-30 01:32:09 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Greg Nichols

Created: 2008-09-30 01:32:09 UTC

Size: 20.92 KB

patch

obsolete

>#!/usr/bin/python
># Copyright (c) 2006 Red Hat, Inc. All rights reserved. This copyrighted material 
># is made available to anyone wishing to use, modify, copy, or 
># redistribute it subject to the terms and conditions of the GNU General 
># Public License v.2.
># 
># This program is distributed in the hope that it will be useful, but WITHOUT ANY
># WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
># PARTICULAR PURPOSE. See the GNU General Public License for more details.
>#
># You should have received a copy of the GNU General Public License
># along with this program; if not, write to the Free Software
># Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
>#
># Author: Greg Nichols
>#         YangKun <ykun@redhat.com>
>#
># This class is a base class for optical disk (CD, DVD, BluRay) testing.
>
>import os
>import sys
>import commands
>import string
>import shutil
>import curses
>import time
>import filecmp
>import re
>
>from hts.test import Test
>from hts.tags import Constants
>from hts.hal import Hal
>from hts.device import Device
>from hts.command import Command, HTSCommandException
>
>class OpticalDiskTest(Test):        
>    
>    def __init__(self, name):
>        Test.__init__(self, name)
>        self.interactive = True
>        self.mediaTypes = None # subclasses must set this list
>        self.supportedMedia = None
>
>        
>    def hasAnyCDROMType(self, device):
>        for type in self.mediaTypes:
>            if device.getProperty("storage.cdrom.%s" % type):
>                return True
>        return False
>    
>    def getDeviceName(self, device):
>        disk = device.getProperty("block.device")
>        """ chop off /dev/ (5 characters) """
>        if disk and len(disk) > 5:
>            disk = disk[5:]
>            return disk
>        return None 
>        
>    def plan(self, devices):
>        tests = list()
>        for device in devices:
>            disk = None
>            if 'storage.cdrom' in device.getProperty("info.capabilities") and self.hasAnyCDROMType(device):
>                disk = self.getDeviceName(device)
>            elif 'CDROM' in device.getProperty("class"):
>                disk = device.getProperty("device")
>             
>            if disk:
>                test = self.makeCopy()
>                test.setDevice(device)
>                test.setLogicalDeviceName(disk)
>                tests.append(test)
>                
>        return tests
>
>
>    def handleEject(self,cdromDevice):
>        """testing the eject and reload functions. also some devices (e.g. Philips) need to eject the medium before creating a new disk."""
>        deviceFile = "/dev/" + cdromDevice
>        print "\n  Reloading the media ... "
>        sys.stdout.flush()
>        try:
>            Command("eject %s" % deviceFile).run()
>            print "\n  tray ejected."
>            sys.stdout.flush()
>        except HTSCommandException, exception:
>            # or the tray is already auto-ejected, so continue
>            pass
>
>        try:
>            Command("eject -t %s" % deviceFile).run()
>            print "\n  tray auto-closed."
>            sys.stdout.flush()
>        except HTSCommandException, exception:
>            print "Could not close the tray"
>            # for those slim-line cdrom drives, auto-close can not work, have to close the tray manually
>            self.prompt("\n Please close the drive tray of %s manually then press ENTER to continue." % deviceFile)
>        
>        return True
>
>    def cmpTree(self,dir1,dir2):
>        """compare regular files in dir1 and dir2; returns true if all match."""
>        for dirpath, dirnames, filenames in os.walk(dir1):
>            for name in filenames:
>                file1 = os.path.join(dirpath,name)
>                file2 = file1.replace(dir1,dir2,1)
>                if os.path.isfile(file1) and not os.path.islink(file1):
>                    if filecmp.cmp(file1,file2):
>                        continue
>                    else:
>                        print "\n  +++ Error: File comparison failed on file : %s" % file1
>                        return False
>                else:
>                    continue
>        return True
>
>    def getWriteMode(self):
>        """return -tao if the drive supports TAO recording, or
>           return -sao if supports SAO, use default otherwise"""
>        deviceFile = "/dev/" + self.getLogicalDeviceName()
>        writeMode=""
>        try:
>            command = Command("cdrecord dev=/dev/%s -checkdrive" % self.getLogicalDeviceName())
>            modes = command.getString(regex="^Supported modes[^:]*:(?P<modes>.*$)", regexGroup="modes", singleLine=False)
>            if "TAO" in modes:
>                writeMode="-tao"
>            if "SAO" in modes:
>                writeMode="-sao"
>                
>            flags = command.getString(regex="^Driver flags[^:]*:(?P<flags>.*$)", regexGroup="flags", singleLine=False)
>            if "BURNFREE" in flags:
>                writeMode += " driveropts=burnfree"
>                
>        except HTSCommandException, exception:
>            print exception
>
>            return None
>        
>        return writeMode
>
>        
>
>    def getSupportedMedia(self):
>        """configure the cdrom based on HAL info, return all supported features.
>           we check cd-rom/cd-r/cd-rw/dvd-rom/dvd-r/dvd-rw/dvd+r/dvd+rw"""
>        # if UDI is available, great ! use it !
>        # this should be the normal case in RHEL5
>        if not self.udi:
>            # on RHEL4, we rely on Kudzu to discover HW,
>            # and HAL somehow works :)
>            # but we need to find the correct UDI first.
>            try:
>                lshalCommand = Command("lshal")
>                udiList = lshalCommand.getStringList(regex="udi = \'(?<udi>[^\']+)\'", regexGroup="udi")
>            except HTSCommandException, exception:
>                print exception
>                
>            for udi in udiList:
>                deviceName = commands.getoutput("hal-get-property --udi %s --key block.device |  colrm 1 5" % udiTmp)
>                if deviceName == self.getLogicalDeviceName():
>                    self.udi = udi
>                    break
>                else:
>                    continue
>        # check the features now
>        if self.udi:
>            self.supportedMedia = list()
>            hal = Hal()
>            cdromDevice = hal.getDevice(self.udi)
>            for mediaType in self.mediaTypes:
>                try: 
>                    if cdromDevice.getProperty("storage.cdrom.%s" % mediaType):
>                        self.supportedMedia.append(mediaType) # return the most-preferred (first) 
>                except AttributeError:
>                    pass 
>            # always add least-preferred one (last).
>            if not mediaType in self.supportedMedia:
>                self.supportedMedia.append(mediaType)
>        
>        # otherwise - HAL didn't work - try proc filesystem      
>        if not self.supportedMedia or len(self.supportedMedia) is 0:
>            self.getSupportedMediaProc()
>    
>    def showSupportedMedia(self):
>        """ pretty-print supported media list """
>        mediaString = ""
>        for media in self.supportedMedia:          
>           
>            if len(mediaString) > 0:
>                mediaString += ", "
>            mediaString += self.printMediaName(media)
>            
>        return mediaString
>    
>    def printMediaName(self, media):
>        mediaName = media.upper()
>        if "PLUS" in media:
>            mediaName = mediaName.replace("PLUS", "+")
>        elif "DVD" in media and media != "DVD":
>            mediaName = mediaName.replace("DVD", "DVD-")
>        elif "CD" in media and media != "CD":
>            mediaName = mediaName.replace("CD", "CD-")
>        elif mediaName == "CD":
>            mediaName = "CDROM"
>        return media
>    
>    def convertMediaName(self, name):
>        name = name.lower()
>        name = name.replace("-", "")
>        # remove -
>        name = name.replace("+", "plus")
>        if name == "cdrom":
>            name = "cd"
>        return name
>                
>    
>    def getSupportedMediaProc(self):
>        """configure the cdrom based on /proc info, return all supported
>           features. this function should only be used for "virtual"
>           cdroms. for example, IBM's VIO devices.
>           Since we can not find DVD+ features under /proc, so only check
>           cd-rom/cd-r/cd-rw/dvd-rom/dvd-r/dvd-rw on virtual cdroms."""
>        procFileTmp = open("/proc/sys/dev/cdrom/info", "r")
>        procFile = procFileTmp.read()
>        procFileTmp.close()
>        features = list()
>        # if the cdrom drive is USB- based, need to change the name
>        # from "scd*" to "sr*" first, because /proc/sys/dev/cdrom/info
>        # represent USB- cdrom drives as "sr*"
>        driveName = self.getLogicalDeviceName().replace('scd','sr')
>        namePattern = re.compile("^drive name:[\t]*%s" % driveName)
>        featurePattern = re.compile("^Can (read|write) (?P<feature>(CD|DVD)[\+-RW]*):[ \t]+(?P<toggle>(1|0))$")
>        # parse the file
>        lines = procFile.split('\n')
>        driveFound = False
>        for line in lines:
>            if namePattern.match(line):
>                if not driveFound:
>                    driveFound = True
>                else:
>                    break
>                
>            if driveFound:
>                match = featurePattern.match(line)
>                if match and match.group("toggle") is "1":
>                    features.append(self.convertMediaName(match.group("feature")))
>        
>        # compare discovered media against the test's list in order of preference
>        self.supportedMedia = list()
>        for media in self.mediaTypes:
>            if media in features:
>                self.supportedMedia.append(media)
>            
>            # early DVD-R drives couldn't write DVD-RW media, but nearly all
>            # modern drives can, so we assume that DVD-R implies DVD-RW.
>            elif media == "dvdrw" and "dvdr" in features:
>                self.supportedMedia.append("dvdrw")
>            
>        return self.supportedMedia
>    
>    def checkMediaSize(self, size):
>        print "Error: abstract OpticalDiskTest.checkMediaSize call"
>        return False
>
>    def reader(self):
>        """handles CD-ROM or DVD-ROM"""
>        cdromDevice = self.getLogicalDeviceName()
>        if os.path.exists("tmp"):
>            shutil.rmtree("tmp")
>        if os.path.exists("device"):
>            shutil.rmtree("device")
>        os.mkdir("tmp")
>        os.mkdir("device")
>        try:
>            Command("umount /dev/%s" % cdromDevice).echoIgnoreErrors()
>            Command("mount -o ro /dev/%s ./device" % cdromDevice).echo()
>            print "\n  +++ Media mounted !"
>       
>            # check mediaSize
>            # we'll take a guess that anything bigger than 800MB is a DVD
>            mediaSize = Command("df /dev/%s | tail -n1 | awk '{print $3}'" % cdromDevice).getInteger()
>            if not self.checkMediaSize(mediaSize):
>                return False         
>    
>            if os.path.exists("deviceCopy"):
>                shutil.rmtree("deviceCopy")
>            # "shutil.copytree" can not handle file copying here,
>            # use "cp" directly
>            print "\n  Copying files ... \n"
>            sys.stdout.flush()
>            Command("cp -dpR ./device ./deviceCopy").echo()
>            print "\n  Comparing files, this may take some time ... \n"
>            sys.stdout.flush()
>            if self.cmpTree("device","deviceCopy"):
>                print "\n  +++ Compare data succeeded ! \n"
>                returnValue = True
>            else:
>                print "\n  +++ Compare data failed ! \n"
>                returnValue = False
>            Command("umount ./device").echoIgnoreErrors()
>            return returnValue
>        except HTSCommandException, exception:
>            print exception
>            return False
>
>    def writer(self, deviceType):
>        cdromDevice = self.getLogicalDeviceName()
>        """handles CD-R or DVD-R or DVD+R
>           writing a chunk of the filesystem to the media and then checking the
>           file sums"""
>        print "\n  +++ Start writing ...\n"
>        sys.stdout.flush()
>
>        if deviceType == "cdrom":
>            fileDir = "/usr/bin"   # ranges from 100MB-270MB
>        elif deviceType == "dvd" or deviceType == "dvdplus":
>            fileDir = "/usr/share" # ranges from 970MB-3.5GB
>        else:
>            print "\n  +++ Error: Un-supported media .\n"
>            return False
>        # Unmount the drive first in case it's mounted
>        try:
>            Command("umount /dev/%s" % cdromDevice).echoIgnoreErrors()
>    
>            # to handle CD-R or DVD-R
>            if deviceType == "cdrom" or deviceType == "dvd":
>                # check the write mode
>                writeOpts = self.getWriteMode()
>                # support burnfree ?
>                
>                try:
>                    mkisofsSize = Command("mkisofs -quiet -R -print-size %s " % fileDir)
>                    cdblocks= mkisofsSize.getInteger()
>                except HTSCommandException, exception:
>                    # try work-around for RHEL/mkisofs bug 193916
>                    # remove the return character first
>                    item = mkisofsSize.output.pop().replace('\n','')
>                    if item.isdigit(): 
>                        cdblocks = string.atoi(item)
>                        print "Warning: mkisofs had errors"
>                        print exception
>                    else:
>                        print "Error: could not determine block size"
>                        print exception
>                        return False
>                
>                # write the directory to disk
>                recordCommand = Command("mkisofs -quiet -R %s | cdrecord -v %s dev=/dev/%s fs=32M tsize=%ss -" % (fileDir,writeOpts,cdromDevice,cdblocks))
>                recordCommand.echoIgnoreErrors()
>                if recordCommand.returnValue is 0:
>                    print "\n  +++ Write disk succeeded ! \n"
>                    self.handleEject(cdromDevice)
>                else:
>                    print "Error: could not write to disk"
>                    sys.stdout.flush()
>                    return False
>            # DVD+R
>            elif deviceType == "dvdplus":
>                # check media
>                try:
>                    Command("dvd+rw-mediainfo /dev/%s | grep 'Media' | grep 'DVD+'" % cdromDevice).echo()
>                    print "\n  +++ media match ! \n"
>                except HTSCommandException, exception:
>                    print "\n  +++ Error: The disk you inserted is not DVD+ media ! \n"
>                    return False
>                # write the directory to disk
>                sys.stdout.flush()
>                (status,junk) = commands.getstatusoutput("growisofs -Z /dev/%s -quiet -R %s" % (cdromDevice,fileDir))
>                print junk
>                if status == 0:
>                    print "\n  +++ Write DVD+ disk succeeded ! \n"
>                else:
>                    print "\n  +++ Error: Write DVD+ disk failed !\n"
>                    return False
>                self.handleEject(cdromDevice)
>                sys.stdout.flush()
>            else:
>                print "\n  +++ Error: Un-supported media .\n"
>                return False
>            
>            # write done, start to compare
>            return self.compare(deviceType,cdromDevice, fileDir)
>        
>        except HTSCommandException, exception:
>            print exception
>            return False
>        
>    def compare(self, deviceType, cdromDevice, fileDir):
>        
>        Command("umount /dev/%s/" % cdromDevice).echoIgnoreErrors()
>       
>        try:
>            if os.path.exists("media"):
>                print "removing media tree"
>                shutil.rmtree("media")
>            os.mkdir("media")
>        except exception:
>            print exception
>            print "Error: could not create \"media\" mount point"
>            return False
>            
>        print "Mounting /dev/%s to ./media " % cdromDevice
>        sys.stdout.flush()
>        tries = 0
>        maxTries = 10
>        mounted = False
>        waitBeforeRetry = 5 # sec
>        mount = "mount -o ro /dev/%s ./media " % cdromDevice
>        while not mounted and (tries < maxTries):
>            try:
>                print mount
>                sys.stdout.flush()
>                Command(mount).echo()
>                mounted = True
>            except HTSCommandException, exception:
>                tries += 1
>                time.sleep(waitBeforeRetry)
>                print "Mount failed, Retrying..."
>                sys.stdout.flush()
>                
>        if tries >= maxTries:
>            print "Error: could not mount media"
>            sys.stdout.flush()
>            return False
>                
>        print "  Start to compare data, this may take some time, please be patient ... "
>        sys.stdout.flush()
>        if self.cmpTree(fileDir,"media"):
>            print "\n  +++ Compare data succeeded ! \n"
>            returnValue = True
>        else:
>            print "\n  +++ Compare data failed ! \n"
>            returnValue = False
>        Command("umount ./media").echo()
>        os.rmdir("media")
>        return returnValue
>        
>    def rewriter(self,deviceType):
>        """handles CD-RW or DVD-RW or DVD+RW"""
>        cdromDevice = self.getLogicalDeviceName()
>        # Unmount the drive first in case it's mounted
>        Command("umount /dev/%s" % cdromDevice).echoIgnoreErrors()
>
>        # cd-rw and dvd-rw
>        try:
>            if deviceType == "cdrom" or deviceType == "dvd":
>                # Blank the disk first
>                print "\nBlanking the disk, this may take some time, please be patient ...\n"
>                sys.stdout.flush()
>                blankCommand = Command("cdrecord -v dev=/dev/%s blank=fast" % cdromDevice)
>                blankCommand.echoIgnoreErrors()
>                if blankCommand.returnValue is 0:
>                    print "\n  +++ Blank disk done !\n"
>                else:
>                    print "Error: blanking failed"
>                    return False
>                
>                # otherwise
>                self.handleEject(cdromDevice)
>                sys.stdout.flush()
>                # re-write now
>                returnValue = self.writer(deviceType)
>    
>            # dvd+rw
>            elif deviceType == "dvdplus":
>                # check media
>                try:
>                    Command("dvd+rw-mediainfo /dev/%s | grep 'Media' | grep 'DVD+RW'" % cdromDevice).echo()
>                    print "\n  +++ media match ! \n"
>                except HTSCommandException, exception:
>                    print "\n  +++ Error: The disk you inserted is not DVD+RW media ! \n"
>                    return False
>                # blank the disk
>                print "\n  Formatting ... \n"
>                sys.stdout.flush()
>                Command("dvd+rw-format -force /dev/%s 2>/dev/null" % cdromDevice).echo()
>                self.handleEject(cdromDevice)
>                sys.stdout.flush()
>                # re-write now
>                returnValue = self.writer(deviceType)            
>            else:
>                print "\n  +++ Error: Un-supported media .\n"
>                return False
>            return returnValue
>        except HTSCommandException, exception:
>            print exception
>            return False
>    
>    def turnOffAutoMountAndAutoBrowse(self):
>        try:
>            Command("gconftool-2 --type bool --set /desktop/gnome/volume_manager/automount_media false").echo()
>            Command("gconftool-2 --type bool --set /desktop/gnome/volume_manager/autobrowse false").echo()
>        except HTSCommandException, exception:
>            print exception
>            print "Warning: Can not turn off the automount and/or autobrowse for removable media."
>            
>
>    def initializeDeviceInfo(self):
>        
>        self.getSupportedMedia() 
>            
>        if not self.supportedMedia or len(self.supportedMedia) is 0:
>            print "Error: No cdrom device found"
>            return False 
>        
>        self.turnOffAutoMountAndAutoBrowse()
>        
>        return True
>    
>    def testDisk(self, message, readWriteMode, type):
>        
>        # ask user to insert a disk
>        try:
>            Command("eject /dev/%s" % self.getLogicalDeviceName()).run()
>            self.prompt("\n  Please insert a %s disk into %s and then press ENTER to continue !" % (message, self.getLogicalDeviceName()))
>            Command("eject -t /dev/%s" % self.getLogicalDeviceName())
>        except HTSCommandException, exception:
>            print exception
>            # should failures in the eject command be test failures?
>            
>        time.sleep(10)
>        print "\n  +++ Start testing ... \n"
>        sys.stdout.flush()
>        if readWriteMode == "read":
>            return  self.reader()
>        if readWriteMode == "write":
>            return self.writer(type)
>        if readWriteMode == "rewrite":
>            return self.rewriter(type)
>        
>        print "Error: unknown read/write mode %s" % readWriteMode
>        return False
>
>if __name__ == "__main__":
>    test = OpticalDiskTest()
>    returnValue = test.do(sys.argv)
>    sys.exit(returnValue)
>   
>   
>
>

Actions: View

Attachments on bug 462349: 318031 | 318032 | 318033