move test files out of py lib proper
* separate all tests from plugins * simplify implicit inclusion of plugins under test * have test_initpkg perform direct checks instead of yielding tests * fix example tests for 3k --HG-- branch : trunk
This commit is contained in:
holger krekel
443
testing/path/test_svnauth.py
Normal file
443
testing/path/test_svnauth.py
Normal file
@@ -0,0 +1,443 @@
|
||||
import py
|
||||
from testing.path import svntestbase
|
||||
from py.path import SvnAuth
|
||||
import time
|
||||
import sys
|
||||
|
||||
def make_repo_auth(repo, userdata):
|
||||
""" write config to repo
|
||||
|
||||
user information in userdata is used for auth
|
||||
userdata has user names as keys, and a tuple (password, readwrite) as
|
||||
values, where 'readwrite' is either 'r' or 'rw'
|
||||
"""
|
||||
confdir = py.path.local(repo).join('conf')
|
||||
confdir.join('svnserve.conf').write('''\
|
||||
[general]
|
||||
anon-access = none
|
||||
password-db = passwd
|
||||
authz-db = authz
|
||||
realm = TestRepo
|
||||
''')
|
||||
authzdata = '[/]\n'
|
||||
passwddata = '[users]\n'
|
||||
for user in userdata:
|
||||
authzdata += '%s = %s\n' % (user, userdata[user][1])
|
||||
passwddata += '%s = %s\n' % (user, userdata[user][0])
|
||||
confdir.join('authz').write(authzdata)
|
||||
confdir.join('passwd').write(passwddata)
|
||||
|
||||
def serve_bg(repopath):
|
||||
pidfile = py.path.local(repopath).join('pid')
|
||||
port = 10000
|
||||
e = None
|
||||
while port < 10010:
|
||||
cmd = 'svnserve -d -T --listen-port=%d --pid-file=%s -r %s' % (
|
||||
port, pidfile, repopath)
|
||||
print(cmd)
|
||||
try:
|
||||
py.process.cmdexec(cmd)
|
||||
except py.process.cmdexec.Error:
|
||||
e = sys.exc_info()[1]
|
||||
else:
|
||||
# XXX we assume here that the pid file gets written somewhere, I
|
||||
# guess this should be relatively safe... (I hope, at least?)
|
||||
counter = pid = 0
|
||||
while counter < 10:
|
||||
counter += 1
|
||||
try:
|
||||
pid = pidfile.read()
|
||||
except py.error.ENOENT:
|
||||
pass
|
||||
if pid:
|
||||
break
|
||||
time.sleep(0.2)
|
||||
return port, int(pid)
|
||||
port += 1
|
||||
raise IOError('could not start svnserve: %s' % (e,))
|
||||
|
||||
class TestSvnAuth(object):
|
||||
def test_basic(self):
|
||||
auth = SvnAuth('foo', 'bar')
|
||||
assert auth.username == 'foo'
|
||||
assert auth.password == 'bar'
|
||||
assert str(auth)
|
||||
|
||||
def test_makecmdoptions_uname_pw_makestr(self):
|
||||
auth = SvnAuth('foo', 'bar')
|
||||
assert auth.makecmdoptions() == '--username="foo" --password="bar"'
|
||||
|
||||
def test_makecmdoptions_quote_escape(self):
|
||||
auth = SvnAuth('fo"o', '"ba\'r"')
|
||||
assert auth.makecmdoptions() == '--username="fo\\"o" --password="\\"ba\'r\\""'
|
||||
|
||||
def test_makecmdoptions_no_cache_auth(self):
|
||||
auth = SvnAuth('foo', 'bar', cache_auth=False)
|
||||
assert auth.makecmdoptions() == ('--username="foo" --password="bar" '
|
||||
'--no-auth-cache')
|
||||
|
||||
def test_makecmdoptions_no_interactive(self):
|
||||
auth = SvnAuth('foo', 'bar', interactive=False)
|
||||
assert auth.makecmdoptions() == ('--username="foo" --password="bar" '
|
||||
'--non-interactive')
|
||||
|
||||
def test_makecmdoptions_no_interactive_no_cache_auth(self):
|
||||
auth = SvnAuth('foo', 'bar', cache_auth=False,
|
||||
interactive=False)
|
||||
assert auth.makecmdoptions() == ('--username="foo" --password="bar" '
|
||||
'--no-auth-cache --non-interactive')
|
||||
|
||||
class svnwc_no_svn(py.path.svnwc):
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.commands = []
|
||||
super(svnwc_no_svn, self).__init__(*args, **kwargs)
|
||||
|
||||
def _svn(self, *args):
|
||||
self.commands.append(args)
|
||||
|
||||
class TestSvnWCAuth(object):
|
||||
def setup_method(self, meth):
|
||||
self.auth = SvnAuth('user', 'pass', cache_auth=False)
|
||||
|
||||
def test_checkout(self):
|
||||
wc = svnwc_no_svn('foo', auth=self.auth)
|
||||
wc.checkout('url')
|
||||
assert wc.commands[0][-1] == ('--username="user" --password="pass" '
|
||||
'--no-auth-cache')
|
||||
|
||||
def test_commit(self):
|
||||
wc = svnwc_no_svn('foo', auth=self.auth)
|
||||
wc.commit('msg')
|
||||
assert wc.commands[0][-1] == ('--username="user" --password="pass" '
|
||||
'--no-auth-cache')
|
||||
|
||||
def test_checkout_no_cache_auth(self):
|
||||
wc = svnwc_no_svn('foo', auth=self.auth)
|
||||
wc.checkout('url')
|
||||
assert wc.commands[0][-1] == ('--username="user" --password="pass" '
|
||||
'--no-auth-cache')
|
||||
|
||||
def test_checkout_auth_from_constructor(self):
|
||||
wc = svnwc_no_svn('foo', auth=self.auth)
|
||||
wc.checkout('url')
|
||||
assert wc.commands[0][-1] == ('--username="user" --password="pass" '
|
||||
'--no-auth-cache')
|
||||
|
||||
class svnurl_no_svn(py.path.svnurl):
|
||||
cmdexec_output = 'test'
|
||||
popen_output = 'test'
|
||||
def __init__(self, *args, **kwargs):
|
||||
py.path.svnurl.__init__(self, *args, **kwargs)
|
||||
self.commands = []
|
||||
|
||||
def _cmdexec(self, cmd):
|
||||
self.commands.append(cmd)
|
||||
return self.cmdexec_output
|
||||
|
||||
def _popen(self, cmd):
|
||||
self.commands.append(cmd)
|
||||
return self.popen_output
|
||||
|
||||
class TestSvnURLAuth(object):
|
||||
def setup_method(self, meth):
|
||||
self.auth = SvnAuth('foo', 'bar')
|
||||
|
||||
def test_init(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn')
|
||||
assert u.auth is None
|
||||
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
assert u.auth is self.auth
|
||||
|
||||
def test_new(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/foo', auth=self.auth)
|
||||
new = u.new(basename='bar')
|
||||
assert new.auth is self.auth
|
||||
assert new.url == 'http://foo.bar/svn/bar'
|
||||
|
||||
def test_join(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
new = u.join('foo')
|
||||
assert new.auth is self.auth
|
||||
assert new.url == 'http://foo.bar/svn/foo'
|
||||
|
||||
def test_listdir(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
u.cmdexec_output = '''\
|
||||
1717 johnny 1529 Nov 04 14:32 LICENSE.txt
|
||||
1716 johnny 5352 Nov 04 14:28 README.txt
|
||||
'''
|
||||
paths = u.listdir()
|
||||
assert paths[0].auth is self.auth
|
||||
assert paths[1].auth is self.auth
|
||||
assert paths[0].basename == 'LICENSE.txt'
|
||||
|
||||
def test_info(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/LICENSE.txt', auth=self.auth)
|
||||
def dirpath(self):
|
||||
return self
|
||||
u.cmdexec_output = '''\
|
||||
1717 johnny 1529 Nov 04 14:32 LICENSE.txt
|
||||
1716 johnny 5352 Nov 04 14:28 README.txt
|
||||
'''
|
||||
org_dp = u.__class__.dirpath
|
||||
u.__class__.dirpath = dirpath
|
||||
try:
|
||||
info = u.info()
|
||||
finally:
|
||||
u.dirpath = org_dp
|
||||
assert info.size == 1529
|
||||
|
||||
def test_open(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
foo = u.join('foo')
|
||||
foo.check = lambda *args, **kwargs: True
|
||||
ret = foo.open()
|
||||
assert ret == 'test'
|
||||
assert '--username="foo" --password="bar"' in foo.commands[0]
|
||||
|
||||
def test_dirpath(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/foo', auth=self.auth)
|
||||
parent = u.dirpath()
|
||||
assert parent.auth is self.auth
|
||||
|
||||
def test_mkdir(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/qweqwe', auth=self.auth)
|
||||
assert not u.commands
|
||||
u.mkdir(msg='created dir foo')
|
||||
assert u.commands
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
def test_copy(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
u2 = svnurl_no_svn('http://foo.bar/svn2')
|
||||
u.copy(u2, 'copied dir')
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
def test_rename(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/foo', auth=self.auth)
|
||||
u.rename('http://foo.bar/svn/bar', 'moved foo to bar')
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
def test_remove(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/foo', auth=self.auth)
|
||||
u.remove(msg='removing foo')
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
def test_export(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
target = py.path.local('/foo')
|
||||
u.export(target)
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
def test_log(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn/foo', auth=self.auth)
|
||||
u.popen_output = py.io.TextIO('''\
|
||||
<?xml version="1.0"?>
|
||||
<log>
|
||||
<logentry revision="51381">
|
||||
<author>guido</author>
|
||||
<date>2008-02-11T12:12:18.476481Z</date>
|
||||
<msg>Creating branch to work on auth support for py.path.svn*.
|
||||
</msg>
|
||||
</logentry>
|
||||
</log>
|
||||
''')
|
||||
u.check = lambda *args, **kwargs: True
|
||||
ret = u.log(10, 20, verbose=True)
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
assert len(ret) == 1
|
||||
assert int(ret[0].rev) == 51381
|
||||
assert ret[0].author == 'guido'
|
||||
|
||||
def test_propget(self):
|
||||
u = svnurl_no_svn('http://foo.bar/svn', auth=self.auth)
|
||||
u.propget('foo')
|
||||
assert '--username="foo" --password="bar"' in u.commands[0]
|
||||
|
||||
class pytest_funcarg__setup:
|
||||
def __init__(self, request):
|
||||
if not request.config.option.runslowtests:
|
||||
py.test.skip('use --runslowtests to run these tests')
|
||||
|
||||
tmpdir = request.getfuncargvalue("tmpdir")
|
||||
repodir = tmpdir.join("repo")
|
||||
py.process.cmdexec('svnadmin create %s' % repodir)
|
||||
if sys.platform == 'win32':
|
||||
repodir = '/' + str(repodir).replace('\\', '/')
|
||||
self.repo = py.path.svnurl("file://%s" % repodir)
|
||||
if py.std.sys.platform == 'win32':
|
||||
# remove trailing slash...
|
||||
repodir = repodir[1:]
|
||||
self.repopath = py.path.local(repodir)
|
||||
self.temppath = tmpdir.mkdir("temppath")
|
||||
self.auth = SvnAuth('johnny', 'foo', cache_auth=False,
|
||||
interactive=False)
|
||||
make_repo_auth(self.repopath, {'johnny': ('foo', 'rw')})
|
||||
self.port, self.pid = serve_bg(self.repopath.dirpath())
|
||||
# XXX caching is too global
|
||||
py.path.svnurl._lsnorevcache._dict.clear()
|
||||
request.addfinalizer(lambda: py.process.kill(self.pid))
|
||||
|
||||
class TestSvnWCAuthFunctional:
|
||||
def test_checkout_constructor_arg(self, setup):
|
||||
wc = py.path.svnwc(setup.temppath, auth=setup.auth)
|
||||
wc.checkout(
|
||||
'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename))
|
||||
assert wc.join('.svn').check()
|
||||
|
||||
def test_checkout_function_arg(self, setup):
|
||||
wc = py.path.svnwc(setup.temppath, auth=setup.auth)
|
||||
wc.checkout(
|
||||
'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename))
|
||||
assert wc.join('.svn').check()
|
||||
|
||||
def test_checkout_failing_non_interactive(self, setup):
|
||||
auth = SvnAuth('johnny', 'bar', cache_auth=False,
|
||||
interactive=False)
|
||||
wc = py.path.svnwc(setup.temppath, auth)
|
||||
py.test.raises(Exception,
|
||||
("wc.checkout('svn://localhost:%(port)s/%(repopath)s')" %
|
||||
setup.__dict__))
|
||||
|
||||
def test_log(self, setup):
|
||||
wc = py.path.svnwc(setup.temppath, setup.auth)
|
||||
wc.checkout(
|
||||
'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename))
|
||||
foo = wc.ensure('foo.txt')
|
||||
wc.commit('added foo.txt')
|
||||
log = foo.log()
|
||||
assert len(log) == 1
|
||||
assert log[0].msg == 'added foo.txt'
|
||||
|
||||
def test_switch(self, setup):
|
||||
wc = py.path.svnwc(setup.temppath, auth=setup.auth)
|
||||
svnurl = 'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename)
|
||||
wc.checkout(svnurl)
|
||||
wc.ensure('foo', dir=True).ensure('foo.txt').write('foo')
|
||||
wc.commit('added foo dir with foo.txt file')
|
||||
wc.ensure('bar', dir=True)
|
||||
wc.commit('added bar dir')
|
||||
bar = wc.join('bar')
|
||||
bar.switch(svnurl + '/foo')
|
||||
assert bar.join('foo.txt')
|
||||
|
||||
def test_update(self, setup):
|
||||
wc1 = py.path.svnwc(setup.temppath.ensure('wc1', dir=True),
|
||||
auth=setup.auth)
|
||||
wc2 = py.path.svnwc(setup.temppath.ensure('wc2', dir=True),
|
||||
auth=setup.auth)
|
||||
wc1.checkout(
|
||||
'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename))
|
||||
wc2.checkout(
|
||||
'svn://localhost:%s/%s' % (setup.port, setup.repopath.basename))
|
||||
wc1.ensure('foo', dir=True)
|
||||
wc1.commit('added foo dir')
|
||||
wc2.update()
|
||||
assert wc2.join('foo').check()
|
||||
|
||||
auth = SvnAuth('unknown', 'unknown', interactive=False)
|
||||
wc2.auth = auth
|
||||
py.test.raises(Exception, 'wc2.update()')
|
||||
|
||||
def test_lock_unlock_status(self, setup):
|
||||
port = setup.port
|
||||
wc = py.path.svnwc(setup.temppath, auth=setup.auth)
|
||||
wc.checkout(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename,))
|
||||
wc.ensure('foo', file=True)
|
||||
wc.commit('added foo file')
|
||||
foo = wc.join('foo')
|
||||
foo.lock()
|
||||
status = foo.status()
|
||||
assert status.locked
|
||||
foo.unlock()
|
||||
status = foo.status()
|
||||
assert not status.locked
|
||||
|
||||
auth = SvnAuth('unknown', 'unknown', interactive=False)
|
||||
foo.auth = auth
|
||||
py.test.raises(Exception, 'foo.lock()')
|
||||
py.test.raises(Exception, 'foo.unlock()')
|
||||
|
||||
def test_diff(self, setup):
|
||||
port = setup.port
|
||||
wc = py.path.svnwc(setup.temppath, auth=setup.auth)
|
||||
wc.checkout(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename,))
|
||||
wc.ensure('foo', file=True)
|
||||
wc.commit('added foo file')
|
||||
wc.update()
|
||||
rev = int(wc.status().rev)
|
||||
foo = wc.join('foo')
|
||||
foo.write('bar')
|
||||
diff = foo.diff()
|
||||
assert '\n+bar\n' in diff
|
||||
foo.commit('added some content')
|
||||
diff = foo.diff()
|
||||
assert not diff
|
||||
diff = foo.diff(rev=rev)
|
||||
assert '\n+bar\n' in diff
|
||||
|
||||
auth = SvnAuth('unknown', 'unknown', interactive=False)
|
||||
foo.auth = auth
|
||||
py.test.raises(Exception, 'foo.diff(rev=rev)')
|
||||
|
||||
class TestSvnURLAuthFunctional:
|
||||
def test_listdir(self, setup):
|
||||
port = setup.port
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=setup.auth)
|
||||
u.ensure('foo')
|
||||
paths = u.listdir()
|
||||
assert len(paths) == 1
|
||||
assert paths[0].auth is setup.auth
|
||||
|
||||
auth = SvnAuth('foo', 'bar', interactive=False)
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=auth)
|
||||
py.test.raises(Exception, 'u.listdir()')
|
||||
|
||||
def test_copy(self, setup):
|
||||
port = setup.port
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=setup.auth)
|
||||
foo = u.mkdir('foo')
|
||||
assert foo.check()
|
||||
bar = u.join('bar')
|
||||
foo.copy(bar)
|
||||
assert bar.check()
|
||||
assert bar.auth is setup.auth
|
||||
|
||||
auth = SvnAuth('foo', 'bar', interactive=False)
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=auth)
|
||||
foo = u.join('foo')
|
||||
bar = u.join('bar')
|
||||
py.test.raises(Exception, 'foo.copy(bar)')
|
||||
|
||||
def test_write_read(self, setup):
|
||||
port = setup.port
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=setup.auth)
|
||||
foo = u.ensure('foo')
|
||||
fp = foo.open()
|
||||
try:
|
||||
data = fp.read()
|
||||
finally:
|
||||
fp.close()
|
||||
assert data == ''
|
||||
|
||||
auth = SvnAuth('foo', 'bar', interactive=False)
|
||||
u = py.path.svnurl(
|
||||
'svn://localhost:%s/%s' % (port, setup.repopath.basename),
|
||||
auth=auth)
|
||||
foo = u.join('foo')
|
||||
py.test.raises(Exception, 'foo.open()')
|
||||
|
||||
# XXX rinse, repeat... :|
|
||||
Reference in New Issue
Block a user