Changeset 29

Timestamp:

2008年07月30日 16时18分37秒 (4 years ago)

Author:

jiangx

Message:

See #7: add get_logs, get_page_logs method to RcsLog? class.

Location:

trunk/pysvnmanager

Files:

• model/rcsbackup.py (modified) (3 diffs)
• tests/test_rcs_backup.py (modified) (1 diff)

trunk/pysvnmanager/model/rcsbackup.py

@@ -2 +2 @@
 # -*- coding: utf-8 -*-
 
+from __future__ import division
+
 import os
+import sys
+import re
+import math
 
 import logging
 log = logging.getLogger(__name__)
 
-import sys
 #reload(sys) # in Python2.5, method sys.setdefaultencoding 
             #will be delete after initialize. we need reload it.
@@ -15 +19 @@
 CMD_CO="RCSINIT= co"
 CMD_RCS="RCSINIT= rcs"
+CMD_RLOG="RCSINIT= rlog"
+CMD_RCSDIFF="RCSINIT= rcsdiff"
 
 
@@ -91 +97 @@
         raise Exception, "Command: %s\nError Message: %s\n" % (cmd, buff)
     
+
+class RcsLog(object):
+    
+    def __init__(self, filename):
+        assert filename and isinstance(filename, basestring)
+        self.__file = filename
+        self.__log_per_page = 10
+        self.p = {}
+        self.p['rcs'] = re.compile(r'^RCS file:\s*(.*)$', re.M)
+        self.p['head'] = re.compile(r'^head:\s*(.*)$', re.M)
+        self.p['total'] = re.compile(r'^total revisions:\s*(.*)$', re.M)
+        self.p['revision'] = re.compile(r'^revision\s+([^\s]*)', re.M)
+        self.p['date'] = re.compile(r'^date:\s+(.*);\s+author:', re.M)
+        self.p['author'] = re.compile(r'^date:.*;\s+author:\s*(.*?);', re.M)
+        self.reload()
+    
+    def __get_page_count(self):
+        return int(math.ceil(self.__total/self.__log_per_page))
+    
+    total_page = property(__get_page_count)
+
+    def __get_log_count(self):
+        return self.__total
+    
+    total = property(__get_log_count)
+    
+    def __get_rcsfile(self):
+        return self.__rcsfile
+    
+    rcsfile = property(__get_rcsfile)
+        
+    def __get_head(self):
+        return self.__head
+    
+    head = property(__get_head)
+        
+    def __get_log_per_page(self):
+        return self.__log_per_page
+    
+    def __set_log_per_page(self, count):
+        count = int(count)
+        if count <=0:
+            count = 10
+        self.__log_per_page = count
+
+    log_per_page = property(__get_log_per_page, __set_log_per_page)
+    
+    def reload(self):
+        cmd = '%(cmd)s -L -h -N %(file)s' % {'cmd':CMD_RLOG, 'file':self.__file}
+        buff = os.popen(cmd).read().strip()
+        
+        # RCS file: 1,v
+        m = self.p['rcs'].search(buff)
+        if m:
+            self.__rcsfile = m.group(1)
+        else:
+            self.__rcsfile = ""
+
+        # head: 1.XX
+        m = self.p['head'].search(buff)
+        if m:
+            self.__head = m.group(1)
+        else:
+            self.__head = ""
+            
+        # total revisions: XX
+        m = self.p['total'].search(buff)
+        if m:
+            self.__total = int(m.group(1))
+        else:
+            self.__total = 0
+    
+    def get_page_logs(self, pagenum):
+        total_page = self.total_page
+        
+        if total_page ==0:
+            return []
+
+        if pagenum<=0:
+            pagenum=1
+        elif pagenum>total_page:
+            pagenum=total_page
+        
+        heads = self.__head.rsplit('.',1)
+        rev0=int(heads[1])
+        
+        rev2 = rev0 - ((pagenum-1) * self.__log_per_page) -1
+        if rev2<1: rev2=1
+        rev1 = rev2 - self.__log_per_page + 1
+        if rev1<1:
+            rev1=1
+
+        rev0 = self.__head
+        rev1 = "%s.%d" % (heads[0], rev1)
+        rev2 = "%s.%d" % (heads[0], rev2)
+
+        self.get_logs(rev1, rev2, rev0)
+        return self.revs
+    
+    def get_logs(self, rev1="", rev2="", rev3=""):
+        self.revs=[]
+        opts=""
+        if not rev1:
+            if not rev2:
+                opts=""
+            else:
+                opts="-r:%s" % rev2
+        else:
+            if not rev2:
+                opts="-r%s:" % rev1
+            else:
+                opts="-r%s:%s" % (rev1, rev2)
+
+        if rev3:
+            opts="%s,%s" % (opts, rev3)
+
+        cmd = '%(cmd)s %(opts)s -L -N %(file)s' % {'cmd':CMD_RLOG, 'opts':opts, 'file':self.__file}
+        log.debug('Command: '+cmd)
+        buff = os.popen(cmd).read().strip().rstrip('=').rstrip()
+        
+        while True:
+            pos = buff.rfind('\n'+'-'*28+'\n')
+            if pos==-1:
+                break
+            match = buff[pos+30:]
+            buff=buff[:pos]
+
+            lines=match.split('\n')
+            if len(lines)<3:
+                log.error("wrong rcs format: %s" % match)
+                continue
+
+            # revision 1.XX    locked by: XXX;
+            m = self.p['revision'].search(lines[0])
+            commit_revision = ""
+            if m:
+                commit_revision = m.group(1)
+            else:
+                log.error("not find revision in line: %s" % lines[1])
+                continue
+                        
+            # date: YYYY/MM/DD hh:mm:ss;  author: XX;  ...
+            m = self.p['date'].search(lines[1])
+            commit_time = ""
+            if m:
+                commit_time = get_unicode(m.group(1))
+            else:
+                log.error("not find date in line: %s" % lines[1])
+                continue
+
+            # date: YYYY/MM/DD hh:mm:ss;  author: XX;  ...
+            m = self.p['author'].search(lines[1])
+            commit_author = ""
+            if m:
+                commit_author = get_unicode(eval("'%s'" % m.group(1)))
+            else:
+                log.error("not find author in line: %s" % lines[1])
+                continue
+            
+            # logs...
+            commit_log = get_unicode('\n'.join(lines[2:]))
+            
+            self.revs.append({'revision':commit_revision,
+                              'date':commit_time,
+                              'author':commit_author,
+                              'log':commit_log})
+        
+        return self.revs

trunk/pysvnmanager/tests/test_rcs_backup.py

@@ -122 +122 @@
 
     def testLogs(self):
-        self.writefile()
-        assert os.access(self.wcfile, os.R_OK)
-        print "testLogs"
-        pass
-
+        for i in range(1,11):
+            # new file, backup to r1.1
+            self.writefile(i)
+            rcs.backup(self.wcfile, comment="Test no. %d" % i, user="User1")
+        
+        rcslog = rcs.RcsLog(self.wcfile)
+        assert "rcstest.txt,v" in rcslog.rcsfile, rcslog.rcsfile
+        assert rcslog.head=='1.10', rcslog.head
+        assert rcslog.total==10, rcslog.total
+
+        for i in range(11,15):
+            # new file, backup to r1.1
+            self.writefile(i)
+            rcs.backup(self.wcfile, comment="第 %d 次提交测试。" % i, user="蒋鑫")
+
+        assert rcslog.head=='1.10', rcslog.head
+        rcslog.reload()
+        assert rcslog.head=='1.14', rcslog.head
+        assert rcslog.total==14, rcslog.total
+        
+        logs = rcslog.get_logs()
+        assert len(logs)==14, logs
+        assert logs[3]['revision'] == u'1.4', logs[3]['revision']
+        assert logs[3]['author'] == u'User1', logs[3]['author']
+        assert logs[3]['log'] == u'Test no. 4', logs[3]['log']
+        assert logs[13]['revision'] == u'1.14', logs[13]['revision']
+        assert logs[13]['author'] == u'蒋鑫', logs[13]['author'].encode('utf-8')
+        assert logs[13]['log'] == u'第 14 次提交测试。', logs[13]['log'].encode('utf-8')
+        
+        logs = rcslog.get_logs('1.9','1.12')
+        assert len(logs)==4, logs
+        assert logs[1]['revision'] == u'1.10', logs[1]['revision']
+        assert logs[1]['author'] == u'User1', logs[1]['author']
+        assert logs[1]['log'] == u'Test no. 10', logs[1]['log']
+        assert logs[2]['revision'] == u'1.11', logs[2]['revision']
+        assert logs[2]['author'] == u'蒋鑫', logs[2]['author'].encode('utf-8')
+        assert logs[2]['log'] == u'第 11 次提交测试。', logs[2]['log'].encode('utf-8')
+
+
+        logs = rcslog.get_logs('','1.12')
+        assert len(logs)==12, logs
+        assert logs[10]['revision'] == u'1.11', logs[10]['revision']
+
+        logs = rcslog.get_logs('1.12','')
+        assert len(logs)==3, len(logs)
+        assert logs[1]['revision'] == u'1.13', logs[1]['revision']
+
+        logs = rcslog.get_logs('1.7','1.11', '1.14')
+        assert len(logs)==6, len(logs)
+        assert logs[4]['revision'] == u'1.11', logs[1]['revision']
+        assert logs[5]['revision'] == u'1.14', logs[1]['revision']
+        
+        assert rcslog.total_page == 2, rcslog.total_page
+        rcslog.log_per_page=0
+        assert rcslog.log_per_page==10, rcslog.log_per_page
+        rcslog.log_per_page=5
+        assert rcslog.log_per_page==5, rcslog.log_per_page
+        assert rcslog.total_page == 3, rcslog.total_page
+        logs = rcslog.get_page_logs(1)
+        logs2= rcslog.get_page_logs(0)
+        assert logs == logs2
+        assert [x['revision'] for x in logs] == \
+                ['1.9', '1.10', '1.11', '1.12', '1.13', '1.14'], \
+                [x['revision'] for x in logs]
+
+        logs = rcslog.get_page_logs(2)
+        assert [x['revision'] for x in logs] == \
+                ['1.4', '1.5', '1.6', '1.7', '1.8', '1.14'], \
+                [x['revision'] for x in logs]
+
+        logs = rcslog.get_page_logs(3)
+        logs2= rcslog.get_page_logs(30)
+        assert logs == logs2
+        assert [x['revision'] for x in logs] == \
+                ['1.1', '1.2', '1.3', '1.14'], \
+                [x['revision'] for x in logs]
+
+        
+    def testLogsNone(self):
+        rcslog = rcs.RcsLog(self.wcfile)
+        assert rcslog.rcsfile == "", rcslog.rcsfile
+        assert rcslog.total == 0, rcslog.total
+        assert rcslog.get_page_logs(1) == [], rcslog.get_page_logs(1)
+        
 if __name__ == '__main__': 
     import unittest