[Pysvnmanager-svn] SF.net SVN: pysvnmanager:[41] trunk
Status: Alpha
Brought to you by:
jiangx
From: <ji...@us...> - 2008-08-25 10:19:13
|
Revision: 41 http://pysvnmanager.svn.sourceforge.net/pysvnmanager/?rev=41&view=rev Author: jiangx Date: 2008-08-25 10:19:22 +0000 (Mon, 25 Aug 2008) Log Message: ----------- add model for repository add/remove. Modified Paths: -------------- trunk/development.ini trunk/pysvnmanager/config/DefaultConfig.py Added Paths: ----------- trunk/pysvnmanager/model/repos.py trunk/pysvnmanager/tests/test_repos.py Modified: trunk/development.ini =================================================================== --- trunk/development.ini 2008-07-31 13:53:39 UTC (rev 40) +++ trunk/development.ini 2008-08-25 10:19:22 UTC (rev 41) @@ -54,6 +54,9 @@ # authz_file: svn authz config file with administrative extension. (ossxp.com) authz_file = %(here)s/config/svn.access +# svn repository root path +repos_root = %(here)s/svnroot/ + #[filter:proxy-prefix] #use = egg:PasteDeploy#prefix #prefix = /svnadmin Modified: trunk/pysvnmanager/config/DefaultConfig.py =================================================================== --- trunk/pysvnmanager/config/DefaultConfig.py 2008-07-31 13:53:39 UTC (rev 40) +++ trunk/pysvnmanager/config/DefaultConfig.py 2008-08-25 10:19:22 UTC (rev 41) @@ -19,6 +19,9 @@ # Numbers of logs in each page. log_per_page = 10 + # svn repository root + repos_root = config.get('repos_root', "") % {'here': config.get('here')} + # pysvnmanager authentication method. # You can use htpasswd_login, or ldap_login, or both. # You can also implement your own auth module. Added: trunk/pysvnmanager/model/repos.py =================================================================== --- trunk/pysvnmanager/model/repos.py (rev 0) +++ trunk/pysvnmanager/model/repos.py 2008-08-25 10:19:22 UTC (rev 41) @@ -0,0 +1,106 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +"""Subversion authz config file management. + +Basic classes used for Subversion authz management. +""" + +import re +import sys +import os +import StringIO +import logging +log = logging.getLogger(__name__) + +# i18n works only as pysvnmanager (a pylons app) model. +from pylons import config +if config.get('package') and not config.has_key('unittest'): + from pylons.i18n import _ +else: + def _(message): return message + +#reload(sys) # in Python2.5, method sys.setdefaultencoding + #will be delete after initialize. we need reload it. +#sys.setdefaultencoding('utf-8') + +class Repos: + + def __init__(self, repos_root): + if not repos_root or not os.path.exists(repos_root): + raise Exception, _("Repos root does not exist: %s") % repos_root + + self.__repos_root = os.path.abspath(repos_root) + self.__repos_list = [] + self.svnversion_re = re.compile(ur'version\s+(?P<main>[\S]+)(\s+\((?P<sub>.*)\))?') + + def __get_repos_root(self): + return self.__repos_root + + repos_root = property(__get_repos_root) + + def __get_repos_list(self): + self.__repos_list = [] + for i in os.listdir(self.repos_root): + i=unicode(i, 'utf-8') + if self.is_svn_repos(i): + self.__repos_list.append(i) + + return self.__repos_list + + repos_list = property(__get_repos_list) + + def create(self, repos_name): + repos_path = "%(root)s/%(entry)s" % { "root": self.repos_root, "entry": repos_name} + if os.path.exists(repos_path): + raise Exception, _("Repos %s already exists.") % repos_name + from svn import repos as _repos + _repos.create(repos_path, "", "", None, { "fs-type": "fsfs" }) + self.hooks_init(repos_name) + + def hooks_init(self, repos_name): + version = self.svnversion() + repos_path = "%(root)s/%(entry)s" % { "root": self.repos_root, "entry": repos_name} + + def svnversion(self): + cmd = 'LC_ALL=C svn --version' + buff = os.popen(cmd).readline().strip() + m = self.svnversion_re.search(buff) + if m: + return (m.group('main'), m.group('sub')) + else: + return (None, None) + + def is_svn_repos(self, repos_name): + repos_path = "%(root)s/%(entry)s" % { "root": self.repos_root, "entry": repos_name} + if os.path.exists(repos_path): + if os.path.exists("%s/db/revs/0" % repos_path) and \ + os.path.exists("%s/hooks" % repos_path): + return True + return False + + def is_blank_svn_repos(self, repos_name): + if self.is_svn_repos(repos_name): + repos_path = "%(root)s/%(entry)s" % { "root": self.repos_root, "entry": repos_name} + if len(os.listdir("%s/db/revs" % repos_path))!= 1: + return False + elif os.path.isdir("%s/db/revs/0" % repos_path) and \ + len(os.listdir("%s/db/revs/0" % repos_path))!= 1: + return False + else: + return True + + return False + + def delete(self, repos_name): + if self.is_blank_svn_repos(repos_name): + repos_path = "%(root)s/%(entry)s" % { "root": self.repos_root, "entry": repos_name} + from svn import repos as _repos + _repos.delete(repos_path) + else: + raise Exception, _("Repos %s is not a blank repository.") % repos_name + + +if __name__ == '__main__': + import doctest + doctest.testmod() Added: trunk/pysvnmanager/tests/test_repos.py =================================================================== --- trunk/pysvnmanager/tests/test_repos.py (rev 0) +++ trunk/pysvnmanager/tests/test_repos.py 2008-08-25 10:19:22 UTC (rev 41) @@ -0,0 +1,52 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +import os +import sys +sys.path.insert(0,os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + +from pysvnmanager.tests import * +from pysvnmanager import model +from pysvnmanager.model import repos +import StringIO +from pprint import pprint + +class TestRepos(TestController): + + def __init__(self, *args): + repos_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + '/svnroot' + self.repos = repos.Repos(repos_root) + super(TestRepos, self).__init__(*args) + + def setUp(self): + #print '+'*40, 'setup' + pass + + def tearDown(self): + #print '+'*40, 'teardown' + pass + + def testReposRoot(self): + repos.Repos('/tmp') + self.assertRaises(Exception, repos.Repos, '/tmp/svnroot.noexists') + + def testReposlist(self): + self.assert_(sorted(self.repos.repos_list) == ['repos1', 'repos2'], u','.join(self.repos.repos_list).encode('utf-8')) + + def testReposCreate(self): + self.assertRaises(Exception, self.repos.create, 'repos1') + self.repos.create('repos3') + self.assert_(sorted(self.repos.repos_list) == ['repos1', 'repos2', 'repos3'], self.repos.repos_list) + self.repos.delete('repos3') + + def testReposDelete(self): + self.assertRaises(Exception, self.repos.delete, 'repos1') + + def testSvnVersion(self): + svnversion = self.repos.svnversion() + self.assert_(svnversion[0]!='', svnversion[0]) + self.assert_(svnversion[1]!='', svnversion[1]) + +if __name__ == '__main__': + import unittest + unittest.main() This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site. |