* refactor some setup/teardown/ensuretemp usages to use funcargs
* introduce monkeypatch.syspath_prepend for safe monkey patching of module import path * fix monkeypatch naming --HG-- branch : trunk
This commit is contained in:
@@ -689,8 +689,7 @@ class TestSshGateway(BasicRemoteExecution):
|
||||
cls.sshhost = getspecssh().ssh
|
||||
cls.gw = py.execnet.SshGateway(cls.sshhost)
|
||||
|
||||
def test_sshconfig_functional(self):
|
||||
tmpdir = py.test.ensuretemp("test_sshconfig")
|
||||
def test_sshconfig_functional(self, tmpdir):
|
||||
ssh_config = tmpdir.join("ssh_config")
|
||||
ssh_config.write(
|
||||
"Host alias123\n"
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
"""
|
||||
|
||||
import py
|
||||
import os
|
||||
from py.__.execnet.gwmanage import GatewayManager, HostRSync
|
||||
|
||||
class TestGatewayManagerPopen:
|
||||
@@ -75,7 +76,6 @@ class TestGatewayManagerPopen:
|
||||
call = rec.popcall("pyexecnet_gwmanage_rsyncfinish")
|
||||
|
||||
def test_multi_chdir_popen_with_path(self, testdir):
|
||||
import os
|
||||
hm = GatewayManager(["popen//chdir=hello"] * 2)
|
||||
testdir.tmpdir.chdir()
|
||||
hellopath = testdir.tmpdir.mkdir("hello").realpath()
|
||||
@@ -117,7 +117,7 @@ class TestGatewayManagerPopen:
|
||||
|
||||
class pytest_funcarg__mysetup:
|
||||
def __init__(self, request):
|
||||
tmp = request.config.mktemp(request.function.__name__, numbered=True)
|
||||
tmp = request.getfuncargvalue('tmpdir')
|
||||
self.source = tmp.mkdir("source")
|
||||
self.dest = tmp.mkdir("dest")
|
||||
request.getfuncargvalue("_pytest") # to have patching of py._com.comregistry
|
||||
|
||||
@@ -2,38 +2,37 @@ import py
|
||||
from py.execnet import RSync
|
||||
|
||||
|
||||
def setup_module(mod):
|
||||
mod.gw = py.execnet.PopenGateway()
|
||||
mod.gw2 = py.execnet.PopenGateway()
|
||||
def pytest_funcarg__gw1(request):
|
||||
return request.cached_setup(
|
||||
setup=py.execnet.PopenGateway,
|
||||
teardown=lambda val: val.exit(),
|
||||
scope="module"
|
||||
)
|
||||
pytest_funcarg__gw2 = pytest_funcarg__gw1
|
||||
|
||||
def teardown_module(mod):
|
||||
mod.gw.exit()
|
||||
mod.gw2.exit()
|
||||
def pytest_funcarg__dirs(request):
|
||||
t = request.getfuncargvalue('tmpdir')
|
||||
class dirs:
|
||||
source = t.join("source")
|
||||
dest1 = t.join("dest1")
|
||||
dest2 = t.join("dest2")
|
||||
return dirs
|
||||
|
||||
|
||||
class DirSetup:
|
||||
def setup_method(self, method):
|
||||
name = "%s.%s" %(self.__class__.__name__, method.__name__)
|
||||
self.tmpdir = t = py.test.ensuretemp(name)
|
||||
self.source = t.join("source")
|
||||
self.dest1 = t.join("dest1")
|
||||
self.dest2 = t.join("dest2")
|
||||
|
||||
class TestRSync(DirSetup):
|
||||
def test_notargets(self):
|
||||
rsync = RSync(self.source)
|
||||
class TestRSync:
|
||||
def test_notargets(self, dirs):
|
||||
rsync = RSync(dirs.source)
|
||||
py.test.raises(IOError, "rsync.send()")
|
||||
assert rsync.send(raises=False) is None
|
||||
|
||||
def test_dirsync(self):
|
||||
dest = self.dest1
|
||||
dest2 = self.dest2
|
||||
source = self.source
|
||||
def test_dirsync(self, dirs, gw1, gw2):
|
||||
dest = dirs.dest1
|
||||
dest2 = dirs.dest2
|
||||
source = dirs.source
|
||||
|
||||
for s in ('content1', 'content2', 'content2-a-bit-longer'):
|
||||
source.ensure('subdir', 'file1').write(s)
|
||||
rsync = RSync(self.source)
|
||||
rsync.add_target(gw, dest)
|
||||
rsync = RSync(dirs.source)
|
||||
rsync.add_target(gw1, dest)
|
||||
rsync.add_target(gw2, dest2)
|
||||
rsync.send()
|
||||
assert dest.join('subdir').check(dir=1)
|
||||
@@ -49,76 +48,70 @@ class TestRSync(DirSetup):
|
||||
source.join('subdir').remove('file1')
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw2, dest2)
|
||||
rsync.add_target(gw, dest)
|
||||
rsync.add_target(gw1, dest)
|
||||
rsync.send()
|
||||
assert dest.join('subdir', 'file1').check(file=1)
|
||||
assert dest2.join('subdir', 'file1').check(file=1)
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw, dest, delete=True)
|
||||
rsync.add_target(gw1, dest, delete=True)
|
||||
rsync.add_target(gw2, dest2)
|
||||
rsync.send()
|
||||
assert not dest.join('subdir', 'file1').check()
|
||||
assert dest2.join('subdir', 'file1').check()
|
||||
|
||||
def test_dirsync_twice(self):
|
||||
source = self.source
|
||||
def test_dirsync_twice(self, dirs, gw1, gw2):
|
||||
source = dirs.source
|
||||
source.ensure("hello")
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw, self.dest1)
|
||||
rsync.add_target(gw1, dirs.dest1)
|
||||
rsync.send()
|
||||
assert self.dest1.join('hello').check()
|
||||
assert dirs.dest1.join('hello').check()
|
||||
py.test.raises(IOError, "rsync.send()")
|
||||
assert rsync.send(raises=False) is None
|
||||
rsync.add_target(gw, self.dest2)
|
||||
rsync.add_target(gw1, dirs.dest2)
|
||||
rsync.send()
|
||||
assert self.dest2.join('hello').check()
|
||||
assert dirs.dest2.join('hello').check()
|
||||
py.test.raises(IOError, "rsync.send()")
|
||||
assert rsync.send(raises=False) is None
|
||||
|
||||
def test_rsync_default_reporting(self):
|
||||
source = self.source
|
||||
def test_rsync_default_reporting(self, capsys, dirs, gw1):
|
||||
source = dirs.source
|
||||
source.ensure("hello")
|
||||
cap = py.io.StdCapture()
|
||||
try:
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw, self.dest1)
|
||||
rsync.send()
|
||||
finally:
|
||||
out, err = cap.reset()
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw1, dirs.dest1)
|
||||
rsync.send()
|
||||
out, err = capsys.readouterr()
|
||||
assert out.find("hello") != -1
|
||||
|
||||
def test_rsync_non_verbose(self):
|
||||
source = self.source
|
||||
def test_rsync_non_verbose(self, capsys, dirs, gw1):
|
||||
source = dirs.source
|
||||
source.ensure("hello")
|
||||
cap = py.io.StdCapture()
|
||||
try:
|
||||
rsync = RSync(source, verbose=False)
|
||||
rsync.add_target(gw, self.dest1)
|
||||
rsync.send()
|
||||
finally:
|
||||
out, err = cap.reset()
|
||||
rsync = RSync(source, verbose=False)
|
||||
rsync.add_target(gw1, dirs.dest1)
|
||||
rsync.send()
|
||||
out, err = capsys.readouterr()
|
||||
assert not out
|
||||
assert not err
|
||||
|
||||
def test_symlink_rsync(self):
|
||||
def test_symlink_rsync(self, dirs, gw1):
|
||||
if py.std.sys.platform == 'win32':
|
||||
py.test.skip("symlinks are unsupported on Windows.")
|
||||
source = self.source
|
||||
dest = self.dest1
|
||||
self.source.ensure("existant")
|
||||
source = dirs.source
|
||||
dest = dirs.dest1
|
||||
dirs.source.ensure("existant")
|
||||
source.join("rellink").mksymlinkto(source.join("existant"), absolute=0)
|
||||
source.join('abslink').mksymlinkto(source.join("existant"))
|
||||
|
||||
rsync = RSync(source)
|
||||
rsync.add_target(gw, dest)
|
||||
rsync.add_target(gw1, dest)
|
||||
rsync.send()
|
||||
|
||||
assert dest.join('rellink').readlink() == dest.join("existant")
|
||||
assert dest.join('abslink').readlink() == dest.join("existant")
|
||||
|
||||
def test_callback(self):
|
||||
dest = self.dest1
|
||||
source = self.source
|
||||
def test_callback(self, dirs, gw1):
|
||||
dest = dirs.dest1
|
||||
source = dirs.source
|
||||
source.ensure("existant").write("a" * 100)
|
||||
source.ensure("existant2").write("a" * 10)
|
||||
total = {}
|
||||
@@ -127,14 +120,14 @@ class TestRSync(DirSetup):
|
||||
|
||||
rsync = RSync(source, callback=callback)
|
||||
#rsync = RSync()
|
||||
rsync.add_target(gw, dest)
|
||||
rsync.add_target(gw1, dest)
|
||||
rsync.send()
|
||||
|
||||
assert total == {("list", 110):True, ("ack", 100):True, ("ack", 10):True}
|
||||
|
||||
def test_file_disappearing(self):
|
||||
dest = self.dest1
|
||||
source = self.source
|
||||
def test_file_disappearing(self, dirs, gw1):
|
||||
dest = dirs.dest1
|
||||
source = dirs.source
|
||||
source.ensure("ex").write("a" * 100)
|
||||
source.ensure("ex2").write("a" * 100)
|
||||
|
||||
@@ -147,7 +140,7 @@ class TestRSync(DirSetup):
|
||||
return True
|
||||
|
||||
rsync = DRsync(source)
|
||||
rsync.add_target(gw, dest)
|
||||
rsync.add_target(gw1, dest)
|
||||
rsync.send()
|
||||
assert rsync.x == 1
|
||||
assert len(dest.listdir()) == 1
|
||||
|
||||
Reference in New Issue
Block a user