antoine 2004/03/04 13:38:47
Modified: python/gump/test pyunit.py
Added: python/gump/utils sync.py
python/gump/test sync.py
Log:
a sync utility with 4 unit tests
Revision Changes Path
1.1 gump/python/gump/utils/sync.py
Index: sync.py
===================================================================
#!/usr/bin/env python
# $Header: /home/cvs/gump/python/gump/utils/sync.py,v 1.1 2004/03/04 21:38:47
antoine Exp $
# $Revision: 1.1 $
# $Date: 2004/03/04 21:38:47 $
#
# ====================================================================
#
# The Apache Software License, Version 1.1
#
# Copyright (c) 2004 The Apache Software Foundation. All rights
# reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# 3. The end-user documentation included with the redistribution, if
# any, must include the following acknowlegement:
# "This product includes software developed by the
# Apache Software Foundation (http://www.apache.org/)."
# Alternately, this acknowlegement may appear in the software itself,
# if and wherever such third-party acknowlegements normally appear.
#
# 4. The names "The Jakarta Project", "Alexandria", and "Apache Software
# Foundation" must not be used to endorse or promote products derived
# from this software without prior written permission. For written
# permission, please contact [EMAIL PROTECTED]
#
# 5. Products derived from this software may not be called "Apache"
# nor may "Apache" appear in their names without prior written
# permission of the Apache Group.
#
# THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
# WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
# ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
# OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
# ====================================================================
#
# This software consists of voluntary contributions made by many
# individuals on behalf of the Apache Software Foundation. For more
# information on the Apache Software Foundation, please see
# <http://www.apache.org/>.
"""
Helper Stuff
"""
import logging
import os.path
from stat import *
import shutil
from gump import log
from gump.utils.work import *
from gump.utils.file import *
from gump.utils.launcher import *
class Sync:
"""
this class can be used to sync two directories
x = Sync(sourcedir, targetdir) # construct
x.execute() #run
if targetdir does not exist, it will be created
if sourcedir does not exist, the class will raise an IOError
"""
def __init__(self, sourcedir, targetdir):
self.sourcedir = sourcedir
self.targetdir = targetdir
def execute(self):
log.info('Starting sync from [' + self.sourcedir + ']')
log.info(' target dir [' + self.targetdir + ']')
if not os.path.exists(self.sourcedir):
log.error('Exiting sync, source directory does not exist [' +
self.sourcedir + ']')
raise IOError, 'source directory does not exist [' + self.sourcedir + ']'
return
if not os.path.isdir(self.sourcedir):
log.error('Exiting sync, source is not a directory [' + self.sourcedir +
']')
raise IOError, 'source is not a directory [' + self.sourcedir + ']'
return
if not os.path.exists(self.targetdir):
try:
os.makedirs(self.targetdir)
except Exception, details:
log.exception('failed on ' + str(details))
return
copytree(self.sourcedir, self.targetdir, 1)
def copytree(src, dst, symlinks=0):
names = os.listdir(src)
try:
result = os.stat(dst)
except Exception, details:
result = None
# handle case where result exists but is not a directory
if result and not S_ISDIR(result[ST_MODE]):
remove(dst)
result = None
if not result:
os.makedirs(dst)
if result:
names2 = os.listdir(dst)
epurate(src, dst, names, names2)
for name in names:
srcname = os.path.join(src, name)
dstname = os.path.join(dst, name)
try:
if symlinks and os.path.islink(srcname):
linkto = os.readlink(srcname)
os.symlink(linkto, dstname)
elif os.path.isdir(srcname):
copytree(srcname, dstname, symlinks)
else:
maybecopy(srcname, dstname)
except (IOError, os.error), why:
message = "Can't copy [%s] to [%s]: [%s]" % (`srcname`, `dstname`,
str(why))
log.exception(message)
raise IOError, message
def epurate(sourcedir, destdir, acceptablefiles, existingfiles):
"""
this routine will delete from a set of existing files
in a directory the one which are not part of an
array of acceptablefiles
somedir = directory where the epuration is to take place
"""
for afile in existingfiles:
fullsourcefile = os.path.join(sourcedir, afile)
fulldestfile = os.path.join(destdir, afile)
if not afile in acceptablefiles:
tobedeleted = os.path.join(destdir, afile)
result = os.stat(tobedeleted)
if S_ISDIR(result[ST_MODE]):
log.debug('attempting to remove directory [%s]' % (`tobedeleted`))
shutil.rmtree(tobedeleted)
else:
log.debug('attempting to remove file [%s]' % (`tobedeleted`))
os.remove(tobedeleted)
elif os.path.isdir(fullsourcefile) and not os.path.isdir(fulldestfile):
log.debug('removing file [%s] to be replaced by directory'
%(`fulldestfile`))
os.remove(fulldestfile)
elif os.path.isfile(fullsourcefile) and os.path.isdir(fulldestfile):
log.debug('removing directory [%s] to be replaced by file'
%(`fulldestfile`))
shutil.rmtree(fulldestfile)
def maybecopy(srcname, dstname):
"""
copy a file from srcname to dstname if
dstname does not exist
or srcname and dstname have different dates
or srcname and dstname have different sizes
"""
result = os.stat(srcname)
try:
result2 = os.stat(dstname)
except (Exception), details:
result2 = None
okcopy = 0
if not result2:
okcopy = 1
elif S_ISDIR(result2[ST_MODE]):
shutil.rmtree(dstname)
okcopy = 1
elif result[ST_SIZE] != result2[ST_SIZE]:
okcopy = 1
elif result[ST_MTIME] != result2[ST_MTIME]:
okcopy = 1
if okcopy:
log.debug("Attempting copy from [%s] to [%s]" %(`srcname`, `dstname`))
shutil.copy2(srcname, dstname)
1.22 +7 -4 gump/python/gump/test/pyunit.py
Index: pyunit.py
===================================================================
RCS file: /home/cvs/gump/python/gump/test/pyunit.py,v
retrieving revision 1.21
retrieving revision 1.22
diff -u -r1.21 -r1.22
--- pyunit.py 18 Feb 2004 00:13:54 -0000 1.21
+++ pyunit.py 4 Mar 2004 21:38:47 -0000 1.22
@@ -334,6 +334,9 @@
#:TODO: Figure out Python search/introspection to find these...
+ from gump.test.sync import SyncTestSuite
+ runner.addSuite(SyncTestSuite())
+
from gump.test.utils import UtilsTestSuite
runner.addSuite(UtilsTestSuite())
@@ -377,4 +380,4 @@
# Perform the tests...
runner.run(patterns)
-
\ No newline at end of file
+
1.1 gump/python/gump/test/sync.py
Index: sync.py
===================================================================
#!/usr/bin/env python
#
# $Header: /home/cvs/gump/python/gump/test/sync.py,v 1.1 2004/03/04 21:38:47 antoine
Exp $
# $Revision: 1.1 $
# $Date: 2004/03/04 21:38:47 $
#
# ====================================================================
#
# The Apache Software License, Version 1.1
#
# Copyright (c) 2003 The Apache Software Foundation. All rights
# reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in
# the documentation and/or other materials provided with the
# distribution.
#
# 3. The end-user documentation included with the redistribution, if
# any, must include the following acknowlegement:
# "This product includes software developed by the
# Apache Software Foundation (http://www.apache.org/)."
# Alternately, this acknowlegement may appear in the software itself,
# if and wherever such third-party acknowlegements normally appear.
#
# 4. The names "The Jakarta Project", "Alexandria", and "Apache Software
# Foundation" must not be used to endorse or promote products derived
# from this software without prior written permission. For written
# permission, please contact [EMAIL PROTECTED]
#
# 5. Products derived from this software may not be called "Apache"
# nor may "Apache" appear in their names without prior written
# permission of the Apache Group.
#
# THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
# WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
# OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL THE APACHE SOFTWARE FOUNDATION OR
# ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
# OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
# SUCH DAMAGE.
# ====================================================================
#
# This software consists of voluntary contributions made by many
# individuals on behalf of the Apache Software Foundation. For more
# information on the Apache Software Foundation, please see
# <http://www.apache.org/>.
"""
Utils Testing
"""
from gump.utils.sync import Sync
from gump.test.pyunit import UnitTestSuite
from gump import log
import os.path
import shutil
import time
import stat
class SyncTestSuite(UnitTestSuite):
def __init__(self):
UnitTestSuite.__init__(self)
self.source = "./test/source"
self.destination = "./test/destination"
self.source_subdir1 = os.path.join(self.source, 'subdir1')
self.destination_subdir1 = os.path.join(self.destination, 'subdir1')
self.alphatxt = 'alpha.txt'
self.source_alphatxt = os.path.join(self.source_subdir1, self.alphatxt)
self.destination_alphatxt = os.path.join(self.destination_subdir1,
self.alphatxt)
def setUp(self):
"""
this setup creates a subdirectory at source
then a file in the subdirectory, with an arbitrary date time
"""
self.tearDown()
os.makedirs(self.source_subdir1)
myfile = file(self.source_alphatxt, 'w+')
myfile.write('Hello World')
myfile.close()
# Sat, 20 May 2000 12:07:40 +0000
sometime=[2000,5,20,12,7,40,5,141,-1]
epoch_sometime = time.mktime(sometime)
os.utime(self.source_alphatxt, (epoch_sometime, epoch_sometime))
def tearDown(self):
if os.path.exists(self.source):
log.debug('attempting to remove directory [%s]' % (`self.source`))
shutil.rmtree(self.source)
if os.path.exists(self.destination):
log.debug('attempting to remove directory [%s]' % (`self.destination`))
shutil.rmtree(self.destination)
def testSimpleSync(self):
"""
assuming the setUp gets done,
a sync runs
the test checks :
- that the destination directory gets created
- that the destination subdirectory gets created
- that a file exists in the destination subdir
with the same size and modification time as
the original file
"""
mySync = Sync(self.source, self.destination)
mySync.execute()
try:
result = os.stat(self.destination)
except Exception, details:
raiseIssue(['destination directory was not created', self.destination])
try:
result = os.stat(self.destination_subdir1)
except Exception, details:
raiseIssue(['destination_subdir1 directory was not created',
self.destination_subdir1])
result_source = None
result_destination = None
try:
result_source = os.stat(self.source_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.source_alphatxt])
try:
result_destination = os.stat(self.destination_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.destination_alphatxt])
log.debug("size of file [%s] is %i" % (`self.destination_alphatxt`,
result_destination[stat.ST_SIZE]))
log.debug("modification date of file [%s] is %s" %
(`self.destination_alphatxt`,
time.ctime(result_destination[stat.ST_MTIME])))
self.assertTrue("modtime is equal for [%s] compared to [%s]"
%(`self.source_alphatxt`,`self.destination_alphatxt`),
result_source[stat.ST_MTIME]==result_destination[stat.ST_MTIME])
self.assertTrue("size is equal for [%s] compared to [%s]"
%(`self.source_alphatxt`,`self.destination_alphatxt`),
result_source[stat.ST_SIZE]==result_destination[stat.ST_SIZE])
def testRemoveJunkDestinationFile(self):
"""
assuming the setUp gets done,
a sync runs
then a file is added at destination
the timestamp of the destination file gets changed
then another sync
the test checks :
- that the extra destination file is removed
- that the destination file gets copied again
with the same size and modification time as
the original file
"""
mySync = Sync(self.source, self.destination)
mySync.execute()
# create the destination junk file
destination_junktxt = os.path.join(self.destination_subdir1,
'junk.txt')
shutil.copy2(self.destination_alphatxt, destination_junktxt)
sometime=[2000,5,20,12,7,45,5,141,-1]
epoch_sometime = time.mktime(sometime)
os.utime(self.destination_alphatxt, (epoch_sometime, epoch_sometime))
mySync.execute()
if os.path.exists(destination_junktxt):
raiseIssue(['junk file was not deleted', destination_junktxt])
result_source = None
result_destination = None
try:
result_source = os.stat(self.source_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.source_alphatxt])
try:
result_destination = os.stat(self.destination_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.destination_alphatxt])
log.debug("size of file [%s] is %i" % (`self.destination_alphatxt`,
result_destination[stat.ST_SIZE]))
log.debug("modification date of file [%s] is %s" %
(`self.destination_alphatxt`,
time.ctime(result_destination[stat.ST_MTIME])))
self.assertTrue("modtime is equal for [%s] compared to [%s]"
%(`self.source_alphatxt`,`self.destination_alphatxt`),
result_source[stat.ST_MTIME]==result_destination[stat.ST_MTIME])
def testDestinationFileBecomesDirectory(self):
"""
assuming the setUp gets done,
a sync runs
then destination_alphatxt is deleted and replaced by a directory
in this directory another subdir, a file, and another file in the subdir
then another sync
the test checks :
- that the destination file gets copied again
with the same size and modification time as
the original file
"""
mySync = Sync(self.source, self.destination)
mySync.execute()
os.remove(self.destination_alphatxt)
junk_subdir = os.path.join(self.destination_alphatxt, "junk.dir")
os.makedirs(junk_subdir)
junk_file1 = os.path.join(self.destination_alphatxt, "junk.txt")
junk_file2 = os.path.join(junk_subdir, "junk.txt")
shutil.copy2(self.source_alphatxt, junk_file1)
shutil.copy2(self.source_alphatxt, junk_file2)
mySync.execute()
if os.path.isdir(self.destination_alphatxt):
raiseIssue(['destination text file remained a directory',
self.destination_alphatxt])
result_source = None
result_destination = None
try:
result_source = os.stat(self.source_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.source_alphatxt])
try:
result_destination = os.stat(self.destination_alphatxt)
except Exception, details:
raiseIssue(['file was not created', self.destination_alphatxt])
log.debug("size of file [%s] is %i" % (`self.destination_alphatxt`,
result_destination[stat.ST_SIZE]))
log.debug("modification date of file [%s] is %s" %
(`self.destination_alphatxt`,
time.ctime(result_destination[stat.ST_MTIME])))
self.assertTrue("modtime is equal for [%s] compared to [%s]"
%(`self.source_alphatxt`,`self.destination_alphatxt`),
result_source[stat.ST_MTIME]==result_destination[stat.ST_MTIME])
def testOriginFileBecomesDirectory(self):
"""
assuming the setUp gets done,
a sync runs
then source_alphatxt is deleted and replaced by a directory
in this directory another subdir, a file, and another file in the subdir
then another sync
the test checks :
- that the alpha.txt file gets replaced by a directory at destination
- that the directory tree below alpha.txt is the same at
destination like at source
"""
mySync = Sync(self.source, self.destination)
mySync.execute()
os.remove(self.source_alphatxt)
junk_subdir = os.path.join(self.source_alphatxt, "junk.dir")
os.makedirs(junk_subdir)
junk_source_file1 = os.path.join(self.source_alphatxt, "junk.txt")
myfile = file(junk_source_file1, 'w+')
myfile.write('Hello World')
myfile.close()
junk_source_file2 = os.path.join(junk_subdir, "junk.txt")
shutil.copy2(junk_source_file1, junk_source_file2)
mySync.execute()
if os.path.isfile(self.destination_alphatxt):
raiseIssue(['destination text file remained a file',
self.destination_alphatxt])
self.genericCompare(self.source_alphatxt, self.destination_alphatxt)
log.debug('finished')
def genericCompare(self, mysource, mydestination):
"""
compare 2 directories source and destination
"""
if not os.path.isdir(mysource):
raiseIssue([mysource, ' not a directory'])
if not os.path.isdir(mydestination):
raiseIssue([mydestination, ' not a directory'])
names = os.listdir(mysource)
for aname in names:
inode_source = os.path.join(mysource, aname)
inode_dest = os.path.join(mydestination, aname)
if os.path.isfile(inode_source):
self.compareFiles(inode_source, inode_dest)
elif os.path.islink(inode_source):
if not os.path.islink(inode_dest):
raiseIssue([inode_dest, ' not a symbolic link'])
linkto_source = os.readlink(inode_source)
linkto_dest = os.readlink(inode_dest)
self.assertTrue([inode_dest, ' points to ', linkto_source ],
linkto_source==linkto_dest)
elif os.path.isdir(inode_source):
self.genericCompare(inode_source, inode_dest)
def compareFiles(self, inode_source, inode_dest):
"""
compare 2 files
"""
result_source = None
result_dest = None
try:
result_source = os.stat(inode_source)
except Exception, details:
raiseIssue(['could not stat ', inode_source])
try:
result_dest = os.stat(inode_dest)
except Exception, details:
raiseIssue(['could not stat ', inode_dest])
self.assertTrue("modtime is equal for [%s] compared to [%s]"
%(`inode_source`,`inode_dest`),
result_source[stat.ST_MTIME]==result_dest[stat.ST_MTIME])
self.assertTrue("size is equal for [%s] compared to [%s]"
%(`inode_source`,`inode_dest`),
result_source[stat.ST_SIZE]==result_dest[stat.ST_SIZE])
---------------------------------------------------------------------
To unsubscribe, e-mail: [EMAIL PROTECTED]
For additional commands, e-mail: [EMAIL PROTECTED]
