pymoul-svn Mailing List for pyMoul (Page 9)

SourceForge Headquarters 1320 Columbia Street Suite 310 San Diego, CA 92101 +1 (858) 422-6466

Revision: 99
          http://pymoul.svn.sourceforge.net/pymoul/?rev=99&view=rev
Author:   tiran
Date:     2007-01-29 14:07:42 -0800 (Mon, 29 Jan 2007)

Log Message:
-----------
First working implementation of log zipper and chatlog mover with unit tests

Modified Paths:
--------------
    pymoul/trunk/Makefile.in
    pymoul/trunk/src/moul/file/chatlog.py
    pymoul/trunk/src/moul/file/plasmalog.py
    pymoul/trunk/src/moul/file/tests/test_plasmalog.py

Modified: pymoul/trunk/Makefile.in
===================================================================

--- pymoul/trunk/Makefile.in	2007-01-29 18:46:30 UTC (rev 98)
+++ pymoul/trunk/Makefile.in	2007-01-29 22:07:42 UTC (rev 99)
@@ -59,11 +59,12 @@
 doc: doc_html
 
 clean:
-	find . \( -name '*.o' -o -name '*.c' -o -name '*.so' -o -name '*.py[cod]' -o -name '*.dll' \) -exec rm -f {} \;
+	find . \( -name '*.o' -o -name '*~' -o -name '*.so' -o -name '*.py[cod]' -o -name '*.dll' \) -exec rm -f {} \;
 	rm -rf build
 
 realclean: clean
 	rm -f TAGS
+	rm -rf dist
 	$(PYTHON) setup.py clean -a
 
 propset:

Modified: pymoul/trunk/src/moul/file/chatlog.py
===================================================================
--- pymoul/trunk/src/moul/file/chatlog.py	2007-01-29 18:46:30 UTC (rev 98)
+++ pymoul/trunk/src/moul/file/chatlog.py	2007-01-29 22:07:42 UTC (rev 99)
@@ -45,8 +45,8 @@
     r"^\((?P<M>\d{1,2})/(?P<D>\d{1,2})\ " # MM/DD
     r"(?P<h>\d{1,2}):(?P<m>\d\d{1,2}):(?P<s>\d{1,2})\)" # hh:mm:ss
     r"(?P<space>\ {1,2})(?P<msg>.*)$") # text
-CHAT_STARTED = "Chat.log started..."
-CHAT_STOPPED = "...Chat.log stopped."
+CHATLOG_STARTED = "Chat.log started..."
+CHATLOG_STOPPED = "...Chat.log stopped."
 TEXT_MSGFROM = re.compile(
     r"From (?P<user>.*) in (?P<location>.*): (?P<msg>.*)"
     # From USER in LOCATION: msg
@@ -76,17 +76,17 @@
     """
     """
     _pat = "chat*.log"
-    _filename = "chatlog_%(sY)i%(sM)i%(sD)i_%(ch)i%(cm)i_%(sh)i%(sm)i.txt"
+    _filename = "chatlog_%(sY)02i%(sM)02i%(sD)02i_%(ch)02i%(cm)02i_%(sh)02i%(sm)02i.txt"
 
     def __init__(self, srcdir, destdir):
-         self._path = srcdir
-         self._dest = destdir
-         self._logs = []
-         if not os.path.isdir(srcdir):
-             LOG.critical("%s is not a directory" % srcdir)
-         if not os.path.isdir(destdir):
-             LOG.info("Creating chatlog directory %s" % destdir)
-             os.mkdir(dest)
+        self._path = srcdir
+        self._dest = destdir
+        self._logs = []
+        if not os.path.isdir(srcdir):
+            LOG.critical("%s is not a directory" % srcdir)
+        if not os.path.isdir(destdir):
+            LOG.info("Creating chatlog directory %s" % destdir)
+            os.mkdir(destdir)
 
     def findLogs(self):
         """Find log files in self._path
@@ -95,7 +95,7 @@
         for the file.
         """
         for file in os.listdir(self._path):
-            if not fnmatch(file, self._pat):
+            if not fnmatch(file.lower(), self._pat):
                 continue
 
             fpath = os.path.join(self._path, file)
@@ -106,13 +106,13 @@
             created = self._findCreated(fd)
             if not created:
                 LOG.warning("Failed to parse chatlog %s" % fpath)
-            modtime = localtime(modtime(fpath))
-            newname = self._makeFilename(created, modtime)
+            mtime = localtime(modtime(fpath))
+            newname = self._makeFilename(created, mtime)
 
             details = {
                 'fpath' : fpath,
                 'filename' : file,
-                'modtime' : localtime(modtime(fpath)),
+                'modtime' : mtime,
                 'created' : created,
                 'newname' : newname,
                 'newpath' : os.path.join(self._dest, newname),
@@ -132,8 +132,9 @@
             if mo:
                 data = mo.groupdict()
                 if data['msg'].startswith(CHATLOG_STARTED):
-                    created = (None, mo['M'], mo['D'], mo['h'], mo['m'],
-                                mo['s'], None, None, None)
+                    d = mo.groupdict()
+                    created = (None, int(d['M']), int(d['D']), int(d['h']),
+                               int(d['m']), int(d['s']), None, None, None)
         return created
 
     def _makeFilename(self, created, stopped):
@@ -141,10 +142,10 @@
        """
        c, s = created, stopped
        mapping = {
-           'sY' : s['Y'], 'sM' : s['YM'], 'sD' : s['D'],
-           'sh' : s['h'], 'sm' : s['m'], 'ss' : s['s'],
-                          'cM' : c['YM'], 'cD' : c['D'],
-           'ch' : c['h'], 'cm' : c['m'], 'cs' : c['s'],
+           'sY' : s[0], 'sM' : s[1], 'sD' : s[2],
+           'sh' : s[3], 'sm' : s[4], 'ss' : s[5],
+                        'cM' : c[1], 'cD' : c[2],
+           'ch' : c[3], 'cm' : c[4], 'cs' : c[5],
            }
        return self._filename % mapping
 
@@ -158,3 +159,7 @@
                 LOG.warning("%s exists - overwriting" % new)
             LOG.info("Move chatlog %s to %s" % (old, new))
             move(old, new)
+
+    def __call__(self):
+        self.findLogs()
+        self.moveChatlogs()

Modified: pymoul/trunk/src/moul/file/plasmalog.py
===================================================================
--- pymoul/trunk/src/moul/file/plasmalog.py	2007-01-29 18:46:30 UTC (rev 98)
+++ pymoul/trunk/src/moul/file/plasmalog.py	2007-01-29 22:07:42 UTC (rev 99)
@@ -28,8 +28,11 @@
 from fnmatch import fnmatch
 
 from moul.crypt.elf import decryptElf
+from moul.log import getLogger
 
 PLASMA_LOG = "plasmalog.txt"
+
+LOG = getLogger('moul.plasma')
 _marker = object()
 
 class PlasmalogZipper(object):
@@ -40,38 +43,52 @@
     def __init__(self, srcdir, destdir):
         self._srcdir = srcdir
         self._destdir = destdir
+        self._zipped_dirs = {} # dirname -> zipfile
+        self._not_removed = [] # files with full path
 
-    def getTimeStamp(self, path):
+        if not os.path.isdir(srcdir):
+            LOG.critical("%s is not a directory" % srcdir)
+        if not self.isPlasmaLogDir():
+            self._ispldir = False
+            LOG.critical("%s is not a plasma log directory" % srcdir)
+        else:
+            self._ispldir = True
+        if not os.path.isdir(destdir):
+            LOG.info("Creating chatlog directory %s" % destdir)
+            os.mkdir(destdir)
+
+    @staticmethod
+    def getTimeStamp(path):
         """Get time stamp yyyymmdd_hhmm based in the modification time
         """
         sec = os.stat(path)[ST_MTIME]
         return time.strftime("%Y%m%d_%H%M", time.localtime(sec))
-    
-    def isLogDir(self, path):
+
+    def isPlasmaLogDir(self, path=None):
         """Check if a path is a valid plasma log directory
-    
+
         Just checks for the plasmalog.txt file
-    
         Returns either False or a time stamp    
         """
+        if path is None:
+            path = self._srcdir
         pl = os.path.join(path, PLASMA_LOG)
         if not os.path.isfile(pl):
             return False
-        return getTimeStamp(pl)
+        return self.getTimeStamp(pl)
     
     def zipLogDir(self):
         """Zip all log files
     
         This function also zips subdirectories.    
         """
-        stored_dirs = []
-    
         for root, dirs, files in os.walk(self._srcdir):
-            stamp = isLogDir(root)
+            stamp = self.isPlasmaLogDir(root)
             if not stamp:
                 continue
             name = os.path.basename(root)
             zipname = "%s_%s.zip" % (name, stamp)
+            self._zipped_dirs[name] = zipname
             zipfd = zipfile.ZipFile(os.path.join(self._destdir, zipname),
                                     'w', zipfile.ZIP_DEFLATED)
             for file in files:
@@ -83,24 +100,27 @@
                 zipfd.write(fname, arcname)
             #zipfd.printdir()
             zipfd.close()
-            stored_dirs.append((root, zipname))
 
-        return stored_dirs
+    def removeLogs(self):
+        """Removes log directories
 
-    def removeLogs(self, logdir):
-        """Removes log directories
-    
         The removeLogs function removes only files considered as safe
         """
-        for root, dirs, files in os.walk(logdir, topdown=False):
+        for root, dirs, files in os.walk(self._srcdir, topdown=False):
             for name in files:
                 matches = [pat for pat in self._save_patterns
-                           if fnmatch(basename, pat)]
-                if match:
-                    os.remove(os.path.join(root, name))
+                           if fnmatch(name, pat)]
+                fpath = os.path.join(root, name)
+                if matches:
+                    os.remove(fpath)
                 else:
+                    self._not_removed.append(fpath)
                     LOG.warning("Won't remove %s from %s" % (name, root))
             for name in dirs:
                 os.rmdir(os.path.join(root, name))
 
-        os.rmdir(logdir)
+        os.rmdir(self._srcdir)
+
+    def __call__(self):
+        self.zipLogDir()
+        self.removeLogs()

Modified: pymoul/trunk/src/moul/file/tests/test_plasmalog.py
===================================================================
--- pymoul/trunk/src/moul/file/tests/test_plasmalog.py	2007-01-29 18:46:30 UTC (rev 98)
+++ pymoul/trunk/src/moul/file/tests/test_plasmalog.py	2007-01-29 22:07:42 UTC (rev 99)
@@ -24,10 +24,102 @@
 import unittest
 from doctest import DocTestSuite
 
-import moul.file.plasmalog
+import os
+import shutil
+from tempfile import mkdtemp
 
+from moul.file.plasmalog import PlasmalogZipper
+from moul.file.chatlog import ChatlogMover
+
+FILE_LIST = ['audio.0.elf', 'audiocaps.0.elf', 'audioTimes.0.elf',
+    'Avatar.0.elf', 'impacts.0.elf', 'LocalizationDataMgr.0.elf',
+    'Marker.0.elf', 'NetScreener.0.elf', 'patcher.0.elf', 'pipeline.0.elf',
+    'plasmalog.txt', 'Python.0.elf', 'Simulation.0.elf', 'voice.0.elf',
+    'Chat.0.log']
+
+CHATLOG = """(01/02 03:04:05) Chat.log started...
+(01/02 03:04:06)  Testuser: foo
+(01/02 03:04:07) ...Chat.log stopped.
+"""
+
+base = os.path.dirname(__file__)
+
+def _fakeLogdir(path):
+    """Create a fake logdir
+    """
+    if not os.path.isdir(path):
+        os.mkdir(path)
+    for fname in FILE_LIST:
+        fd = open(os.path.join(path, fname), 'w')
+        fd.write('dummy')
+        fd.close()
+
+    fd = open(os.path.join(path, 'Chat.0.log'), 'w')
+    fd.write(CHATLOG)
+    fd.close()
+    fd = open(os.path.join(path, 'plasmalog.txt'), 'w')
+    fd.write('')
+    fd.close()
+
+def _fakeUruLiveDir(path):
+    """Create a fake Uru Live directory
+    """
+    for dname in ('Avatars', 'init', 'KIimages'):
+        os.mkdir(os.path.join(path, dname))
+    _fakeLogdir(os.path.join(path, 'Logs'))
+    _fakeLogdir(os.path.join(path, 'Logs', 'faketest'))
+    shutil.copyfile(os.path.join(base, 'avatar.jpg'),
+                    os.path.join(path, 'Avatars', '001.jpg'))
+    shutil.copyfile(os.path.join(base, 'graphics.ini'),
+                    os.path.join(path, 'init', 'graphics.ini'))
+    shutil.copyfile(os.path.join(base, 'audio.ini'),
+                    os.path.join(path, 'init', 'audio.ini'))
+
+class PlasmaChatTest(unittest.TestCase):
+    def setUp(self):
+        self._tmpdir = mkdtemp()
+        self._chatdest = os.path.join(self._tmpdir, 'Chatlogs')
+        self._logdest = os.path.join(self._tmpdir, 'ZippedLogs')
+        self._logdir = os.path.join(self._tmpdir, 'Logs')
+        _fakeUruLiveDir(self._tmpdir)
+        self._clm = ChatlogMover(self._logdir, self._chatdest)
+        self._plz = PlasmalogZipper(self._logdir, self._logdest)
+
+    def test_createDir(self):
+        self.failUnless(os.path.isdir(self._logdest), self._logdest)
+        self.failUnless(os.path.isdir(self._chatdest), self._chatdest)
+
+    def test_plasmaLogZipper(self):
+        plz = self._plz
+        self.failUnless(plz.isPlasmaLogDir())
+        plz.zipLogDir()
+        content = os.listdir(self._logdest)
+        content.sort()
+        self.failUnlessEqual(len(content), 2, content)
+        self.failUnless(content[0].startswith("Logs_"), content[0])
+        self.failUnless(content[1].startswith("faketest_"), content[1])
+
+        plz.removeLogs()
+        self.failIf(os.path.isdir(self._logdir))
+
+    def test_chatLogMover(self):
+        clm = self._clm
+        clm.findLogs()
+        clm.moveChatlogs()
+        content = os.listdir(self._chatdest)
+        self.failUnlessEqual(len(content), 1, content)
+        length = len("chatlog_20070129_0304_2302.txt")
+        c = content[0]
+        self.failUnless(c.startswith("chatlog_"))
+        self.failUnless(c.endswith(".txt"))
+        self.failUnlessEqual(len(c), length, c)
+
+    def tearDown(self):
+        shutil.rmtree(self._tmpdir)
+
 def test_suite():
     return unittest.TestSuite((
+        unittest.makeSuite(PlasmaChatTest),
         DocTestSuite('moul.file.plasmalog')
         ))
 


This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.




pymoul-svn Mailing List for pyMoul (Page 9)

pymoul-svn — Subversion commit list